PyTorch Ignite 0.4.8 : AI Tutorials : Beginner – テキスト分類のための Transformers (翻訳/解説)

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

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

• Tutorials : Beginner – Transformers for Text Classification with IMDb Reviews

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

 

 

PyTorch Ignite 0.4.8 : AI Tutorials : テキスト分類のための Transformers

このチュートリアルではテキスト分類のための Transformers ライブラリからのモデルを PyTorch-Ignite を使用して再調整します。テキストを前処理してモデル, optimizer とデータローダを定義するために Fine-tuning a pretrained model チュートリアルに従います。そして以下のために Ignite を使用していきます :

• モデルの訓練と評価
• メトリクスの計算
• 実験のセットアップとモデルのモニタリング

チュートリアルに従って、レビューをポジティブかネガティブに分類するために IMDb Movie Reviews データセット を使用します。

 

必要な依存関係

!pip install pytorch-ignite transformers datasets

詳細に進む前に、manual_seed を使用して総てにシードを設定します。

from ignite.utils import manual_seed

manual_seed(42)

 

基本的なセットアップ

次にチュートリアルに従ってデータセットと tokenizer をロードしてデータを前処理します。

 

データ前処理

from datasets import load_dataset

raw_datasets = load_dataset("imdb")

from transformers import AutoTokenizer

tokenizer = AutoTokenizer.from_pretrained("bert-base-cased")

def tokenize_function(examples):
    return tokenizer(examples["text"], padding="max_length", truncation=True)

tokenized_datasets = raw_datasets.map(tokenize_function, batched=True)

PyTorch 固有の手順についてはこのチュートリアルの最後の方にあります。ここでは元のデータセットから更に大きなサブセットを抽出しています。また最初に総てにシードを設定しましたのでシードを提供する必要はありません。

tokenized_datasets = tokenized_datasets.remove_columns(["text"])
tokenized_datasets = tokenized_datasets.rename_column("label", "labels")
tokenized_datasets.set_format("torch")

small_train_dataset = tokenized_datasets["train"].shuffle().select(range(5000))
small_eval_dataset = tokenized_datasets["test"].shuffle().select(range(5000))

 

データローダ

from torch.utils.data import DataLoader

train_dataloader = DataLoader(small_train_dataset, shuffle=True, batch_size=8)
eval_dataloader = DataLoader(small_eval_dataset, batch_size=8)

 

モデル

from transformers import AutoModelForSequenceClassification

model = AutoModelForSequenceClassification.from_pretrained("bert-base-cased", num_labels=2)

 

Optimizer

from transformers import AdamW

optimizer = AdamW(model.parameters(), lr=5e-5)

 

LR スケジューラ

線形スケジューラの組み込みの Ignite 代替を使用します、これは PiecewiseLinear です。エポック数もまた増やします。

from ignite.contrib.handlers import PiecewiseLinear

num_epochs = 10
num_training_steps = num_epochs * len(train_dataloader)

milestones_values = [
        (0, 5e-5),
        (num_training_steps, 0.0),
    ]
lr_scheduler = PiecewiseLinear(
        optimizer, param_name="lr", milestones_values=milestones_values
    )

 

デバイスの設定

import torch

device = torch.device("cuda") if torch.cuda.is_available() else torch.device("cpu")
model.to(device)

 

トレーナーの作成

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

チュートリアルの訓練データのバッチ処理のためのコードは次のようなものです :

for batch in train_dataloader:
    batch = {k: v.to(device) for k, v in batch.items()}
    outputs = model(**batch)
    loss = outputs.loss
    loss.backward()

    optimizer.step()
    lr_scheduler.step()
    optimizer.zero_grad()
    progress_bar.update(1)

そして上のタスクを遂行するために (下で train_step と呼称する) process_function を定義します :

• モデルを train モードに設定する。
• バッチの項目をデバイスに転送する。
• forward パスを実行して出力を生成する。
• 損失を抽出する。
• モデルパラメータのための勾配を計算するために損失を使用して backward パスを実行する。
• 勾配と optimizer を使用してモデルパラメータを最適化する。

最後に損失を返すことを選択しますので、更なる処理にそれを利用できます。

train_step で lr_scheduler と progress_bar の更新をしていないことに気付いたでしょう。これは、このチュートリアルで後で見るように Ignite が自動的に処理するからです。

def train_step(engine, batch):  
    model.train()
    
    batch = {k: v.to(device) for k, v in batch.items()}
    outputs = model(**batch)
    loss = outputs.loss
    loss.backward()

    optimizer.step()
    optimizer.zero_grad()

    return loss

それから train_step を訓練エンジンに装着することによりモデルトレーナーを作成します。後で num_epochs 数、訓練データセットに渡りループさせるためにトレーナーを使用します。

from ignite.engine import Engine

trainer = Engine(train_step)

前に定義した lr_scheduler はハンドラでした。

ハンドラ は任意のタイプの関数 (lambda 関数, クラスメソッド 等) であり得ます。その上、Ignite は冗長なコードを削減するために幾つかの組み込みハンドラを提供しています。これらのハンドラをエンジンに装着します、これは特定の イベント でトリガーされます。これらのイベントはイテレーションの開始やエポックの終了のようなものであり得ます。ここ に組み込みイベントの完全なリストがあります。

そして、add_event_handler() を通して lr_scheduler (ハンドラ) をトレーナー (エンジン) に装着します、その結果それは自動的に Events.ITERATION_STARTED (イテレーションの開始) でトリガーされます。

from ignite.engine import Events

trainer.add_event_handler(Events.ITERATION_STARTED, lr_scheduler)

これが train_step() に lr_scheduler.step() を含めなかった理由です。

 

プログレスバー

次に Ignite の ProgessBar() を作成して progress_bar.update(1) を置き換えるためにそれをトレーナーに装着します。

from ignite.contrib.handlers import ProgressBar

pbar = ProgressBar()

進捗を単純に追跡するか :

pbar.attach(trainer)

トレーナー (or train_step) の出力を追跡することもできます :

pbar.attach(trainer, output_transform=lambda x: {'loss': x})

 

Evaluator の作成

訓練 process_function と同様に、train/validation/test データの単一バッチを評価する関数をセットアップします :

model.eval()
for batch in eval_dataloader:
    batch = {k: v.to(device) for k, v in batch.items()}
    with torch.no_grad():
        outputs = model(**batch)

    logits = outputs.logits
    predictions = torch.argmax(logits, dim=-1)
    metric.add_batch(predictions=predictions, references=batch["labels"])

以下、ここに evaluate_step() が行なうことがあります :

• モデルを eval モードに設定する。
• バッチの項目をデバイスに転送する。
• torch.no_grad() で、後続ステップについて勾配は計算されません。
• バッチからの出力を計算するためにモデル上で forward パスを実行する。
• ロジット (ポジティブとネガティブクラスの確率) から実際の予測値を取得する。

最後に、メトリクスを計算できるように予測値と実際のラベルを返します。

evaluate_step() でメトリクスを計算しなかったことに気付くでしょう。これは、Ignite が組み込み メトリクス を提供するからです、これは後でエンジンに装着できます。

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

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

def evaluate_step(engine, batch):
    model.eval()

    batch = {k: v.to(device) for k, v in batch.items()}
    with torch.no_grad():
        outputs = model(**batch)

    logits = outputs.logits
    predictions = torch.argmax(logits, dim=-1)

    return {'y_pred': predictions, 'y': batch["labels"]}

下では 2 つのエンジン, 訓練 evaluator と検証 evaluator を作成します。train_evaluator と validation_evaluator は同じ関数を使用しますが、それらはこのチュートリアルで後で見るように異なる目的に役立ちます。

train_evaluator = Engine(evaluate_step)
validation_evaluator = Engine(evaluate_step)

 

メトリクスの装着

チュートリアルは評価のために使用される 1 つのメトリック, accuracy を定義しています :

metric= load_metric("accuracy")

train_evaluator と validation_evaluator に Ignite の組み込み Accuracy() を容易に装着できます。メトリック名 (下では accracy) もまた指定する必要があります。内部的には、それは精度を計算するために y_pred と y を使用します。

from ignite.metrics import Accuracy

Accuracy().attach(train_evaluator, 'accuracy')
Accuracy().attach(validation_evaluator, 'accuracy')

 

ログ・メトリクス

次にカスタムハンドラ (関数) を定義してそれらを訓練プロセスの様々な Event に装着します。

下の関数は両者とも類似のタスクを遂行します。それらはデータセットで実行された evaluator の結果をプリントします。log_training_results() はこれを訓練 evaluator と訓練データセット上で、一方で log_validation_results() は検証 evaluator と検証データセット上で行ないます。もう一つの違いはこれらの関数がトレーナー・エンジンに装着される方法です。

最初のメソッドはデコレータを使用しています、構文は単純です – @ trainer.on(Events.EPOCH_COMPLETED), これはデコレートされた関数は trainer に装着されて各エポックの終了時に呼び出されることを意味します。

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_dataloader)
    metrics = train_evaluator.state.metrics
    avg_accuracy = metrics['accuracy']
    print(f"Training Results - Epoch: {engine.state.epoch}  Avg accuracy: {avg_accuracy:.3f}")
    
def log_validation_results(engine):
    validation_evaluator.run(eval_dataloader)
    metrics = validation_evaluator.state.metrics
    avg_accuracy = metrics['accuracy']
    print(f"Validation Results - Epoch: {engine.state.epoch}  Avg accuracy: {avg_accuracy:.3f}")

trainer.add_event_handler(Events.EPOCH_COMPLETED, log_validation_results)

 

Early Stopping

次に訓練プロセスのために EarlyStopping ハンドラをセットアップします。EarlyStopping は score_function を必要とします、これはユーザが訓練を停止する基準をどのようなものでも定義することを可能にします。この場合には、検証セットの損失が 2 エポック (patience) 内に減少しなければ、訓練プロレスは早期に停止します。

from ignite.handlers import EarlyStopping

def score_function(engine):
    val_accuracy = engine.state.metrics['accuracy']
    return val_accuracy

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

 

モデル・チェックポイント

最後に、最良のモデル重みをセーブすることを望みます。そこで各エポックの終了時にモデルをチェックポイントする Ignite の ModelCheckpoint ハンドラを使用します。これは models ディレクトリを作成して、プレフィックス bert-base-cased を持つ 2 つの最良モデル (n_saved) をセーブします。

from ignite.handlers import ModelCheckpoint

checkpointer = ModelCheckpoint(dirname='models', filename_prefix='bert-base-cased', n_saved=2, create_dir=True)
trainer.add_event_handler(Events.EPOCH_COMPLETED, checkpointer, {'model': model})

 

Begin Training!

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

trainer.run(train_dataloader, max_epochs=num_epochs)

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

以上