概要
エンコード済みデータセットの作成で作成したデータセットを用いて、固有表現抽出用に、BERTモデルをファインチューニングします。
ポートフォリオとして、自作のデータセットでファインチューニングした言語モデルを使ったアプリを公開しました。
この記事を読むだけでも、この記事の内容をある程度理解できるとは思いますが、プロジェクト全体を理解している前提で書かれています。
プロジェクト全体の概要については、こちらの記事をご覧ください。
この記事で作成されるもの
固有表現抽出BERTモデル
モデルのテスト結果
eval_f1 0.9961977186311787
eval_accuracy 0.9995689655172414
eval_precision 0.9940978077571669
eval_recall 0.9983065198983911
自作データセットでファインチューニングしたため、全ての入力文が似たような構造になり、このようなテスト結果になったと考えられます。
しかし、この記事の方針での考察と、それに応じた問題への対策方法が有効だったように見えるので、その点は良かったかなと思います。
補足情報
開発はGoogle Colaboratoryで行われ、このノートブックで作成しました。
このノートブックを含むリポジトリの構造は、実際の開発環境と同一です。
from
で参照されている自作モジュールは、このディレクトリにあるものです。
その他のコード内で参照しているパスやディレクトリから/content/drive/MyDrive
を省くと、そのパスやディレクトリの中身をリポジトリから確認することができます。
方針
エンコード済みデータセットの作成で作成したデータセットを使って、モデルをファインチューニングし、性能をテストします。
”cl-tohoku/bert-base-japanese-v2”を、Transformersのトレーナーを使ってファインチューニングします。
大まかな手順
- エンコード済みデータセットの作成で作成したデータセットをもとに、トレーナーとテスト用データセットの作成
- トレーナーによる、モデルのファインチューニング
(early stoppingにより、3エポックで終了しました。) - ファインチューニング済みモデルのテスト
コード
import、install
from google.colab import drive
drive.mount('/content/drive')
!pip install datasets
!pip install seqeval
!pip install transformers[torch]
!pip install transformers fugashi ipadic
!pip install unidic-lite
from typing import List, Dict, Tuple
import random
import numpy as np
import pandas as pd
from sklearn.model_selection import train_test_split
import torch
from datasets import Dataset, Features, Sequence, ClassLabel, Value
from seqeval.metrics import (
accuracy_score, f1_score, precision_score, recall_score
)
from transformers import (
BertConfig,
BertForTokenClassification,
TrainingArguments,
BertJapaneseTokenizer,
DataCollatorForTokenClassification,
EarlyStoppingCallback,
Trainer,
pipeline
)
from transformers.trainer_utils import EvalPrediction
import sys
sys.path.append('/content/drive/MyDrive/local_cuisine_search_app/modules')
from utility import load_json_obj
from pandas_utility import read_csv_df
from pipeline import NaturalLanguageProcessing
関数とクラスの定義
def train_test_and_save(
dataset_path: str,
id2label_path: str,
model_name: str,
checkpoint_dir: str,
logs_dir: str,
seed: int = 42
) -> None:
"""
ファインチューニングとテストとモデルの保存
Parameters
----------
dataset_path : str
データセットのデータフレームが保存されているパス
id2label_path : str
ラベルのidとラベルの辞書が保存されているパス
model_name : str
事前学習済みモデルの名前
checkpoint_dir : str
チェックポイントを保存するディレクトリ
logs_dir : str
ログを保存するディレクトリ
seed : int, optional
乱数固定用のシード値, by default 42
"""
TrainingController.set_seed(seed)
trainer, test_dataset = TrainingController.create_trainer_and_test_dataset(
dataset_path, id2label_path, model_name, checkpoint_dir, logs_dir, seed
)
trainer.train(
ignore_keys_for_eval=['last_hidden_state', 'hidden_states', 'attentions']
)
TrainingController.test(trainer, test_dataset)
class TrainingController:
"""
学習用ヘルパークラス
Attributes
----------
_not_show_metrics : List[str]
テスト結果で表示させないmetricsのリスト
"""
_not_show_metrics = ['eval_samples_per_second', 'eval_steps_per_second']
@staticmethod
def set_seed(seed: int = 42) -> None:
"""
シードの固定
Parameters
----------
seed : int, optional
シード値, by default 42
"""
random.seed(seed)
np.random.seed(seed)
torch.manual_seed(seed)
if torch.cuda.is_available():
torch.cuda.manual_seed_all(seed)
@staticmethod
def create_trainer_and_test_dataset(
dataset_path: str,
id2label_path: str,
model_name: str,
checkpoint_dir: str,
logs_dir: str,
seed: int = 42
) -> Tuple[Trainer, Dataset]:
"""
トレーナーとテスト用データセットの作成
Parameters
----------
dataset_path : str
データセットのデータフレームが保存されているパス
id2label_path : str
ラベルのidとラベルの辞書が保存されているパス
model_name : str
事前学習済みモデル名
checkpoint_dir : str
チェックポイントを保存するディレクトリ
logs_dir : str
ログを保存するディレクトリ
seed : int, optional
乱数固定用のシード値, by default 42
Returns
-------
Tuple[Trainer, Dataset]
トレーナーとテスト用データセット
"""
id2label = load_json_obj(id2label_path)
id2label: Dict[int, str] = {
int(id): label for id, label in id2label.items()
}
label2id = {label: id for id, label in id2label.items()}
train_dataset, eval_dataset, test_dataset = DatasetMaker.create_datasets(
dataset_path, label2id, seed
)
trainer = TrainerMaker.create(
model_name, id2label, label2id, checkpoint_dir, logs_dir,
train_dataset, eval_dataset
)
return trainer, test_dataset
@staticmethod
def test(trainer: Trainer, test_dataset: Dataset) -> None:
"""
テスト
Parameters
----------
trainer : Trainer
トレーナー
test_dataset : Dataset
テスト用データセット
"""
print('\nテスト')
results = trainer.evaluate(test_dataset)
for metric_name, metric in results.items():
if metric_name not in TrainingController._not_show_metrics:
print(f'{metric_name: <20} {metric}')
class DatasetMaker:
"""
データセット作成用クラス
"""
@staticmethod
def create_datasets(
dataset_path: str, label2id: Dict[str, int], seed: int = 42
) -> Tuple[Dataset, Dataset, Dataset]:
"""
全データセットの作成
訓練用、評価用、テスト用のデータセットを作成する
Parameters
----------
dataset_path : str
データセットのデータフレームが保存されているパス
label2id : Dict[str, int]
ラベルのidとラベルの辞書が保存されているパス
seed : int, optional
乱数固定用のシード値, by default 42
Returns
-------
Tuple[Dataset, Dataset, Dataset]
訓練用データセット、評価用データセット、テスト用データセットのタプル
"""
dataset_df = read_csv_df(dataset_path)
train_df, eval_df, test_df = DatasetMaker._split_dataset_df(
dataset_df, seed
)
train_dataset, eval_dataset, test_dataset = DatasetMaker._df_to_dataset(
train_df, eval_df, test_df, label2id
)
return train_dataset, eval_dataset, test_dataset
@staticmethod
def _split_dataset_df(
dataset_df: pd.DataFrame, seed: int = 42
) -> Tuple[pd.DataFrame, pd.DataFrame, pd.DataFrame]:
"""
データセットの分割
Parameters
----------
dataset_df : pd.DataFrame
全データのデータフレーム
seed : int, optional
乱数固定用のシード値, by default 42
Returns
-------
Tuple[pd.DataFrame, pd.DataFrame, pd.DataFrame]
訓練用データフレーム、評価用データフレーム、テスト用データフレームのタプル
"""
train_eval_df, test_df = train_test_split(
dataset_df, test_size=0.15, random_state=seed
)
train_df, eval_df = train_test_split(
train_eval_df, test_size=0.2, random_state=seed
)
return train_df, eval_df, test_df
@staticmethod
def _df_to_dataset(
train_df: pd.DataFrame,
eval_df: pd.DataFrame,
test_df: pd.DataFrame,
label2id: Dict[str, int]
) -> Tuple[Dataset, Dataset, Dataset]:
"""
全データフレームから全データセットの作成
Parameters
----------
train_df : pd.DataFrame
訓練用データフレーム
eval_df : pd.DataFrame
評価用データフレーム
test_df : pd.DataFrame
テスト用データフレーム
label2id : Dict[str, int]
ラベルとラベルのidの辞書
Returns
-------
Tuple[Dataset, Dataset, Dataset]
訓練用データセット、評価用データセット、テスト用データセットのタプル
"""
input_ids, attention_mask, labels = train_df.columns
label_names = list(label2id.keys())
features = Features({
input_ids: Sequence(Value('int64')),
attention_mask: Sequence(Value('int64')),
labels: Sequence(ClassLabel(names=label_names))
})
train_dataset = DatasetMaker._create_dataset(train_df, features)
eval_dataset = DatasetMaker._create_dataset(eval_df, features)
test_dataset = DatasetMaker._create_dataset(test_df, features)
return train_dataset, eval_dataset, test_dataset
@staticmethod
def _create_dataset(df: pd.DataFrame, features: Features) -> Dataset:
"""
データセットの作成
Parameters
----------
df : pd.DataFrame
データフレーム
features : Features
データセットの構造を示すオブジェクト
Returns
-------
Dataset
データセット
"""
df.reset_index(drop=True, inplace=True) # ※1
dataset = Dataset.from_pandas(df, features)
return dataset
class TrainerMaker:
@staticmethod
def create(
model_name: str,
id2label: Dict[int, str],
label2id: Dict[str, int],
checkpoint_dir: str,
logs_dir: str,
train_dataset: Dataset,
eval_dataset: Dataset
) -> Trainer:
"""
トレーナーの作成
Parameters
----------
model_name : str
事前学習済みモデルの名前
id2label : Dict[int, str]
ラベルのidとラベルの辞書
label2id : Dict[str, int]
ラベルとラベルのidの辞書
checkpoint_dir : str
チェックポイントを保存するディレクトリ
logs_dir : str
ログを保存するディレクトリ
train_dataset : Dataset
訓練用データセット
eval_dataset : Dataset
評価用データセット
Returns
-------
Trainer
トレーナー
"""
model = TrainerMaker._create_model(model_name, id2label, label2id)
training_args = TrainerMaker._create_training_args(
checkpoint_dir, logs_dir
)
tokenizer = BertJapaneseTokenizer.from_pretrained(model_name)
data_collator = DataCollatorForTokenClassification(tokenizer)
metrics_computer = MetricsComputer(id2label)
early_stopping = EarlyStoppingCallback(early_stopping_patience=2)
trainer = Trainer(
model=model,
args=training_args,
data_collator=data_collator,
train_dataset=train_dataset,
eval_dataset=eval_dataset,
tokenizer=tokenizer,
compute_metrics=metrics_computer.compute,
callbacks=[early_stopping]
)
return trainer
@staticmethod
def _create_model(
model_name: str, id2label: Dict[int, str], label2id: Dict[str, int]
) -> BertForTokenClassification:
"""
モデルの作成
Parameters
----------
model_name : str
事前学習済みモデルの名前
id2label : Dict[int, str]
ラベルのidとラベルの辞書
label2id : Dict[str, int]
ラベルと、ラベルのidの辞書
Returns
-------
BertForTokenClassification
モデル
"""
config = BertConfig.from_pretrained(
model_name, id2label=id2label, label2id=label2id
)
model = BertForTokenClassification.from_pretrained(
model_name, config=config
)
return model
@staticmethod
def _create_training_args(
checkpoint_dir: str, logs_dir: str
) -> TrainingArguments:
"""
トレーナーのargsの作成
Parameters
----------
checkpoint_dir : str
チェックポイントを保存するディレクトリ
logs_dir : str
ログを保存するディレクトリ
Returns
-------
TrainingArguments
トレーナーのargs
"""
training_args = TrainingArguments(
output_dir=checkpoint_dir,
evaluation_strategy='epoch',
per_device_train_batch_size=8,
per_device_eval_batch_size=32,
weight_decay=0.01,
num_train_epochs=5,
warmup_steps=500,
logging_dir=logs_dir,
logging_strategy='epoch',
save_strategy='epoch',
save_total_limit=1,
load_best_model_at_end=True,
metric_for_best_model='f1',
report_to='tensorboard'
)
return training_args
class MetricsComputer:
"""
モデルの性能計算用クラス
Attributes
----------
_id2label : Dict[int, str]
ラベルのidとラベルの辞書
"""
def __init__(self, id2label: Dict[int, str]):
self._id2label = id2label
def compute(self, pred: EvalPrediction) -> Dict[str, np.float64 | float]:
"""
性能の計算
Parameters
----------
pred : EvalPrediction
モデルの予測結果
Returns
-------
Dict[str, np.float64 | float]
性能の計算結果
"""
truth_ids_lst = pred.label_ids.tolist()
pred_ids_lst = pred.predictions.argmax(-1).tolist()
truths = [[self._id2label[id] for id in ids] for ids in truth_ids_lst]
preds = [[self._id2label[id] for id in ids] for ids in pred_ids_lst]
f1: np.float64 = f1_score(truths, preds)
accuracy: float = accuracy_score(truths, preds)
precision: np.float64 = precision_score(truths, preds)
recall: np.float64 = recall_score(truths, preds)
metrics = {
'f1': f1,
'accuracy': accuracy,
'precision': precision,
'recall': recall
}
return metrics
ファインチューニング、テスト、モデルの保存
dataset_path = '/content/drive/MyDrive/local_cuisine_search_app/data/processed_data/04_encoded_dataset_dataframe/encoded_dataset_dataframe.csv'
id2label_path = '/content/drive/MyDrive/local_cuisine_search_app/data/processed_data/03_labels_dictionary/labels_dictionary.json'
model_name = 'cl-tohoku/bert-base-japanese-v2'
checkpoint_dir = '/content/drive/MyDrive/local_cuisine_search_app/data/processed_data/05_finetuned_model/checkpoint'
logs_dir = '/content/drive/MyDrive/local_cuisine_search_app/data/processed_data/05_finetuned_model/logs'
train_test_and_save(
dataset_path, id2label_path, model_name, checkpoint_dir, logs_dir
)
ファインチューニング済みモデルの動作確認
model_dir = '/content/drive/MyDrive/local_cuisine_search_app/data/processed_data/05_finetuned_model/checkpoint/checkpoint-164'
nlp = NaturalLanguageProcessing(model_dir)
input = '仙豆を使った愛知県の料理を教えて下さい'
classified_words = nlp.classify_and_show(input)
INGR 仙豆
AREA 愛知県
ログの確認
%load_ext tensorboard
%tensorboard --logdir '/content/drive/MyDrive/local_cuisine_search_app/data/processed_data/05_finetuned_model/logs'
メモ
※1
データフレームからデータセットを作成する際は、インデックス列を渡しません。