PyTorch : AllenNLP チュートリアル : Getting Started – Welcome (翻訳)

翻訳 : (株)クラスキャット セールスインフォメーション
作成日時 : 09/26/2018 (v0.6.1)

* 本ページは、github 上の allenai/allennlp の Tutorials : Getting Started – Welcome を動作確認・翻訳した上で適宜、補足説明したものです：

• https://github.com/allenai/allennlp/tree/v0.6.1/tutorials

* ご自由にリンクを張って頂いてかまいませんが、sales-info@classcat.com までご一報いただけると嬉しいです。

 

Welcome to AllenNLP ! このチュートリアルは AllenNLP モデルの構築と訓練の基礎を貴方にウォークスルーして頂きます。

このチュートリアルでは 品詞タギングのための LSTM PyTorch チュートリアルの僅かに拡張したバージョンを幾つかの特徴を追加して実装します、それらの特徴はそれを僅かにより現実的なタスクにします (そしてまた AllenNLP の恩恵の幾つかのショーケースです) :

1. 私達のデータをファイルから読みます (チュートリアルのサンプルは Python コードの一部として与えられるデータを使用します)。
2. パフォーマンスをチェックするために別の検証データセットを使用します (チュートリアルは同じデータセット上で訓練して評価します)。
3. 訓練の進捗を追跡するために tqdm を使用します。
4. 検証データセット上の損失に基づいて early stopping を実装します。
5. モデルを訓練するとき訓練と検証セットの両者上で精度を追跡します。

 

課題

センテンス (e.g. “The dog ate the apple”) が与えられたとき各単語に対して品詞タグを予測することを望みます (e.g [“DET”, “NN”, “V”, “DET”, “NN”])。

PyTorch チュートリアル内のように、各単語を低次元空間に埋め込み、エンコーディングのシークエンスを得るためにそれらを LSTM を通して渡し、そしてそれらを (可能な品詞タグに対応する) ロジットのシークエンスに変換するために feedforward 層を使用します。

◆ 以下はこれを成すための注釈付きのコードです。

AllenNLP ではあらゆることのために型アノテーションを使用します :

from typing import Iterator, List, Dict

AllenNLP は PyTorch の上に構築されていますので、そのコードを自由に使用できます :

import torch
import torch.optim as optim
import numpy as np

AllenNLP では各訓練サンプルを様々なタイプの Field を含む Instance として表します。ここで各サンプルはセンテンスを含む TextField、そして対応する品詞タグを含む SequenceLabelField を持ちます :

from allennlp.data import Instance
from allennlp.data.fields import TextField, SequenceLabelField

典型的には AllenNLP を使用してこのような問題を解くためには、2 つのクラスを実装しなければなりません。最初は DatasetReader で、これはデータのファイルを読み、そして Instance のストリームを生成するためのロジックを含みます :

from allennlp.data.dataset_readers import DatasetReader

頻繁に URL からデータセットやモデルをロードすることを望みます。cached_path ヘルパーはそのようなファイルをダウンロードして、それらをローカルにキャッシュして、そしてローカル・パスを返します。それはまたローカル・ファイルパスも受け取ります (それは単に as-is で返します) :

from allennlp.common.file_utils import cached_path

単語を一つまたはそれ以上のインデックスとして表わすために様々な方法があります。例えば、一意の単語の語彙を保持して各単語に対応する id を与えることができるかもしれません。あるいは単語の文字毎に一つの id を持って各単語を id のシークエンスとして表わせるかもしれません。AllenNLP はこの表現のための TokenIndexer 抽象を使用します :

from allennlp.data.token_indexers import TokenIndexer, SingleIdTokenIndexer
from allennlp.data.tokenizers import Token

TokenIndexer がトークンをどのようにインデックスに変えるかのためのルールを表わす一方で、Vocabulary は文字列から整数への対応するマッピングを含みます。例えば、トークン indexer がトークンを文字 id のシークエンスとして表わすことを指定するかもしれません、その場合には Vocabulary はマッピング {character -> id} を含むでしょう。この特定のサンプルではSingleIdTokenIndexer を使用します、これは各トークンを一意の id に割り当て、その結果 Vocabulary はマッピング {token -> id} (reverse mapping とともに) を含みます :

from allennlp.data.vocabulary import Vocabulary

DatasetReader 以外に、典型的には貴方が実装する必要がある他のクラスは Model です、これは tensor 入力を取り (最適化することを望む訓練損失を含む) tensor 出力の dict を生成する PyTorch Module です :

from allennlp.models import Model

上述したように、私達のモデルは埋め込み層、続いて LSTM 層、そして feedforward 層から成ります。AllenNLP は、パディングとバッチ処理に加えて様々なユティリティ関数をスマートに扱う、これら総てのための抽象を含みます :

from allennlp.modules.text_field_embedders import TextFieldEmbedder, BasicTextFieldEmbedder
from allennlp.modules.token_embedders import Embedding
from allennlp.modules.seq2seq_encoders import Seq2SeqEncoder, PytorchSeq2SeqWrapper
from allennlp.nn.util import get_text_field_mask, sequence_cross_entropy_with_logits

訓練と検証データセット上で精度を追跡することを望みます :

from allennlp.training.metrics import CategoricalAccuracy

訓練においてデータを知的にバッチ処理できる DataIterators を必要とします :

from allennlp.data.iterators import BucketIterator

そしてフル機能の AllenNLP の Trainer を使用します :

from allennlp.training.trainer import Trainer

最終的に、新しい入力上で予測を行なうことを望みます、これについては下に詳しいです :

from allennlp.predictors import SentenceTaggerPredictor

torch.manual_seed(1)

最初にやらなければならないことは DatasetReader サブクラスを実装します :

class PosDatasetReader(DatasetReader):
    """
    DatasetReader for PoS tagging data, one sentence per line, like

        The###DET dog###NN ate###V the###DET apple###NN
    """

DatasetReader が必要とする唯一のパラメータはトークンをどのようにインデックスに変換するかを指定する TokenIndexers の dict です。デフォルトでは各トークンに対して (それぞれの個別のトークンのために一意の id である) 単一のインデックスだけを生成します (これは NLP タスクで使用する標準的な “word to index” マッピングです) :

    def __init__(self, token_indexers: Dict[str, TokenIndexer] = None) -> None:
        super().__init__(lazy=False)
        self.token_indexers = token_indexers or {"tokens": SingleIdTokenIndexer()}

DatasetReader.text_to_instance は訓練サンプル (この場合はセンテンスのトークンと対応する品詞タグ) に対応する入力を取り、対応する Fields (この場合はセンテンスのための TextField とそのタグのための SequenceLabelField) をインスタンス化し、そしてそれらのフィールドを含む Instance を返します。タグはオプションであることに注意してください、何故ならばラベル付けされていないデータから (それらの上で予測するために) インスタンスを作成することも可能にしたいからです :

    def text_to_instance(self, tokens: List[Token], tags: List[str] = None) -> Instance:
        sentence_field = TextField(tokens, self.token_indexers)
        fields = {"sentence": sentence_field}

        if tags:
            label_field = SequenceLabelField(labels=tags, sequence_field=sentence_field)
            fields["labels"] = label_field

        return Instance(fields)

実装しなければならない他のピースは _read で、これはファイル名を取り Instances のストリームを生成します。ワークの殆どは既に text_to_instance で成されています :

    def _read(self, file_path: str) -> Iterator[Instance]:
        with open(file_path) as f:
            for line in f:
                pairs = line.strip().split()
                sentence, tags = zip(*(pair.split("###") for pair in pairs))
                yield self.text_to_instance([Token(word) for word in sentence], tags)

 
◆ 基本的に常に実装しなければならない他のクラスは Model で、これは torch.nn.Module のサブクラスです。それがどのように動作するかは大部分貴方次第で、殆どそれは tensor 入力を取り (モデルを訓練するために使用する損失を含む) tensor 出力の dict を生成する forward メソッドを必要とするだけです。上述したように、モデルは埋め込み層、シークエンス・エンコーダ、そして feedforward から成ります :

class LstmTagger(Model):

通常ではないように見えるかもしれない一つのことは embedder とシークエンス・エンコーダをコンストラクタのパラメータとして渡していくことです。これは、モデルコードを変更しなければならないことなしに、異なる embedder とエンコーダで実験することを可能にします :

    def __init__(self,
                 word_embeddings: TextFieldEmbedder,
                 encoder: Seq2SeqEncoder,
                 vocab: Vocabulary) -> None:

embedding 層は AllenNLP TextFieldEmbedder として指定され、これはトークンを tensor に変える一般的な方法を表します (ここで各一意の単語を学習された tensor で表わすことを望むことを知っていますが、一般的なクラスの使用は embedding の異なるタイプ、例えば ELMo で簡単に実験することを可能にします)。

同様に、エンコーダは一般的な Seq2SeqEncoder として指定されます、LSTM を使用することを望むことを知ってさえいますが。再度、これは他のシークエンス・エンコーダ、例えば Transformer で実験することを容易にします。

総ての AllenNLP モデルはまた Vocabulary を想定します、これはトークンのインデックスとラベルのインデックスへの名前空間のマッピングを含みます。

 
vocab を基底クラス・コンストラクタに渡さなければならないことに気づいてください :

        super().__init__(vocab)
        self.word_embeddings = word_embeddings
        self.encoder = encoder

feed forward 層はパラメータとしては渡されませんが、我々によって構築されます。それは正しい入力次元を見つけるためにエンコーダを見てそして正しい出力次元を見つけるために vocabulary (そして特に、label -> index マッピング) を見ることに注意してください。

        self.hidden2tag = torch.nn.Linear(in_features=encoder.get_output_dim(),
                                          out_features=vocab.get_vocab_size('labels'))

注意すべき最後のことは CategoricalAccuracy メトリックもインスタンス化することです、これは各訓練と検証エポックの間に精度を追跡するために使用します :

        self.accuracy = CategoricalAccuracy()

次に forward を実装する必要があります、これは実際の計算が発生する場所です。データセットの各インスタンスは (他のインスタンスとともにバッチ化されて) foward に供給されます。forward メソッドは入力として tensor の dict を想定し、そしてそれはそれらの名前がインスタンスのフィールドの名前であることを想定します。この場合 sentence フィールドと (多分) labels フィールドを持ちますので、従って foward を次のように構築します :

    def forward(self,
                sentence: Dict[str, torch.Tensor],
                labels: torch.Tensor = None) -> torch.Tensor:

AllenNLP はバッチ化された入力上で動作するように設計されていますが、異なる入力シークエンスは異なる長さを持ちます。裏では AllenNLP はバッチが均一の shape を持つようにより短い入力をパディングしています、これはパディングを排除するためには計算がマスクを使用することを必要とすることを意味します。ここでは単にユティリティ関数 get_text_field_mask を使用します、これはパッドとアンパッドされた位置に対応する 0s と 1s の tensor を返します :

        mask = get_text_field_mask(sentence)

sentence tensor (各 sentence はトークン id のシークエンス) を word_embeddings モジュールに渡すことで開始します、これは各センテンスを embedded tensor のシークエンスに変換します :

        embeddings = self.word_embeddings(sentence)

次に embedded tensor (そして mask) を LSTM に渡します、これはエンコードされた出力のシークエンスを生成します :

        encoder_out = self.encoder(embeddings, mask)

最後に、様々なタグに対応するロジットを生成するために各エンコードされた出力 tensor を feedforward 層に渡します :

        tag_logits = self.hidden2tag(encoder_out)
        output = {"tag_logits": tag_logits}

前のように、labels はオプションです、何故ならばラベル付けされていないデータ上で予測するためにこのモデルを実行することを望むからです。labels を持つ場合には、精度メトリックを更新して “loss” を計算するためにそれらを使用します :

        if labels is not None:
            self.accuracy(tag_logits, labels, mask)
            output["loss"] = sequence_cross_entropy_with_logits(tag_logits, labels, mask)

        return output

各 forward パスで更新される精度メトリックを含めました。それは get_metrics メソッドを override する必要があることを意味します。裏では、CategoricalAccuracy メトリックが予測の数と正しい予測の数をストアして、forward への呼び出しの間にそれらのカウントを更新しています。get_metric への各呼び出しは計算された精度を返して (オプションで) カウントをリセットします、これは各エポックのために新たに精度を追跡することを可能にするものです :

    def get_metrics(self, reset: bool = False) -> Dict[str, float]:
        return {"accuracy": self.accuracy.get_metric(reset)}

 
◆ DatasetReader と Model を実装した今、訓練する準備ができました。最初に dataset reader のインスタンスが必要です :

reader = PosDatasetReader()

これは訓練データと検証データを読むために使用できます。ここでは URL からそれらを読みますが、貴方のデータがローカルにあればローカルファイルから読むこともできるでしょう。ファイルをローカルにキャッシュするために cached_path を使用します (そして reader.read にローカル cached バージョンへのパスを渡します) :

train_dataset = reader.read(cached_path(
    'https://raw.githubusercontent.com/allenai/allennlp'
    '/master/tutorials/tagger/training.txt'))
validation_dataset = reader.read(cached_path(
    'https://raw.githubusercontent.com/allenai/allennlp'
    '/master/tutorials/tagger/validation.txt'))

ひとたびデータセットを読み取れば、Vocabulary を作成するためにそれらを使用します (つまり、トークン/ラベルから id へのマッピング) :

vocab = Vocabulary.from_instances(train_dataset + validation_dataset)

今ではモデルを構築するする必要があります。embedding 層と LSTM の隠れ層のためのサイズを選択します :

EMBEDDING_DIM = 6
HIDDEN_DIM = 6

トークンを埋め込むために単に BasicTextFieldEmbedder を使用します、これはインデックス名から embedding へのマッピングを取ります。DatasetReader を定義した場所へ戻れば、デフォルト・パラメータは “tokens” と呼ばれる単一のインデックスを含みますので、私達のマッピングはそのインデックスに対応する embedding を単に必要とします。幾つの embedding を必要とするかを見つけるために Vocabulary を使用して 出力次元を指定するために EMBEDDING_DIM パラメータを使用します。事前訓練された embedding (例えば、GloVe ベクトル) で始めることも可能ですが、この tiny toy データセット上ではそれを行なう必要はありません :

token_embedding = Embedding(num_embeddings=vocab.get_vocab_size('tokens'),
                            embedding_dim=EMBEDDING_DIM)
word_embeddings = BasicTextFieldEmbedder({"tokens": token_embedding})

次にシークエンス・エンコーダを指定する必要があります。PytorchSeq2SeqWrapper のための必要性はここでは少し unfortunate ですが (そして configuration ファイルを使用する場合それについて心配する必要はありません)、ここでは幾つかの特別な機能 (そして cleaner I/F) を組み込み PyTorch モジュールに追加することが必要とされます。AllenNLP では総てを batch first で行ないますので、それもまた指定します :

lstm = PytorchSeq2SeqWrapper(torch.nn.LSTM(EMBEDDING_DIM, HIDDEN_DIM, batch_first=True))

最後に、モデルをインスタン化することができます :

model = LstmTagger(word_embeddings, lstm, vocab)

今ではモデルを訓練する準備ができました。PyTorch の stochastic gradient descent を単に使用できます :

optimizer = optim.SGD(model.parameters(), lr=0.1)

そしてデータセットのためのバッチ処理を扱う DataIterator を必要とします。BucketIterator は同様なシークエンス長でバッチを作成するためにフィールドを指定することによってインスタンスをソートします。ここでインスタンスを sentence フィールドのトークンの数でソートすることを望むことを示します :

iterator = BucketIterator(batch_size=2, sorting_keys=[("sentence", "num_tokens")])

iterator がそのインスタンスが私達の vocabulary を使用してインデックスされることを確実にするべきであるとまた指定します ; つまり、それらの文字列が前に作成したマッピングを使用して整数に変換されたことをです :

iterator.index_with(vocab)

さて Trainer をインスタンス化してそれを実行します。ここで 1000 エポックの間実行することと検証メトリックが改善することなしに 10 エポックを費やす場合には早期に訓練を停止することをそれに伝えます。デフォルト検証メトリックは loss ですが、異なるメトリックと方向 (e.g. accuracy はより大きくなるべきです) を指定することも可能です :

trainer = Trainer(model=model,
                  optimizer=optimizer,
                  iterator=iterator,
                  train_dataset=train_dataset,
                  validation_dataset=validation_dataset,
                  patience=10,
                  num_epochs=1000)

それを launch したときそれは各エポックについて進捗バーをプリントします、それは “loss” と “accuracy” メトリックの両者を含みます。モデルが良ければ、訓練するにつれて loss は下がり accuracy は上がります :

trainer.train()

元の PyTorch チュートリアル内のように、モデルが生成する予測を見ることを望みます。AllenNLP は Predictor 抽象を含みます、これは入力を取り、それらをインスタンスに変換し、それらをモデルを通して供給し、そして JSON-serializable 結果を返します。しばしば貴方自身の Predictor を実装する必要がありますが、AllenNLP は既にここでは完全に動作する SentenceTaggerPredictor を持ちますので、それを使用できます。それはモデル (予測を行なうため) とデータセット reader (インスタンスを作成するため) を要求します :

predictor = SentenceTaggerPredictor(model, dataset_reader=reader)

それは predict メソッドを持ち、これは単に sentence を必要として forward からの出力 dict (の JSON-serializable バージョン) を返します。ここでは tag_logits はロジットの (5, 3) 配列で、5 単語の各々に対する 3 つの可能なタグに対応します :

tag_logits = predictor.predict("The dog ate the apple")['tag_logits']

実際の “predictions” を得るために単に argmax を取ることができます :

tag_ids = np.argmax(tag_logits, axis=-1)

そしてそれから予測されたタグを見つけるために vocabulary を使用します :

print([model.vocab.get_token_from_index(i, 'labels') for i in tag_ids])

 

Appendix

from typing import Iterator, List, Dict

import torch
import torch.optim as optim
import numpy as np

from allennlp.data import Instance
from allennlp.data.fields import TextField, SequenceLabelField

from allennlp.data.dataset_readers import DatasetReader

from allennlp.common.file_utils import cached_path

from allennlp.data.token_indexers import TokenIndexer, SingleIdTokenIndexer
from allennlp.data.tokenizers import Token

from allennlp.data.vocabulary import Vocabulary

from allennlp.models import Model

from allennlp.modules.text_field_embedders import TextFieldEmbedder, BasicTextFieldEmbedder
from allennlp.modules.token_embedders import Embedding
from allennlp.modules.seq2seq_encoders import Seq2SeqEncoder, PytorchSeq2SeqWrapper
from allennlp.nn.util import get_text_field_mask, sequence_cross_entropy_with_logits

from allennlp.training.metrics import CategoricalAccuracy

from allennlp.data.iterators import BucketIterator

from allennlp.training.trainer import Trainer

from allennlp.predictors import SentenceTaggerPredictor

torch.manual_seed(1)

class PosDatasetReader(DatasetReader):
    """
    DatasetReader for PoS tagging data, one sentence per line, like

        The###DET dog###NN ate###V the###DET apple###NN
    """

    def __init__(self, token_indexers: Dict[str, TokenIndexer] = None) -> None:
        super().__init__(lazy=False)
        self.token_indexers = token_indexers or {"tokens": SingleIdTokenIndexer()}

    def text_to_instance(self, tokens: List[Token], tags: List[str] = None) -> Instance:
        sentence_field = TextField(tokens, self.token_indexers)
        fields = {"sentence": sentence_field}

        if tags:
            label_field = SequenceLabelField(labels=tags, sequence_field=sentence_field)
            fields["labels"] = label_field

        return Instance(fields)

    def _read(self, file_path: str) -> Iterator[Instance]:
        with open(file_path) as f:
            for line in f:
                pairs = line.strip().split()
                sentence, tags = zip(*(pair.split("###") for pair in pairs))
                yield self.text_to_instance([Token(word) for word in sentence], tags)

class LstmTagger(Model):

    def __init__(self,
                 word_embeddings: TextFieldEmbedder,
                 encoder: Seq2SeqEncoder,
                 vocab: Vocabulary) -> None:

        super().__init__(vocab)
        self.word_embeddings = word_embeddings
        self.encoder = encoder

        self.hidden2tag = torch.nn.Linear(in_features=encoder.get_output_dim(),
                                          out_features=vocab.get_vocab_size('labels'))

        self.accuracy = CategoricalAccuracy()

    def forward(self,
                sentence: Dict[str, torch.Tensor],
                labels: torch.Tensor = None) -> torch.Tensor:

        mask = get_text_field_mask(sentence)
 
        embeddings = self.word_embeddings(sentence)

        encoder_out = self.encoder(embeddings, mask)

        tag_logits = self.hidden2tag(encoder_out)
        output = {"tag_logits": tag_logits}

        if labels is not None:
            self.accuracy(tag_logits, labels, mask)
            output["loss"] = sequence_cross_entropy_with_logits(tag_logits, labels, mask)

        return output

    def get_metrics(self, reset: bool = False) -> Dict[str, float]:
        return {"accuracy": self.accuracy.get_metric(reset)}

reader = PosDatasetReader()

train_dataset = reader.read(cached_path(
    'https://raw.githubusercontent.com/allenai/allennlp'
    '/master/tutorials/tagger/training.txt'))
validation_dataset = reader.read(cached_path(
    'https://raw.githubusercontent.com/allenai/allennlp'
    '/master/tutorials/tagger/validation.txt'))

vocab = Vocabulary.from_instances(train_dataset + validation_dataset)

EMBEDDING_DIM = 6
HIDDEN_DIM = 6

token_embedding = Embedding(num_embeddings=vocab.get_vocab_size('tokens'),
                            embedding_dim=EMBEDDING_DIM)
word_embeddings = BasicTextFieldEmbedder({"tokens": token_embedding})

lstm = PytorchSeq2SeqWrapper(torch.nn.LSTM(EMBEDDING_DIM, HIDDEN_DIM, batch_first=True))

model = LstmTagger(word_embeddings, lstm, vocab)

optimizer = optim.SGD(model.parameters(), lr=0.1)

iterator = BucketIterator(batch_size=2, sorting_keys=[("sentence", "num_tokens")])

iterator.index_with(vocab)

trainer = Trainer(model=model,
                  optimizer=optimizer,
                  iterator=iterator,
                  train_dataset=train_dataset,
                  validation_dataset=validation_dataset,
                  patience=10,
                  num_epochs=1000)
trainer.train()

predictor = SentenceTaggerPredictor(model, dataset_reader=reader)

tag_logits = predictor.predict("The dog ate the apple")['tag_logits']

tag_ids = np.argmax(tag_logits, axis=-1)

print([model.vocab.get_token_from_index(i, 'labels') for i in tag_ids])

以上