PyTorch Ignite 0.4.8 : Tutorials : センテンス分類のための畳込みニューラルネット (翻訳/解説)

翻訳 : (株)クラスキャット セールスインフォメーション
作成日時 : 03/29/2022 (0.4.8)

* 本ページは、Pytorch Ignite の以下のドキュメントを翻訳した上で適宜、補足説明したものです：

• Tutorials : Convolutional Neural Networks for Sentence Classification using Ignite
  

* サンプルコードの動作確認はしておりますが、必要な場合には適宜、追加改変しています。
* ご自由にリンクを張って頂いてかまいませんが、sales-info@classcat.com までご一報いただけると嬉しいです。

 

 

Tutorials : センテンス分類のための畳込みニューラルネット

これは、Ignite を使用して、ニューラルネットワーク・モデルを訓練し、実験をセットアップしてモデルを検証するチュートリアルです。

この実験では、センテンス分類のための畳込みニューラルネットワーク by Yoon Kim を再現していきます。この論文はテキスト分類に対して CNN を使用します、これは典型的には RNN, ロジスティック回帰, ナイーブベイズに委ねられるタスクです。

IMDB 映画レビューを分類してレビューがポジティブかネガティブが予測できることを望みます。IMDB 映画レビューデータセットは 25000 のポジティブと 25000 のネガティブサンプルから成ります。データセットはテキストとラベルのペアから構成されます。これは二値分類問題です。モデルを作成するために PyTorch を、データをインポートするために torchtext をそしてモデルを訓練して監視するために Ignite を使用していきます。

Lets get started!

 

Required Dependencies

このサンプルでは torch と ignite は既にインストールされているとか停止、torchtext と spacy パッケージを必要とするだけです。

!pip install pytorch-ignite torchtext==0.9.1 spacy
!python -m spacy download en_core_web_sm

 

ライブラリのインポート

import random

torchtext は tochrivion と同様に、NLP タスクのための複数のデータセットを提供します。下で以下をインポートします :

• datasets : NLP データセットをダウンロードするモジュール。
• GloVe : 事前訓練済みの GloVe 埋め込みをダウンロードして使用するモジュール。

from torchtext import datasets
from torchtext.vocab import GloVe

モデルを作成するために torch, nn と functional モジュールをインポートします！

import torch
import torch.nn as nn
import torch.nn.functional as F

 
Ignite は PyTorch でニューラルネットワークを訓練するのに役立つ高位ライブラリです。それは訓練ループ, 様々なメトリクス, ハンドラと有用な contrib セクションをセットアップするためのエンジンを装備しています！

下で、以下をインポートします :

• Engine : データセットの各バッチに対して与えられた process_function を実行し、それと共にイベントを発生させます。
• Events: 特定のイベントで関数を起動するようにエンジンに関数を装着することをユーザに可能にします。Eg: EPOCH_COMPLETED, ITERATION_STARTED, etc.
• Accuracy : 二値, マルチクラス, マルチラベル 等のためのデータセットに対して精度を計算するメトリック。
• Loss : パラメータとして損失関数を取る一般的なメトリックで、データセットに対する損失を計算します。
• RunningAverage : 訓練の間にエンジンに装着する一般的なメトリック
• ModelCheckpoint : モデルをチェックポイントするためのハンドラ。
• EarlyStopping : スコア関数に基づいて訓練を停止するハンドラ。
• ProgressBar : tqdm 進捗バーを作成するハンドラ。

from ignite.engine import Engine, Events
from ignite.metrics import Accuracy, Loss, RunningAverage
from ignite.handlers import ModelCheckpoint, EarlyStopping
from ignite.contrib.handlers import ProgressBar
from ignite.utils import manual_seed

SEED = 1234
manual_seed(SEED)

device = torch.device('cuda' if torch.cuda.is_available() else 'cpu')

 

データ処理

最初に torchtext.data.utils を使用してトークナイザーをセットアップします。トークナイザーのジョブはセンテンスを「トークンに」分解することです。その詳細は wikipedia で読むことができます。”spacy” ライブラリからのトークナイザーを利用します、これはポピュラーな選択です。デフォルトのものや望む別のものを使用することを望む場合には、”basic_english” に自由に切り替えてください。

docs: https://pytorch.org/text/stable/data_utils.html

from torchtext.data.utils import get_tokenizer
tokenizer = get_tokenizer("spacy")

tokenizer("Ignite is a high-level library for training and evaluating neural networks.")

次に、IMDB 訓練とテストデータセットがダウンロードされます。torchtext.datasets API は前処理情報なしに訓練/テストデータセット分割を直接返します。各分割は行毎に raw テキストとラベルを生成するイテレータです。

train_iter, test_iter = datasets.IMDB(split=('train','test'))

次に訓練、検証とテスト分割をセットアップします。

# We are using only 1000 samples for faster training
# set to -1 to use full data
N = 1000 

# We will use 80% of the `train split` for training and the rest for validation
train_frac = 0.8
_temp = list(train_iter)


random.shuffle(_temp)
_temp = _temp[:(N if N > 0 else len(_temp) )]
n_train = int(len(_temp)*train_frac)

train_list = _temp[:n_train]
validation_list = _temp[n_train:]
test_list = list(test_iter)
test_list = test_list[:(N if N > 0 else len(test_list))]

データサンプルがどのようなものか、調べましょう。各データサンプルは形式 (label, text) のタプルです。

ラベルの値は ‘pos’ か ‘neg’ であり得ます。

random_sample = random.sample(train_list,1)[0]
print(' text:', random_sample[1])
print('label:', random_sample[0])

 
データセット分割を得た今、語彙を構築しましょう。このため、torchtext.vocab からの Vocab クラスを使用します。訓練データセットに基づいて語彙を構築することは重要です、検証とテスト (データ) は検証作業の間未見 (= unseen) であるためです。

Vocab は事前訓練済みの GloVE 100 次元単語ベクトルを使用することを可能にします。これは、各単語は 100 floats により記述されることを意味します！より詳細を読みたい場合には、ここに幾つかのリソースがあります。

• StanfordNLP – GloVe
• DeepLearning.ai Lecture
• Stanford CS224N Lecture by Richard Socher

GloVE ダウンロードサイズは約 900MB ですので、ダウンロードするのにある程度時間がかかるかもしれないことに注意してください。

Vocab クラスのインスタンスは以下の属性を持ちます :

• extend は語彙を増やすために使用されます。
• freqs は各単語の頻度の辞書です。
• itos は語彙の総ての単語のリストです。
• stoi は総ての単語をインデックスにマップする辞書です。
• vectors はダウンロードされた埋め込みの torch.Tensor です。

from collections import Counter
from torchtext.vocab import Vocab

counter = Counter()

for (label, line) in train_list:
    counter.update(tokenizer(line))

vocab = Vocab(
    counter,
    min_freq=10,
    vectors=GloVe(name='6B', dim=100, cache='/tmp/glove/')
)

print("The length of the new vocab is", len(vocab))
new_stoi = vocab.stoi
print("The index of '' is", new_stoi[''])
new_itos = vocab.itos
print("The token at index 2 is", new_itos[2])

そしてデータセット・イテレータ (or リストのような iterable) からの raw テキストとラベルデータを処理するために、 ここでは lambda func として text_transform と label_transform を作成します、これらは callable オブジェクトです。text_transform のセンテンスに <BOS> と <EOS> のような特殊記号を追加できます。

text_transform = lambda x: [vocab[token] for token in tokenizer(x)]
label_transform = lambda x: 1 if x == 'pos' else 0

# Print out the output of text_transform
print("input to the text_transform:", "here is an example")
print("output of the text_transform:", text_transform("here is an example"))

データバッチの生成のために torch.utils.data.DataLoader を使用します。DataLoader の collate_fn 引数で関数を定義することによってデータバッチをカスタマイズできます。ここでは、collate_batch func で、raw テキストデータを処理してバッチの最長センテンスに動的に適合するようにパディングを追加します。

from torch.utils.data import DataLoader
from torch.nn.utils.rnn import pad_sequence

def collate_batch(batch):
    label_list, text_list = [], []
    for (_label, _text) in batch:
        label_list.append(label_transform(_label))
        processed_text = torch.tensor(text_transform(_text))
        text_list.append(processed_text)
    return torch.tensor(label_list), pad_sequence(text_list, padding_value=3.0)

batch_size = 8  # A batch size of 8

def create_iterators(batch_size=8):
    """Heler function to create the iterators"""
    dataloaders = []
    for split in [train_list, validation_list, test_list]:
        dataloader = DataLoader(
            split, batch_size=batch_size,
            collate_fn=collate_batch
            )
        dataloaders.append(dataloader)
    return dataloaders

train_iterator, valid_iterator, test_iterator = create_iterators()

next(iter(train_iterator))

イテレータの出力が何かを実際に調べましょう、これによりモデルの入力が何か、ラベルを出力と比較する方法、そして Ignite の Engine のために process_functions を設定する方法を知ります。

• batch[0][0] は単一サンプルのラベルです。vocab.stoi が元はテキストのラベルを float にマップするために使用されたことが分かります。
  

• batch[1][0] は単一サンプルのテキストです。ラベルと同様に、vocab.stoi がサンプルのテキストの各トークンをインデックスに変換するために使用されたことが分かります。

train_iterator の最初の 10 バッチのセンテンスの長さをプリントしましょう。ここで総てのバッチは異なる長さであることが分かります、これはイテレータが期待どおりに動作していることを意味します。

batch = next(iter(train_iterator))
print('batch[0][0] : ', batch[0][0])
print('batch[1][0] : ', batch[1][[0] != 1])

lengths = []
for i, batch in enumerate(train_iterator):
    x = batch[1]
    lengths.append(x.shape[0])
    if i == 10:
        break

print ('Lengths of first 10 batches : ', lengths)

 

TextCNN モデル

ここにモデルのレプリカがあり、ここにモデルの演算があります :

• Embedding : shape (N, L) のテキストのバッチを (N, L, D) に埋め込みます、ここで N はバッチサイズ、L はバッチの最大長、D は埋め込み次元です。
  

• Convolutions : 埋め込み単語に対して trigrams, four-grams, five-grams を模倣する (= mimic) ために 3, 4, 5 のカーネルサイズを持つ並列畳込みを実行します。これは畳込み毎に (N, L – k + 1, D) の出力という結果になります、ここで k は kernel_size です。
  

• Activation : ReLu 活性が各畳込み演算に適用されます。
  

• Pooling : L – k + 1 のウィンドウサイズを持つ活性化された畳込みに対して並列 maxpooling 演算を実行し、チャネル毎に 1 つの値という結果になります i.e. プーリング毎に (N, 1, D) の shape です。
  

• Concat : プーリング出力は連結されて squeeze されて結果は (N, 3D) の shape になります。これはセンテンスのための単一埋め込みです。
  

• Dropout : Dropout は埋め込みセンテンスに適用されます。
  

• Fully Connected : dropout 出力は shape (3D, 1) の完全結合層に渡されてバッチの各サンプルのために単一出力を与えます。この層の出力に sigmoid が適用されます。
  

• load_embeddings : ユーザ入力に基づいて埋め込みをロードする、TextCNN のために定義されたメソッドです。3 つのモードがあります – rand, これはランダムに初期化された重み、static, これは凍結された事前訓練済みの重み、そして nonstatic, これは訓練可能な事前訓練済み重みです。

このモデルは可変なテキスト長に対して動作することに注意してください！アイデアはセンテンスの単語を埋め込み、センテンスを単一ベクトルとして埋め込むために畳込み, maxpooling と連結を使用するものです。この単一ベクトルは単一値を出力するために sigmoid を持つ完全結合層に渡されます。この値はセンテンスがポジティブ (1 に近い) かネガティブ (0 に近い) である確率として解釈できます。

モデルにより想定されるテキストの最小の長さはモデルの最小カーネルサイズのサイズです。

lass TextCNN(nn.Module):
    def __init__(
        self,
        vocab_size,
        embedding_dim, 
        kernel_sizes, 
        num_filters, 
        num_classes, d_prob, mode):
        super(TextCNN, self).__init__()
        self.vocab_size = vocab_size
        self.embedding_dim = embedding_dim
        self.kernel_sizes = kernel_sizes
        self.num_filters = num_filters
        self.num_classes = num_classes
        self.d_prob = d_prob
        self.mode = mode
        self.embedding = nn.Embedding(
            vocab_size, embedding_dim, padding_idx=0)
        self.load_embeddings()
        self.conv = nn.ModuleList([nn.Conv1d(in_channels=embedding_dim,
                                             out_channels=num_filters,
                                             kernel_size=k, stride=1) for k in kernel_sizes])
        self.dropout = nn.Dropout(d_prob)
        self.fc = nn.Linear(len(kernel_sizes) * num_filters, num_classes)

    def forward(self, x):
        batch_size, sequence_length = x.shape
        x = self.embedding(x.T).transpose(1, 2)
        x = [F.relu(conv(x)) for conv in self.conv]
        x = [F.max_pool1d(c, c.size(-1)).squeeze(dim=-1) for c in x]
        x = torch.cat(x, dim=1)
        x = self.fc(self.dropout(x))
        return torch.sigmoid(x).squeeze()

    def load_embeddings(self):
        if 'static' in self.mode:
            self.embedding.weight.data.copy_(vocab.vectors)
            if 'non' not in self.mode:
                self.embedding.weight.data.requires_grad = False
                print('Loaded pretrained embeddings, weights are not trainable.')
            else:
                self.embedding.weight.data.requires_grad = True
                print('Loaded pretrained embeddings, weights are trainable.')
        elif self.mode == 'rand':
            print('Randomly initialized embeddings are used.')
        else:
            raise ValueError('Unexpected value of mode. Please choose from static, nonstatic, rand.')

 

モデル, optimizer と損失の作成

以下で TextCNN モデルのインスタンスを作成して static モードで埋め込みをロードします。モデルはデバイスに置かれ、そして Binary Cross Entropy の損失関数と Adam optimizer がセットアップされます。

vocab_size, embedding_dim = vocab.vectors.shape

model = TextCNN(vocab_size=vocab_size,
                embedding_dim=embedding_dim,
                kernel_sizes=[3, 4, 5],
                num_filters=100,
                num_classes=1, 
                d_prob=0.5,
                mode='static')
model.to(device)
optimizer = torch.optim.Adam(model.parameters(), lr=1e-4, weight_decay=1e-3)
criterion = nn.BCELoss()

 

Ignite を使用した訓練と評価

トレーナーエンジン – process_function

Ignite の Engine はユーザが与えられたバッチを処理する process_function を定義することを可能にします、これはデータセットの総てのバッチに適用されます。これはモデルを訓練して検証するために適用できる一般的なクラスです！process_function は 2 つのパラメータ engine と batch を持っています。

トレーナーの機能が何を行なうかをウォークスルーしましょう :

• モデルを train モードに設定します。
  

• optimizer の勾配をゼロに設定する。
  

• バッチから x と y を生成する。
  

• モデルと x を使用して y_pred を計算するために forward パスを実行する。
  

• y_pred と y を使用して損失を計算する。
  

• モデルパラメータの勾配を計算するために損失を使用して backward パスを実行する。
  

• モデルパラメータは勾配と optimizer を使用して最適化される。
  

• スカラー損失を返す。

以下は訓練プロセスの間の単一操作です。この process_function は訓練エンジンに装着されます。

def process_function(engine, batch):
    model.train()
    optimizer.zero_grad()
    y, x = batch
    x = x.to(device)
    y = y.to(device)
    y_pred = model(x)
    loss = criterion(y_pred, y.float())
    loss.backward()
    optimizer.step()
    return loss.item()

 

評価エンジン – process_function

訓練プロセス関数と同様に、単一バッチを評価する関数をセットアップします。ここに eval_function が行なうことがあります :

• モデルを eval モードに設定する。
  

• バッチから x と y を生成する。
  

• torch.no_grad() により、任意の続くステップに対して勾配は計算されません。
  

• モデルと x に基づいて y_pred を計算するためにモデルの forward パスを実行する。
  

• y_pred と y を返す。

Ignite はメトリクスをトレーナーではなく evaluator に装着することを勧めています、というのは訓練の間、モデルパラメータは常に変化していてモデルは安定しているモデルで評価することが最善であるためです。この情報は重要です、訓練と評価のための関数に違いがあるからです。訓練は単一のスカラー損失を返します。評価は y_pred と y を返し、その出力はデータセット全体のバッチ毎のメトリクスを計算するために使用されます。

Ignite の総てのメトリクスはエンジンに装着された関数の出力として y_pred と y を必要とします。

def eval_function(engine, batch):
    model.eval()
    with torch.no_grad():
        y, x = batch
        y = y.to(device)
        x = x.to(device)
        y = y.float()
        y_pred = model(x)
        return y_pred, y

 

訓練と評価エンジンのインスタンス化

以下で 3 つのエンジン – trainer, 訓練 evaluator と検証 evaluator を作成します。train_evaluator と validation_evaluator は同じ関数を使用することに気付くでしょう、何故これが行なわれたかを後で見ます！

trainer = Engine(process_function)
train_evaluator = Engine(eval_function)
validation_evaluator = Engine(eval_function)

 

メトリクス – RunningAverage, Accuracy と Loss

最初に、各バッチのスカラー損失出力の移動平均を追跡するために Running Average のメトリックを装着します。

RunningAverage(output_transform=lambda x: x).attach(trainer, 'loss')

そして評価のために使用したい 2 つのメトリクス – accuracy と loss があります。これは二値問題ですので、損失については loss_function として Binary Cross Entropy 関数を単純に渡すことができます。

精度については、Ignite は y_pred と y が 0 と 1 だけから構成されることを必要とします。モデルは sigmoid 層から出力しますので、値は 0 と 1 の間です。y_pred と y から成る engine.state.output を変換する関数を書く必要があります。

下では thresholded_output_transform がそれを単に行ない、それは y_pred を 0 と 1 に変換するために丸めてから、丸めた y_pred と y を返します。この関数は Accuracy の望まれる目的を達成するために engine.state.output を変換するために使用される output_transform 関数です。

今は、Loss と Accuracy (with thresholded_output_transform) を train_evaluator と validation_evaluator に装着します。

エンジンにメトリックを装着するには、次の形式が使用されます :

• Metric(output_transform=output_transform, …).attach(engine, ‘metric_name’)

def thresholded_output_transform(output):
    y_pred, y = output
    y_pred = torch.round(y_pred)
    return y_pred, y

Accuracy(output_transform=thresholded_output_transform).attach(train_evaluator, 'accuracy')
Loss(criterion).attach(train_evaluator, 'bce')

Accuracy(output_transform=thresholded_output_transform).attach(validation_evaluator, 'accuracy')
Loss(criterion).attach(validation_evaluator, 'bce')

 

プログレスバー

次に Ignite のプログレスバーのインスタンスを作成し、それをトレーナーに装着して、追跡する engine.state.metrics のキーを渡します。この場合、プログレスバーは engine.state.metrics[‘loss’] を追跡していきます。

pbar = ProgressBar(persist=True, bar_format="")
pbar.attach(trainer, ['loss'])

 

EarlyStopping – 検証損失の追跡

次にこの訓練プロセスに対して Early Stopping ハンドラをセットアップします。EarlyStopping は訓練を停止する基準が何であれユーザが定義することを可能にする score_function を必要とします。この場合、検証セットの損失が 5 エポック内に減少しない場合、訓練プロセスは早期に停止します。

def score_function(engine):
    val_loss = engine.state.metrics['bce']
    return -val_loss

handler = EarlyStopping(patience=5, score_function=score_function, trainer=trainer)
validation_evaluator.add_event_handler(Events.COMPLETED, handler)

 

カスタム関数を特定のイベントでエンジンに装着

以下では、独自のカスタム関数を定義してそれらを訓練プロセスの様々な Event に装着する方法を見ます。

下の関数は両者とも同様のタスクを実現し、それらはデータセットで動作する evaluator の結果をプリントします。一つの関数は訓練用 evaluator とデータセット上でそれを行ない、他方は検証用で行ないます。もう一つの違いはこれらの関数が trainer エンジンで装着される方法です。

最初の方法はデコレータを使用し、そのシンタクスは単純です – @ trainer.on(Events.EPOCH_COMPLETED)、これは修飾された関数がトレーナーに装着されて各エポックで呼び出されることを意味します。

2 つ目の方法はトレーナーの add_event_handler を使用します – trainer.add_event_handler(Events.EPOCH_COMPLETED, custom_function)。これは上と同じ結果を得ます。

@trainer.on(Events.EPOCH_COMPLETED)
def log_training_results(engine):
    train_evaluator.run(train_iterator)
    metrics = train_evaluator.state.metrics
    avg_accuracy = metrics['accuracy']
    avg_bce = metrics['bce']
    pbar.log_message(
        "Training Results - Epoch: {}  Avg accuracy: {:.2f} Avg loss: {:.2f}"
        .format(engine.state.epoch, avg_accuracy, avg_bce))
    
def log_validation_results(engine):
    validation_evaluator.run(valid_iterator)
    metrics = validation_evaluator.state.metrics
    avg_accuracy = metrics['accuracy']
    avg_bce = metrics['bce']
    pbar.log_message(
        "Validation Results - Epoch: {}  Avg accuracy: {:.2f} Avg loss: {:.2f}"
        .format(engine.state.epoch, avg_accuracy, avg_bce))
    pbar.n = pbar.last_print_n = 0

trainer.add_event_handler(Events.EPOCH_COMPLETED, log_validation_results)

 

モデルチェックポイント

最後に、このモデルをチェックポイントすることを望みます。それを行なうことは重要です、訓練プロセスは時間がかかる可能性があり、訓練の間に何かの理由で問題が発生した場合、失敗したポイントから訓練を再開するためにモデルチェックポイントは有用であり得るからです。

下では各エポックの最後にモデルをチェックポイントするために Ignite の ModelCheckpoint ハンドラを使用しています。

checkpointer = ModelCheckpoint('/tmp/models', 'textcnn', n_saved=2, create_dir=True, save_as_state_dict=True)
trainer.add_event_handler(Events.EPOCH_COMPLETED, checkpointer, {'textcnn': model})

 

エンジンの実行

次に、20 エポックの間トレーナーを実行して結果をモニタします。下でプログレスバーがイテレーション毎の損失をプリントし、カスタム関数で指定したように訓練と検証の結果をプリントすることを見ます。

trainer.run(train_iterator, max_epochs=20)

That’s it! We have successfully trained and evaluated a Convolutational Neural Network for Text Classification.

以上