NLP前処理ツール「chariot」とPyTorchで日本語テキスト分類
chakki-worksが発表した自然言語処理の前処理ツール「chariot」をさっそく動かしてみました。
公式サンプルは英語でのコードだったので、せっかくなので日本語のデータを使ってPyTorchの日本語分類モデルの学習を行ってみました。
chariotのgithubはこちら
実験で使用したコードはGithubにもあげています。
chariotとは
chariotは自然言語処理の前処理のパイプラインを構築するためのツールです。
前処理を実行するPreprocessorに前処理をstackしていくことで、パイプラインを構築していきます。
公式の紹介記事では以下をポイントにあげています。
chariotでは、以下3点をできるようにしました。 - 前処理を宣言的に定義できる - 前処理が簡単に保存できる(パラメーターごと) - モデルの学習にたどり着くまでの時間を短くする
前処理のモジュールを順々に積み上げていくため、どんなパイプラインが構築されているか把握しやすいのはとても嬉しい特徴です。
preprocessor = Preprocessor() preprocessor\ .stack(ct.text.UnicodeNormalizer())\ .stack(ct.Tokenizer("en"))\ .stack(ct.token.StopwordFilter("en"))\ .stack(ct.Vocabulary(min_df=5, max_df=0.5))\ .fit(train_data)
またこのパイプラインのオブジェクトをpickleなどで保存しくことで、本番環境にパイプラインをそのまま組みこめるようにポータビリティ性も意識されています。
chariotを使った前処理モデルの構築
基本的な使い方は元記事にサンプルコードが載っているので、こちらではそれの簡単な紹介程度に留めます。
文章のリストに対して前処理のパイプラインを定義するのが Preprocessor
です。
import chariot.transformer as ct from chariot.preprocessor import Preprocessor preprocessor = Preprocessor() preprocessor\ .stack(ct.text.UnicodeNormalizer())\ .stack(ct.Tokenizer("en"))\ .stack(ct.token.StopwordFilter("en"))\ .stack(ct.Vocabulary(min_df=5, max_df=0.5))\ .fit(train_data) # scikit-learnのtransformerと同じ感覚で使える processed = preprocessor.transform(train_data) # パイプラインをモデルとしてpickleで保存 preprocessor.save("my_preprocessor.pkl")
Preprocessorは文章のリストに対するものですが、CSVを読み込んで複数の列にまとめて前処理を定義したいことも多いと思います。その時に使うのが DatasetPreprocessor
です。
これはDataFrameの各列にPreprocessorを定義しているようなものです。
from chariot.dataset_preprocessor import DatasetPreprocessor from chariot.transformer.formatter import Padding dp = DatasetPreprocessor() dp.process("review")\ .by(ct.text.UnicodeNormalizer())\ .by(ct.Tokenizer("en"))\ .by(ct.token.StopwordFilter("en"))\ .by(ct.Vocabulary(min_df=5, max_df=0.5))\ .by(Padding(length=pad_length))\ .fit(train_data["review"]) dp.process("polarity")\ .by(ct.formatter.CategoricalLabel(num_class=3)) # 各列に前処理をかけた結果がdictで返ってくる preprocessed = dp.preprocess(data) # 各列に対するpreorocessorがtarで固まって保存される dp.save("my_dataset_preprocessor.tar.gz")
ちなみにpreprocessorの内部でパイプラインを構築する際に、stackしたモジュールは簡単に順番を並び替えてパイプラインが構築されます。
具体的には テキストへの前処理系(unicode正規化とか)
-> tokenizer
-> tokenへの前処理系(stop word filterとか)
-> vocabulary系(vocabの作成)
という形で処理が進むようにパイプラインの中身を並び替えます。
presetのtransformer
Preprocessor/DatasetPreprocessorで使用できる前処理用のモジュールはデフォルトで用意されています。
モジュール群はGithubのtransformerのコード部分を見ると中身を読むことができます。
https://github.com/chakki-works/chariot/tree/master/chariot/transformer
ここではいくつか紹介します。
text/UnicodeNormalizer
Unicode正規化を行うモジュールです。
内部的には受け取ったテキストに unicodedata.normalize(from, text)
を実施しています。
正規化フォームは自分で選べますが、多くの場合は NFKC
で良いと思います
class UnicodeNormalizer(TextNormalizer): def __init__(self, form="NFKC", copy=True): super().__init__(copy) self.form = form def apply(self, text): return unicodedata.normalize(self.form, text)
text/SymbolFilter
指定した記号をテキスト中から除去するモジュールです。
class SymbolFilter(TextFilter): def __init__(self, filters='!"#$%&()*+,-./:;<=>?@[\\]^_`{|}~\t\n', split=" ", copy=True): super().__init__(copy) self.filters = filters self.split = split def apply(self, text): translate_dict = dict((c, self.split) for c in self.filters) translate_map = str.maketrans(translate_dict) _text = text.translate(translate_map) return _text
tokenizer/Tokenizer
テキストのTokenizeを行うモジュールです。
日本語では、Mecabが使える環境ではMecabを、そうでなければJanomeを使ってTokenizeを行います。
日本語以外ではSpacyが持っているtokenizerを使用するため、Spacyが対応している言語であればchariotの中で対応可能(なはず)です。
class Tokenizer(BasePreprocessor): def __init__(self, lang="en", copy=True): super().__init__(copy) self.lang = lang self._tokenizer = None self.set_tokenizer() def set_tokenizer(self): if self.lang == "ja": try: self.tokenizer = MeCabTokenizer() except Exception as ex: self.tokenizer = JanomeTokenizer() elif self.lang is None: self.tokenizer = SplitTokenizer() else: self.tokenizer = SpacyTokenizer(self.lang) def transform(self, X): _X = apply_map(X, self.tokenizer.tokenize, self.copy) return _X def __reduce_ex__(self, proto): return type(self), (self.lang, self.copy)
token/StopwordFilter
Stop wordを除去するモジュールです。
日本語以外では、Spacyのtokenize結果をもとにして除去します。
日本語の場合、デフォルトで用意したStop words辞書を使って辞書形式で削除します。
class StopwordFilter(TokenFilter): def __init__(self, lang, copy=True): super().__init__(copy) self.lang = lang def apply(self, tokens): if len(tokens) == 0: return tokens else: sample = tokens[0] if sample.is_spacy: return [t for t in tokens if not t._token.is_stop] elif sample.is_ja: return [t for t in tokens if t.surface not in STOP_WORD_JA] else: return tokens
formatter/CategoricalLabel
ラベルなどのカテゴリ列をone-hotの表現に変換します。
class CategoricalLabel(BaseFormatter): def __init__(self, num_class=-1): super().__init__() self.num_class = num_class def transfer_setting(self, vocabulary_or_preprocessor): vocabulary = vocabulary_or_preprocessor if isinstance(vocabulary_or_preprocessor, Preprocessor): vocabulary = vocabulary_or_preprocessor.vocabulary self.num_class = vocabulary.count def transform(self, column): y = np.array(column, dtype="int") input_shape = y.shape if input_shape and input_shape[-1] == 1 and len(input_shape) > 1: input_shape = tuple(input_shape[:-1]) y = y.ravel() n = y.shape[0] categorical = np.zeros((n, self.num_class), dtype=np.float32) categorical[np.arange(n), y] = 1 output_shape = input_shape + (self.num_class,) categorical = np.reshape(categorical, output_shape) return categorical def inverse_transform(self, batch): return np.argmax(batch, axis=1)
chariotを使った日本語前処理
chariotを使って日本語ドキュメントの分類を行うPyTorchのモデルを学習させるコードを作成します。
今回はlivedoorのニュースコーパスを使って、ニュースメディアの予測をする9クラス分類を行います。
サイトからコーパスをダウンロードして展開し、pythonで読み込みます。
import os from pathlib import Path # タイトル、記事、メディアを収集 titles, articles, labels = [], [], [] news_list = ['dokujo-tsushin', 'it-life-hack', 'kaden-channel', 'livedoor-homme', 'movie-enter', 'peachy', 'smax', 'sports-watch', 'topic-news'] for i, media in enumerate(news_list): files = os.listdir(Path('text', media)) for file_name in files: if file_name == 'LICENSE.txt': continue with Path('text', media, file_name).open(encoding='utf-8') as f: lines = [line for line in f] title = lines[2].replace('\n', '') text = ''.join(lines[3:]) titles.append(title) articles.append(text.replace('\n', '')) labels.append(i)
データ数はこんな感じです。
分量は大体同じくらい。ちょっとlivedoor HOMMEが少ないくらいです。
from collections import Counter counter = Counter(labels) for com in counter.most_common(): print(com) >>> (7, 900) (0, 870) (1, 870) (4, 870) (6, 870) (2, 864) (5, 842) (8, 770) (3, 511)
データをDataFrameにまとめていきます。あとtrainとtestにデータを分けます。
import pandas as pd from sklearn.model_selection import train_test_split df = pd.DataFrame({'title': titles, 'article': articles, 'label': labels}) train_X, test_X, train_y, test_y = train_test_split(df[['article', 'title']], df['label'], stratify=df['label']) train_df = train_X train_df['label'] = train_y test_df = test_X test_df['label'] = test_y
データの中身を見てみると、ノイズになりそうなテキストが含まれています。
今はどんな内容が話題なの? 今、どんな話題が流行ってるのか知りたくないですか?ちょっとした暇つぶしやブログのネタ探し、営業職の方や美容師さんなどお客さんとの話題作りにも役立てることができるアプリを紹介したいと思います。今回紹介するAndroid向けアプリ「ワダイくん」は旬のキーワードなどをランキング形式で表示してくれます。20分ごとに更新されるのでテレビやSNS、インターネット上で話題のキーワードを旬のうちに確認することができるようになっています。では、早速、紹介していきたいと思います。TOP画面にはその日の話題が20位まで表示されます。気になる話題をタップするとウェブ、イメージ、ニュース、ブログでのカテゴリ別に閲覧することができるようになっています。ウェブ表示イメージ表示ニュース表示ブログ表示今、話題のキーワードをランキング形式でチェックできるので話題に乗り遅れることはないかも!?記事執筆:にゃんこアプリ名:ワダイくん -話題のキーワードランキング-価格:無料カテゴリ:ライフスタイル開発者:catchyバージョン:4.0ANDROID 要件:2.2以上Google Play Store:http://play.google.com/store/apps/details?id=com.trand■関連リンク・エスマックス(S-MAX)・エスマックス(S-MAX) smaxjp on Twitterhttp://livedoor.blogimg.jp/smaxjp/imgs/1/2/12e4b263-s.png000
そこで、正規表現を入力してテキストをフィルタリングする自作のモジュールを作成してchariotのパイプラインに組み込もうと思います。
from chariot.transformer.text.base import TextFilter import re class RegularExpressionReplacer(TextFilter): def __init__(self, pattern, replacement, copy=True): super().__init__(copy) self.pattern = pattern self.replacement = replacement def apply(self, text): # patternにマッチした部分文字列をreplacementに置き換える return re.sub(self.pattern, self.replacement, text)
最終的なパイプラインは以下のように定義しました。
from chariot.dataset_preprocessor import DatasetPreprocessor from chariot.transformer.formatter import Padding import chariot.transformer as ct pad_length = 300 dp = DatasetPreprocessor() dp.process('article')\ .by(ct.text.UnicodeNormalizer())\ .by(ct.text.LowerNormalizer())\ .by(RegularExpressionReplacer(pattern='■.+?。', replacement=''))\ # 自作モジュール .by(ct.text.SymbolFilter())\ .by(ct.Tokenizer('ja'))\ .by(ct.token.StopwordFilter('ja'))\ .by(ct.Vocabulary(min_df=3, max_df=0.8))\ .by(Padding(length=pad_length))\ .fit(train_df['article']) dp.process('label')\ .by(ct.formatter.CategoricalLabel(num_class=9)) # 各列に前処理を施したdictが保存されている preprocessed = dp.preprocess(train_df)
PyTorchのモデル作成に向けた準備をしていきます。今回は記事本文からメディアを予測することにします。
まずは記事本体のarticle列からvocabularyを作成し、そこから学習済み単語ベクトルの取得まで行います。
学習済み単語ベクトルはchakinを使ってfasttextの単語ベクトルを利用しました。
# 特定の列に対応したpreprocessorはこれで取得できる # 列に対応したvocabularyの一覧はこのようにリストで取得できる dp.process('article').preprocessor.vocabulary.get()[:3] from chariot.storage import Storage import os import sys from pathlib import Path def set_path(): if "../" not in sys.path: sys.path.append("../") root_dir = Path.cwd() return root_dir ROOT_DIR = set_path() storage = Storage.setup_data_dir(ROOT_DIR) # fasttextの学習済み単語ベクトルをダウンロード _ = storage.chakin(name='fastText(ja)') # article列のvocabularyに対応した埋め込みベクトルのmatrixを生成する embedding = dp.process('article').preprocessor.vocabulary.make_embedding(storage.data_path('external/fasttext(ja).vec'))
テキスト分類を行うモデルとして、Convolutional Neural Networks for Sentence Classificationで使用されたCNNのテキスト分類モデルを使用します。
import torch import torch.nn as nn import torch.nn.functional as F class CNN_Text(nn.Module): def __init__(self, pretrained_wv, output_dim, kernel_num, kernel_sizes=[3,4,5], dropout=0.5, static=False): super(CNN_Text,self).__init__() weight = torch.from_numpy(pretrained_wv) self.embed = nn.Embedding.from_pretrained(weight, freeze=False) self.convs1 = nn.ModuleList([nn.Conv2d(1, kernel_num, (k, self.embed.weight.shape[1])) for k in kernel_sizes]) self.dropout = nn.Dropout(dropout) self.fc1 = nn.Linear(len(kernel_sizes)*kernel_num, output_dim) self.static = static def conv_and_pool(self, x, conv): x = F.relu(conv(x)).squeeze(3) #(N,Co,W) x = F.max_pool1d(x, x.size(2)).squeeze(2) return x def forward(self, x): x = self.embed(x) # (N,W,D) if self.static: x = x.detach() x = x.unsqueeze(1) # (N,Ci,W,D) x = x.float() x = [F.relu(conv(x)).squeeze(3) for conv in self.convs1] #[(N,Co,W), ...]*len(Ks) x = [F.max_pool1d(i, i.size(2)).squeeze(2) for i in x] #[(N,Co), ...]*len(Ks) x = torch.cat(x, 1) x = self.dropout(x) # (N,len(Ks)*Co) logit = self.fc1(x) # (N,C) return logit
あとはゴリゴリっと学習をまわすだけです。学習のためのミニバッチ作成とepoch管理もDatasetPreprocessorが行ってくれます。
output_dim = 9 kernel_num = 100 kernel_sizes = [3,4,5] dropout = 0.5 model = CNN_Text(embedding, output_dim, kernel_num, kernel_sizes, dropout) use_cuda = True # GPUを使う場合 opt = torch.optim.Adam(model.parameters(), lr=0.001) model.train() if use_cuda: model = model.cuda() loss_list = [] for b in dp.iterate(preprocessed, batch_size=32, epoch=15): feature = torch.LongTensor(b['article']) # PyTorchはラベルのone-hot matrixに対応していないので、ラベルの数字に戻す target = torch.LongTensor(np.argmax(b['label'], axis=1)) if use_cuda: feature = feature.cuda() target = target.cuda() opt.zero_grad() logit = model(feature) loss = F.nll_loss(F.log_softmax(logit), target) loss.backward() opt.step() loss_list.append(loss.item())
学習の結果をTestデータセットで確認してみます。
先ほど作成したTest用のDataFrameをパイプラインにそのまま通せば先ほど設定した前処理を適用することができます。
# 前処理パイプラインを通したデータがdictで出てくる test_preprocessed = dp(test_df).preprocess().format().processed model.eval() feature = torch.LongTensor(test_preprocessed['article']).cuda() forward = model(feature) predicted_label = forward.argmax(dim=1).cpu() test_target = torch.LongTensor(np.argmax(test_preprocessed['label'], axis=1))
結果は以下の通り。微妙なのでもうちょっと前処理は工夫の余地がありそうです。
from sklearn.metrics import classification_report print(classification_report(test_target.numpy(), predicted_label.numpy())) >>> precision recall f1-score support 0 0.60 0.64 0.62 218 1 0.66 0.67 0.67 217 2 0.57 0.56 0.56 216 3 0.49 0.49 0.49 128 4 0.72 0.71 0.71 218 5 0.62 0.57 0.59 210 6 0.82 0.79 0.80 218 7 0.65 0.75 0.69 225 8 0.65 0.58 0.62 192 micro avg 0.65 0.65 0.65 1842 macro avg 0.64 0.64 0.64 1842 weighted avg 0.65 0.65 0.65 1842
from sklearn.metrics import accuracy_score print(accuracy_score(test_target.numpy(), predicted_label.numpy())) >>> 0.6487513572204126
最後に
前処理のパイプラインを分かりやすく、効率的に、ポータビリティをもって構築できるツール「chariot」を紹介し、実際に日本語でPyTorchと組み合わせてテキスト分類を行いました。
Vocabularyの作成、SOSやUNKといった特殊なトークンの挿入、文のpaddingは結構面倒なので、そこをまるっとやってくれるのは便利だと思います。
前処理の実行とembedding matrixの作成はそれなりに時間を取られる(モジュールの数にもよりますが1時間前後)ので、もっと高速化できると嬉しいです。
実は前職で似たようなパイプライン作成ツールを社内で自作して使用しており、似たツールが出たので調査も兼ねて記事にしたのでした。(退職に間に合わなかったよ...><;)
ぜひchariotを使ってproduction readyなNLP前処理パイプラインを作ってみてください!