はじめに
業務効率向上などを見越して、AllenNLPを勉強することになりました。そのため公式チュートリアルをトレースしていこうと思ったのですが、プログラムがそのまま動かなかったので、修正メモがてら記事にまとめます。また、Tutorialは3Stepに分かれているため、Stepごとに記事を分割しようと思います。
2021/07/25現在でチュートリアル内のコードをブラウザ上で実行できるようなのですが、それも動きませんでした。あらら。
AllenNLPは現時点の最新v2.6.0を使用します。今回使用したNoteBookです。
AllenNLPとは
AllenNLPとは何かについて、公式サイトの紹介文を以下に転記します。
AllenNLPは、Allen Institute for Artificial Intelligenceが開発した、自然言語処理用の深層学習モデルを構築するためのオープンソースライブラリです。PyTorchの上に構築されており、高品質なNLPモデルを簡単に構築したいと考えている研究者、エンジニア、学生などをサポートするために設計されています。最新のNLPに共通するコンポーネントやモデルのための高レベルの抽象化とAPIを提供しています。また，実験や実験結果の管理を容易にする拡張可能なフレームワークを提供しています。
実装済みのNLP共通の処理やモデルを利用でき、さらにそれを用いた実験の管理までサポートしてくれるフレームワークということですね。
AllenNLP基礎知識
以下ではAllenNLPの基礎知識を紹介します。
データ形式
AllenNLPでは各データはInstanseによって表現されており、以下のようなTextFieldやLabelFieldを持ちます。また、入力されたテキストはDatasetReaderを用いて読み込み、Instanceへの変換が行われます。
基本的な学習の流れ
学習は以下の図のような流れになります。学習中、Instanceのバッチを受け取り、Model.forward()に通し、lossに基づいて、勾配を計算・モデルパラメータの更新を行います。AllenNLPでは学習のループを実装する必要がありませんが、必要に応じて実装可能となっているようです。
IMDBデータセットによるネガポジ分類
ここからチュートリアルStep1のIMDBデータセットを用いたネガポジ分類モデルの作成を通してAllenNLPの実装を見ます。
実行環境
- Google Colaboratory
準備
AllenNLPチュートリアルが提供するIMDBデータをgithubからダウンロードします。以下のコマンドを実行し、データが格納されているquick_startをカレントディレクトリに配置します。パスが公式チュートリアルのプログラムに沿うようにしています。
!git clone https://github.com/allenai/allennlp-guide.git
!cp -r ./allennlp-guide/quick_start .
また、allennlpのパッケージをインストールします。
!pip install allennlp==2.6.0
プログラムの全体像
プログラムの全体的な流れは以下のようになります。1.ファイル読み込み器の作成,2.訓練・検証データ読み込み,3.語彙の定義,4.データローダーの作成,5.分類モデルの定義,6.訓練の順です。馴染み深い流れですね。
6.訓練のみよくわからないことをやっています。serialization_dirは学習過程の出力ファイルやモデルを保存するためのディレクトリになります。では、tempfile.TemporaryDirectory()は何かというと一時退避ディレクトリになります。プログラム実行時には生成されますが、実際にディレクトリが存在するわけではなく、プログラムが終了次第ディレクトリは削除されます。これは公式チュートリアルの環境で実行する際に必要らしいのですが、Colabでの実行の際には必要ありません。
次に以下プログラムの関数やクラス群が何を表しているかを簡単に見ていこうと思います。
def run_training_loop():
# 1. ファイル読み込み・Instance生成器の作成
dataset_reader = ClassificationTsvReader()
# 2. 訓練・検証データ読み込み
train_data, dev_data = read_data(dataset_reader)
# 3. 語彙の定義
vocab = build_vocab(train_data + dev_data)
# 4. データローダーの作成
train_loader, dev_loader = build_data_loaders(train_data, dev_data)
# データを語彙でインデックス付けする
train_loader.index_with(vocab)
dev_loader.index_with(vocab)
# 5. 分類モデルの定義
model = build_model(vocab)
# GPU or CPU選択
model = model.to(DEVICE)
# 6. 訓練
with tempfile.TemporaryDirectory() as serialization_dir:
trainer = build_trainer(model, serialization_dir, train_loader, dev_loader)
# trainer = build_trainer(model, "./output", train_loader, dev_loader)
trainer.train()
return model, dataset_reader
model, dataset_reader = run_training_loop()
vocab = model.vocab
1. ファイル読み込み・データInstance生成クラス
以下でファイル読み込み・データInstance生成クラスを定義します。今回このクラスでは空白を挟んで並んだ単語列を、空白を境に分割し(WhitespaceTokenizer)、各単語を一意のIDに読み替えます(SingleIdTokenIndexer)。またメンバ関数として、トークン列とラベルをまとめて一つのインスタンスに置き換えるtext_to_instance関数と、TSVファイルを読み込んで、TextとLabelをまとめて一つのインスタンスにする_read関数を持っています。
厳密にはDatasetReaderが持っているtext_to_instanceと_readをoverridesして定義しています。これらの定義は必須のようで、継承元のプログラムにはraise NotImplementedErrorの一文しか書かれていませんので、ご注意を。
class ClassificationTsvReader(DatasetReader):
def __init__(
self,
tokenizer: Tokenizer = None,
token_indexers: Dict[str, TokenIndexer] = None,
max_tokens: int = None,
**kwargs
):
super().__init__(**kwargs)
self.tokenizer = tokenizer or WhitespaceTokenizer()
self.token_indexers = token_indexers or {"tokens": SingleIdTokenIndexer()}
self.max_tokens = max_tokens
@overrides
def text_to_instance(self, tokens: List[Token], label: str = None) -> Instance:
"""トークン列とラベルをまとめて一つのインスタンスに置き換える
"""
if self.max_tokens:
tokens = tokens[: self.max_tokens]
text_field = TextField(tokens, self.token_indexers)
fields: Dict[str, Field] = {"text": text_field}
if label:
fields["label"] = LabelField(label)
return Instance(fields)
@overrides
def _read(self, file_path: str) -> Iterable[Instance]:
"""ファイルを読み込んで、`TextField`と`LabelField`をまとめて一つのインスタンスにする
"""
with open(file_path, "r") as lines:
for line in lines:
text, sentiment = line.strip().split("\t")
tokens = self.tokenizer.tokenize(text)
if self.max_tokens:
tokens = tokens[: self.max_tokens]
text_field = TextField(tokens, self.token_indexers)
label_field = LabelField(sentiment)
yield Instance({"text": text_field, "label": label_field})
2.訓練・検証データ読み込み
訓練・検証データの読み込みを行います。ファイルパスを入力すると、各データがInstanceとなって配列形式で返ってきます。
def read_data(reader: DatasetReader) -> Tuple[List[Instance], List[Instance]]:
print("Reading data")
training_data = list(reader.read("quick_start/data/movie_review/train.tsv"))
validation_data = list(reader.read("quick_start/data/movie_review/dev.tsv"))
return training_data, validation_data
3.語彙の定義
Instancesから語彙を定義します。簡単ですね。
def build_vocab(instances: Iterable[Instance]) -> Vocabulary:
print("Building the vocabulary")
return Vocabulary.from_instances(instances)
4.データローダーの作成
データローダーの作成です。訓練・検証データ各々で作成し、バッチサイズを8と指定しています。
def build_data_loaders(
train_data: List[Instance],
dev_data: List[Instance],
) -> Tuple[DataLoader, DataLoader]:
train_loader = SimpleDataLoader(train_data, 8, shuffle=True)
dev_loader = SimpleDataLoader(dev_data, 8, shuffle=False)
return train_loader, dev_loader
5.モデル定義
以下で分類モデル定義します。今回作成するモデルは以下のようなモデルです。トークンID列を単語ベクトルに直し(2次元配列状)、次にテキストベクトル(1次元配列状)に直して、最後にLinear層で学習を行います。
以下プログラムでは、forward関数でモデルを定義しています。self.embedderがword embeddingsへの変換、self.encoderがsequence embeddingsへの変換を行う層に当たり、最後にLinear層で分類を行っています。また、util.get_text_field_maskはバッチごとに系列長を揃えるためのパディングを行っています。
class SimpleClassifier(Model):
def __init__(
self, vocab: Vocabulary, embedder: TextFieldEmbedder, encoder: Seq2VecEncoder
):
super().__init__(vocab)
self.embedder = embedder
self.encoder = encoder
num_labels = vocab.get_vocab_size("labels")
self.classifier = torch.nn.Linear(encoder.get_output_dim(), num_labels)
self.accuracy = CategoricalAccuracy()
def forward(
self, text: TextFieldTensors, label: torch.Tensor = None
) -> Dict[str, torch.Tensor]:
# print("In model.forward(); printing here just because binder is so slow")
# Shape: (batch_size, num_tokens, embedding_dim)
embedded_text = self.embedder(text)
# Shape: (batch_size, num_tokens)
mask = util.get_text_field_mask(text)
# Shape: (batch_size, encoding_dim)
encoded_text = self.encoder(embedded_text, mask)
# Shape: (batch_size, num_labels)
logits = self.classifier(encoded_text)
# Shape: (batch_size, num_labels)
probs = torch.nn.functional.softmax(logits)
# Shape: (1,)
output = {"probs": probs}
if label is not None:
self.accuracy(logits, label)
output["loss"] = torch.nn.functional.cross_entropy(logits, label)
return output
def get_metrics(self, reset: bool = False) -> Dict[str, float]:
return {"accuracy": self.accuracy.get_metric(reset)}
6.訓練準備
ここで訓練を行う準備です。勾配計算を行うパラメータの列挙、最適化関数の設定、学習方法の指定を行い、Trainerオブジェクトを生成します。メイン部分に戻り、trainer.train()で訓練が始まります。
def build_trainer(
model: Model,
serialization_dir: str,
train_loader: DataLoader,
dev_loader: DataLoader,
) -> Trainer:
parameters = [(n, p) for n, p in model.named_parameters() if p.requires_grad]
optimizer = AdamOptimizer(parameters) # type: ignore
trainer = GradientDescentTrainer(
model=model,
serialization_dir=serialization_dir,
data_loader=train_loader,
validation_data_loader=dev_loader,
num_epochs=5,
optimizer=optimizer,
)
return trainer
訓練過程は以下のようになりました。
Reading data
building vocab: 100%|##########| 1800/1800 [00:01<00:00, 1481.02it/s]
You provided a validation dataset but patience was set to None, meaning that early stopping is disabled
accuracy: 0.7762, batch_loss: 0.1948, loss: 0.4676 ||: 100%|##########| 200/200 [00:04<00:00, 44.60it/s]
accuracy: 0.8500, batch_loss: 0.3786, loss: 0.3854 ||: 100%|##########| 25/25 [00:00<00:00, 482.24it/s]
accuracy: 0.9881, batch_loss: 0.0208, loss: 0.0576 ||: 100%|##########| 200/200 [00:04<00:00, 47.53it/s]
accuracy: 0.8150, batch_loss: 0.6130, loss: 0.4054 ||: 100%|##########| 25/25 [00:00<00:00, 489.77it/s]
accuracy: 1.0000, batch_loss: 0.0070, loss: 0.0105 ||: 100%|##########| 200/200 [00:04<00:00, 46.06it/s]
accuracy: 0.8250, batch_loss: 0.4147, loss: 0.3643 ||: 100%|##########| 25/25 [00:00<00:00, 482.15it/s]
accuracy: 1.0000, batch_loss: 0.0005, loss: 0.0036 ||: 100%|##########| 200/200 [00:04<00:00, 46.01it/s]
accuracy: 0.8300, batch_loss: 0.3910, loss: 0.3648 ||: 100%|##########| 25/25 [00:00<00:00, 419.68it/s]
accuracy: 1.0000, batch_loss: 0.0032, loss: 0.0019 ||: 100%|##########| 200/200 [00:04<00:00, 46.49it/s]
accuracy: 0.8350, batch_loss: 0.3851, loss: 0.3682 ||: 100%|##########| 25/25 [00:00<00:00, 536.92it/s]
予測
最後に学習したモデルを用いて、任意のデータで予測を行います。予測用のクラスとしてPredictorを継承した以下のクラスを定義します。これはテキスト分類用の予測クラスです。AllenNLPでは予測用のクラスをタスクに応じて自作する必要があります。ただし、典型的なタスクにおいては事前に準備されているものがあるようです。詳しくはドキュメント参照。
class SentenceClassifierPredictor(Predictor):
def predict(self, sentence: str) -> JsonDict:
return self.predict_json({"sentence": sentence})
@overrides
def _json_to_instance(self, json_dict: JsonDict) -> Instance:
sentence = json_dict["sentence"]
sentence = self._dataset_reader.tokenizer.tokenize(sentence)
print(sentence)
return self._dataset_reader.text_to_instance(sentence)
predictor = SentenceClassifierPredictor(model, dataset_reader)
output = predictor.predict("A good movie!")
print(
[
(vocab.get_token_from_index(label_id, "labels"), prob)
for label_id, prob in enumerate(output["probs"])
]
)
output = predictor.predict("This was a monstrous waste of time.")
print(
[
(vocab.get_token_from_index(label_id, "labels"), prob)
for label_id, prob in enumerate(output["probs"])
]
)
実行結果は以下となり、良さそうな結果が得られました。
[A, good, movie!]
[('neg', 0.47265103459358215), ('pos', 0.5273489952087402)]
[This, was, a, monstrous, waste, of, time.]
[('neg', 0.5896211266517639), ('pos', 0.41037890315055847)]
まとめ
以上でStep1のチュートリアルを終わります。今回Pythonのプログラムから学習を行いましたが、configファイルやコマンドからモデルを作成することも出来ます。方法は改めてまた追加できると良いかなと思っています。変なことを書いてたらご指摘くださると幸いです。