1.はじめに
今回 Aidemyのpremium planで三か月の間、自然言語処理を学びその知識の定着、復習の意を込め成果物として深層学習によるチャットボットを作成してみました。
Qiitaによる投稿も初めてですので拙い文章だとは思いますが、それでも良い方はご覧ください！
2.使用環境
今回は処理するデータ量が多いためGoogle Colaboratory環境で、無料で使えるGPUを使用しました。
フレームワークのバージョンは以下の通りです。
・janome 0.4.1
・torchvision 0.7.0
・torchtext 0.7.0
・PyTorch 1.6.0
3.記事の内容
タイトルにもあるように、今回は会話の入力に対して、深層学習を用いて応答を予測し出力させるモデルを構築し、ノートブック上で繰り返し会話できるボットを作ることを目標にコーディングしました。
深層学習のフレームワークとしてKeras、Tensorflow等、いろいろなものがありますが今回は最近流行りのPyTorchを使いました。
PyTorchを選んだ理由は、以下のように考えたからです。
① コードが比較的短く、理解すれば便利である
② 入力と正解のペアをまとめて学習できるため、会話を成立させるという
点で有利に働く
③ 今後の開発につながる
また、学習するにあたりデータの前処理、PyTorchによる深層学習はudemyの講座を参考にさせていただきました。
PyTorchによる学習は初めてだったので、かなりつまずきながら手探り状態でしたが、自分でコーディングすることでPyTorchの概要、実装方法などかなり実践的な知識が身についたと思います。
4.実装の手順
それではコードのほうに移っていきます。
工程は大まかに分けて3つです。
① データの前処理
今回訓練に使用した会話コーパスは、NTT ドコモが一般公開している雑談会話コーパスになります。ライセンスはダウンロードページ記載のものに準拠します。
こちらは約1100もの対話が収録されている会話コーパスで、雑談対話APIと人間との間で会話が行われているため、明らかに破綻している会話も収録されているという点も考慮して前処理を行います。
また、チャットボットの作成には約10万もの会話データが必要になるといわれているので今回は一つの入力文に対し、重要である単語を含んだ複数の応答文でペアを作るという方法を用いてデータの水増しを行っています。
以下が、コードになります。
qiita.py
!pip install janome==0.4.1
!pip install torchvision==0.7.0
!pip install torchtext==0.7.0
!pip install torch==1.6.0
# google driveをマウントします
from google.colab import drive
drive.mount("/content/drive/")
#先に必要なものをインポートしときます
import glob
import json
import re
from janome.tokenizer import Tokenizer
import csv
from sklearn.model_selection import train_test_split
import torch
import torch.nn as nn
import torch.nn.functional as F
from torch import optim
import dill
path = "/content/drive/My Drive/対話コーパス 前処理/NTT data/projectnextnlp-chat-dialogue-corpus/json"
# 成り立っている会話を訓練用データとテストデータに分割して、保存します。
files = glob.glob(path + "/*/*.json")
dialogues = []
for file in files:
with open(file, "r") as f:
json_dic = json.load(f)
dialogue = []
for turn in json_dic["turns"]: #会話データの["turns"]には各項目が格納されている
annotations = turn["annotations"] #注釈
speaker = turn["speaker"] #話者
utterance = turn["utterance"] #会話内容
utterance = utterance.replace(".", "。").replace(",", "、") # 全角を統一
utterance = utterance.replace("．", "。").replace("，", "、") # 半角を統一
utterance = utterance.split("。")[0] #ここでは"。"が文中にある長い文章を短くする
if speaker=="U": # "U"=人間
dialogue.append(utterance)
else:
is_wrong = False
for annotation in annotations:
breakdown = annotation["breakdown"] #会話が破綻しているかどうか
if breakdown=="X":
is_wrong = True
break
if is_wrong:
dialogue.clear() #破綻していたらデータをクリア
else:
dialogue.append(utterance)
if len(dialogue) >= 2: #会話成立で格納
dialogues.append(dialogue.copy())
dialogue.pop(0)
#会話の分かち書きを行い、入力と応答それぞれのリストに格納します。
re_kanji = re.compile(r"^[\u4E00-\u9FD0]+$") # 漢字の検出用
re_katakana = re.compile(r"[\u30A1-\u30F4]+") # カタカナの検出用
janome_tk = Tokenizer()
def tokenizer(text):
return [token for token in janome_tk.tokenize(text, wakati=True)]
wakati_input = [] #水増しのために入力文と応答文を分けます
wakati_reply = []
for dialogue in dialogues:
wakati_input.append(tokenizer(dialogue[0])[:10])
wakati_reply.append(tokenizer(dialogue[1])[:10])
#入力分を複数の応答分と組み合わせてデータの水増しを行います。
dialogues_mizumasi = [] #ここで水増しにより会話データ約10万個
for i, word_input in enumerate(wakati_input):
inp_count = 0
for j, word_reply in enumerate(wakati_reply):
if i==j:
dialogues_mizumasi.append(["".join(word_input), "".join(word_reply)])
continue
#入力と応答の類似度を判断し、類似度が1より大きいものを会話に追加します
similarity = 0
for word in word_input: #カタカナや漢字を重要付けている
if (word in word_reply) and (re_kanji.fullmatch(word) or re_katakana.fullmatch(word)):
similarity += 1
if similarity >= 1:
dialogue_mizumasi = ["".join(word_input), "".join(word_reply)]
if dialogue_mizumasi not in dialogues_mizumasi:
dialogues_mizumasi.append(dialogue_mizumasi)
inp_count += 1
if inp_count >= 12: #１つの入力に対し、対応する応答文は１２まで
break
dialogues = dialogues_mizumasi
#作成した会話のリストを訓練用データとテストデータに分けて格納します。
dialogues_train, dialogues_test = train_test_split(dialogues, shuffle=True, test_size=0.05)
#入力分と応答分のデータセットの列を定義します
input_field = torchtext.data.Field(
sequential=True, #文章データのため可変にする
tokenize=tokenizer, #分かち書き
batch_first=True,
lower=True
)
reply_field = torchtext.data.Field(
sequential=True,
tokenize=tokenizer,
init_token="<sos>", #文のはじめに加える
eos_token="<eos>", #文の終わりに加える
batch_first=True,
lower=True
)
path2 = "/content/drive/My Drive/対話コーパス 前処理/NTT data/"
#データセットを作成します
train_data, test_data = torchtext.data.TabularDataset.splits(
path=path2,
train=mizumasi_dialogues_train,
test=mizumasi_dialogues_test,
format="csv",
fields=[("input_text", input_field), ("reply_text", reply_field)]
)
#辞書の作成をします(出現頻度が3以下のものは除く)
input_field.build_vocab(
train_data,
min_freq=3
)
reply_field.build_vocab(
train_data,
min_freq=3,
)
#データセットを保存します
torch.save(train_data.examples, path+"mizumasi_train_examples.pkl", pickle_module=dill)
torch.save(test_data.examples, path+"mizumasi_test_examples.pkl", pickle_module=dill)
torch.save(input_field, path2+"mizumasi_input.pkl", pickle_module=dill)
torch.save(reply_field, path2+"mizumasi_reply.pkl", pickle_module=dill)
②訓練モデルを作成し、学習させる
次に、訓練モデルの作成と学習を行います。
流れとしては、
1.Encoder,Decoder,Seq2Seqのクラス作成
2.誤差を測定する評価関数を定義
3.これらを用いて、誤差が小さくなるようモデルにデータを学習させる
となります。
以下がコードになります。
qiita2.py
#Batchの設定を行います
batch_size = 32
train_iterator = torchtext.data.Iterator(
train_data,
batch_size=batch_size,
train=True
)
test_iterator = torchtext.data.Iterator(
test_data,
batch_size=batch_size,
train=False,
sort=False
)
#Encoderクラスを作成します#
class Encoder(nn.Module):
def __init__(self, n_h, n_vocab, n_emb, num_layers=1, bidirectional=False, dropout=0):
super().__init__()
self.n_h = n_h
self.num_layers = num_layers
self.bidirectional = bidirectional
self.dropout = dropout
self.embedding = nn.Embedding(n_vocab, n_emb)
self.embedding_dropout = nn.Dropout(self.dropout)
self.gru = nn.GRU(
input_size=n_emb, #入力サイズ
hidden_size=n_h,#隠れ層のサイズ（ニューロン数）
batch_first=True, #（バッチサイズ、時系列、入力数）
num_layers=num_layers,
bidirectional=bidirectional,
)
#順伝播の関数を定義
def forward(self, x):
index_pad = input_field.vocab.stoi["<pad>"] #"pad"のインデックス取得
sentence_lengths = x.size()[1] - (x == index_pad).sum(dim=1) #pad部分を引いて本来の文の長さを得る。
y = self.embedding(x) #埋め込みベクトル化
y = self.embedding_dropout(y)#過学習対策
y = nn.utils.rnn.pack_padded_sequence(
y,
sentence_lengths,
batch_first=True,
enforce_sorted=False
) #rnnに入れるためにpackedsequence型にする。
y, h = self.gru(y) #encoderでの出力と隠れ層の値を取得
y, _ = nn.utils.rnn.pad_packed_sequence(y, batch_first=True) #テンソルに戻す
if self.bidirectional:
#会話データにおける最後の重みが大きくなってしまうため
y = y[:, :, :self.n_h] + y[:, :, self.n_h:]#ここでは出力と隠れ層の値は双方向の時間の値を足したものになる。
h = h[:self.num_layers] + h[self.num_layers:]
return y,h
#Decoderクラスを作成します
class Decoder(nn.Module):
def __init__(self, n_h, n_out, n_vocab, n_emb, num_layers=1, dropout=0):
super().__init__()
self.n_h = n_h
self.n_out = n_out
self.num_layers = num_layers
self.dropout = dropout
self.embedding = nn.Embedding(n_vocab, n_emb)
self.embedding_dropout = nn.Dropout(self.dropout)
self.gru = nn.GRU(
input_size=n_emb,
hidden_size=n_h,
batch_first=True,
num_layers=num_layers,
)
self.fc = nn.Linear(n_h*2, self.n_out) #全結合層の導入
def forward(self, x, h_encoder, y_encoder):
y = self.embedding(x) # 単語をベクトルに変換
y = self.embedding_dropout(y)
y, h = self.gru(y, h_encoder) #ここでは出力と最後の時刻の隠れ層の値が渡される
# Attention
y_tr = torch.transpose(y, 1, 2) # 次元1と次元2を入れ替える
ed_mat = torch.bmm(y_encoder, y_tr) # バッチごとに行列積
attn_weight = F.softmax(ed_mat, dim=1) # attention weightの計算
attn_weight_tr = torch.transpose(attn_weight, 1, 2) # 次元1と次元2を入れ替える
context = torch.bmm(attn_weight_tr, y_encoder) # コンテキストベクトルの計算
y = torch.cat([y, context], dim=2) # 出力とコンテキストベクトルの合流
y = self.fc(y)
y = F.softmax(y, dim=2)
return y, h
#seq2seqクラスを作成します
class Seq2Seq(nn.Module):
def __init__(self, encoder, decoder, is_gpu=True):
super().__init__()
self.encoder = encoder
self.decoder = decoder
self.is_gpu = is_gpu
if self.is_gpu:
self.encoder.cuda()
self.decoder.cuda()
def forward(self, x_encoder, x_decoder): #順伝播メソッドの設定 訓練時につかう
if self.is_gpu:
x_encoder = x_encoder.cuda()
x_decoder = x_decoder.cuda()
batch_size = x_decoder.shape[0] #x_decoderは（バッチサイズ,時系列の数,ニューロン数）
n_time = x_decoder.shape[1]
y_encoder, h = self.encoder(x_encoder)
y_decoder = torch.zeros(batch_size, n_time, self.decoder.n_out) #バッチサイズ*時系列の数*出力数のすべて０のテンソル
if self.is_gpu:
y_decoder = y_decoder.cuda()
#教師強制と呼ばれる方法で各時刻の出力が次の入力に近くなるよう学習していきます。
for t in range(0, n_time): #各時刻で処理
x = x_decoder[:, t:t+1]
y, h = self.decoder(x, h, y_encoder)
y_decoder[:, t:t+1, :] = y #得られた出力ｙを各時刻ごとにy_encoderに格納
return y_decoder
def predict(self, x_encoder): #予測に使用 文章の生成、評価するときに使う
if self.is_gpu:
x_encoder = x_encoder.cuda()
batch_size = x_encoder.shape[0]
n_time = x_encoder.shape[1]
y_encoder, h = self.encoder(x_encoder)
y_decoder = torch.zeros(batch_size, n_time, dtype=torch.long) #ここではint型を用いるためtorch.long
if self.is_gpu:
y_decoder = y_decoder.cuda()
y = torch.ones(batch_size, 1, dtype=torch.long) * input_field.vocab.stoi["<sos>"] #予測用では各時刻で処理を行う必要がある。出力が次の入力になるから
for t in range(0, n_time):
x = y #ここでまえの出力を入力に
if self.is_gpu:
x = x.cuda()
y, h = self.decoder(x, h, y_encoder)
y = y.argmax(2) #yの中で最も大きい数値のインデックス得ることで最適な文章が生成
y_decoder[:, t:t+1] = y
return y_decoder
#評価関数を定義します
def evaluate(model, iterator):
model.eval() # 評価モードにできる
batch = next(iter(iterator))
x = batch.inp_text
y = model.predict(x)
for i in range(x.size()[0]):
inp_text = ""
for j in range(x.size()[1]):
word = input_field.vocab.itos[x[i][j]]
if word=="<pad>":
break
inp_text += word
rep_text = ""
for j in range(y.size()[1]):
word = reply_field.vocab.itos[y[i][j]]
if word=="<eos>":
break
rep_text += word
print("input:", inp_text)
print("reply:", rep_text)
print()
is_gpu = True # GPUを使用するかどうか
n_h = 800 #隠れ層のニューロン数
n_vocab_inp = len(input_field.vocab.itos) #入力文の長さ
n_vocab_rep = len(reply_field.vocab.itos) #応答文の長さ
n_emb = 300 #埋め込みベクトルの要素数
n_out = n_vocab_rep #出力の数
early_stop_patience = 5 # 早期終了のタイミング（誤差の最小値が何回更新されなかったら終了か）
num_layers = 1 #中間層の数
bidirectional = True
dropout = 0.1
clip = 100 #勾配の上限
# Seq2Seqのモデルを構築します
encoder = Encoder(n_h, n_vocab_inp, n_emb, num_layers, bidirectional, dropout=dropout)
decoder = Decoder(n_h, n_out, n_vocab_rep, n_emb, num_layers, dropout=dropout)
seq2seq = Seq2Seq(encoder, decoder, is_gpu=is_gpu)
# 誤差関数(今回は分類問題なのでクロスエントロピー誤差を使います)
loss_fnc = nn.CrossEntropyLoss(ignore_index=reply_field.vocab.stoi["<pad>"])
# 最適化アルゴリズム
optimizer_enc = optim.Adam(seq2seq.parameters(), lr=0.0001)
optimizer_dec = optim.Adam(seq2seq.parameters(), lr=0.0005)
record_loss_train = []
record_loss_test = []
min_losss_test = 0.0
# 学習
for i in range(1000): #1000epoch 学習
seq2seq.train()
loss_train = 0
for j, batch in enumerate(train_iterator):
inp, rep = batch.inp_text, batch.rep_text
x_enc = inp
x_dec = rep[:, :-1]
y_dec = seq2seq(x_enc, x_dec)
t_dec = rep[:, 1:]
t_dec = t_dec.cuda() if is_gpu else t_dec
loss = loss_fnc(
y_dec.view(-1, y_dec.size()[2]),
t_dec.reshape(-1)
)
loss_train += loss.item()
optimizer_enc.zero_grad()
optimizer_dec.zero_grad()
loss.backward()
nn.utils.clip_grad_norm_(encoder.parameters(), clip)
nn.utils.clip_grad_norm_(decoder.parameters(), clip)
optimizer_enc.step()
optimizer_dec.step()
if j%1000==0:
print("batch:", str(j)+"/"+str(len(train_data)//batch_size+1), "loss:", loss.item())
loss_train /= j+1
record_loss_train.append(loss_train)
# 評価モードにします
seq2seq.eval()
loss_test = 0
for j, batch in enumerate(test_iterator):
inp, rep = batch.inp_text, batch.rep_text
x_enc = inp
x_dec = torch.ones(rep.size(), dtype=torch.long) * reply_field.vocab.stoi["<sos>"]
x_dec[:, 1:] = rep[:, :-1]
y_dec = seq2seq(x_enc, x_dec)
t_dec = rep.cuda() if is_gpu else rep
loss = loss_fnc(
y_dec.view(-1, y_dec.size()[2]),
t_dec.view(-1)
)
loss_test += loss.item()
loss_test /= j+1
record_loss_test.append(loss_test)
if i%1 == 0:
print("Epoch:", i, "Loss_Train:", loss_train, "Loss_Test:", loss_test)
print()
evaluate(seq2seq, test_iterator)
#早期終了の設定をします
latest_min = min(record_loss_test[-(early_stop_patience):]) # 直近の最小値
if len(record_loss_test) >= early_stop_patience:
if latest_min > min_loss_test: # 直近で最小値が更新されていなければ
print("Early stopping!")
break
min_loss_test = latest_min
else:
min_loss_test = latest_min
torch.save(seq2seq.state_dict(), path2+"chat_bot.pth") #学習済みのモデルのステータスを保存
③学習済みモデルを使ってユーザーの入力に対し応答を予測し、出力
最後に、ユーザーの入力を解析し、応答を出力するといった簡易的なチャットボットをノートブック上で動かします。
今回はユーザーが"さようなら"と入力するまで会話を続けることにしました。
以下が、コードになります。
qiita3.py
#モデルの読み込み
import dill
path2 = "/content/drive/My Drive/対話コーパス 前処理/NTT data/"
input_field = torch.load(path2+"input_field.pkl", pickle_module=dill) #インデックスと単語の対応付け
reply_field = torch.load(path2+"reply_field.pkl", pickle_module=dill)
is_gpu = True
n_h = 800
n_vocab_inp = len(input_field.vocab.itos)
n_vocab_rep = len(reply_field.vocab.itos)
n_emb = 300
n_out = n_vocab_rep
early_stop_patience = 5 # 早期終了のタイミング（誤差の最小値が何回更新されなかったら終了か）
num_layers = 1
bidirectional = True
dropout = 0.0
clip = 100
encoder = Encoder(n_h, n_vocab_inp, n_emb, num_layers, bidirectional)
decoder = Decoder(n_h, n_out, n_vocab_rep, n_emb, num_layers, dropout=dropout)
seq2seq = Seq2Seq(encoder, decoder, is_gpu=is_gpu)
seq2seq.load_state_dict(torch.load(path2+"chat_bot.pth", map_location=torch.device("cpu"))) #パラメータ読み込んでCPU対応
janome_tk = Tokenizer()
#入力に対して応答文生成する関数 定義
def reply(inp_text, tokenizer, max_length=10):
wakati_list = [token for token in janome_tk.tokenize(inp_text, wakati=True)]
word_index = []
for word in wakati_list:
index = input_field.vocab.stoi[word]
word_index.append(index)
x = torch.tensor(word_index) #テンソル型に変換する
x = x.view(1, -1) #バッチサイズの１に調整
y = seq2seq.predict(x, max_length) #予測文がインデックスで出力
reply_text = ""
for j in range(y.size()[1]):
word = reply_field.vocab.itos[y[0][j]]#インデックスを単語に変換
if word=="<eos>":
break
reply_text += word
reply_text = reply_text.replace("<sos>", "") #不要なものを削除
reply_text = reply_text.replace("<eos>", "")
reply_text = reply_text.replace("<pad>", "")
reply_text = reply_text.replace("<unk>", "")
return reply_text
#ノートブック上でAIと会話してみる
bot_name = "chat AI bot"
your_name = input("あなたの名前を教えてください:")
print()
print(bot_name + ": " + "こんにちは！" + your_name + "さん" )
message = ""
while message != "さようなら":
message = input(your_name+ ": ")
response = reply(message, janome_tk
, max_length=10)
print(bot_name + ": " + response)
print(bot_name + ": 会話が出来て嬉しかったです！")
5.結果
結果は以下のような感じになりました。
result.py
あなたの名前を教えてください:mrtts0622
chat AI bot: こんにちは！mrtts0622さん
mrtts0622: こんにちは
chat AI bot: こんにちはこんにちは
mrtts0622: かなり暖かくなってきましたね
chat AI bot:
mrtts0622: お元気ですか？
chat AI bot: 今日はいいですね
mrtts0622: 夏の風物詩といえば？
chat AI bot: 海水浴スイカも好きです
mrtts0622: 少しおなかがすきました
chat AI bot: 海水浴か？
mrtts0622: いや、なんでやねん
chat AI bot: が好きです
mrtts0622: さようなら
chat AI bot:
chat AI bot: 会話が出来て嬉しかったです！
話し手が相手の意図を読み取ることで何とか会話が成立していますね。さすがにツッコミにまでは対応できなかったようです( ´∀｀ )
また、会話のデータに偏りがあったのか夏の話題には強いようで会話の節々に夏好きがあふれている気がします。
6.まとめ
PyTorchによるチャットボットの作成はかなり複雑で一行一行なにを行っているか理解するのにかなり時間がかかってしまいました。
また、精度を上げることもなかなか骨が折れる作業だったのでやはりLINEやGoogleはとても偉大であるということが実感できる制作でもありました。
今後の課題としては、
・Bertをつかって前後の文脈も考慮したより自然なボットを作る。
・twitterなどで、より自然でかつ大量の会話データを用意する。
などが挙げられそうです。
また、僕自身の課題として来月はKaggleにも挑戦しようと思っています。
最後までお付き合いいただきありがとうございました！
参考にさせてもらった文献
雑談会話コーパス-対話破綻検出チャレンジhttps://sites.google.com/site/dialoguebreakdowndetection/chat-dialogue-corpus
○ 雑談対話APIサイト
https://www.nttdocomo.co.jp/service/developer/smart_phone/analysis/chat/
udemyの講座：人工知能(AI)を搭載したTwitterボットを作ろう
https://www.udemy.com/
PyTorchチュートリアル(日本語翻訳版)
https://yutaroogawa.github.io/pytorch_tutorials_jp/