#1. 再帰型ニューラルネットワークの概念
##1.1. 要点まとめ
再帰型ニューラルネットワーク（RNN）はテキストデータや音声データ、株価などの時系列データをうまく扱うことができるニューラルネットワークである。通常のニューラルネットワークで扱えなかった系列間の依存関係をRNNでは扱うことができる。
時刻$t$での入力層の出力を$x^t$、中間層の入出力を$u^t$、$z^t$、出力層の入出力を$v^t$、$y^t$、入力層と中間層の重みを$W_{(in)}$、中間層と出力層の重みを$W_{(out)}$、中間層間の重みを$W$、中間層の活性化関数を$f$、出力層の活性化関数を$g$、$b$と$c$をそれぞれバイアスとすると、RNNは数学的に以下の形で表現される。
u^t = W_{(in)}x^t + W z^{t-1} + b,\ z^t = f(u^t),\\
v^t = W_{(out)}z^t + c,\ y^t = g(v^t)
BPTTはRNNにおけるパラメータ調整法の１つで誤差逆伝播の１種である。数学的には以下のように表現される。
\frac{\partial E}{\partial W_{(in)}}=\frac{\partial E}{\partial u^t}\left[\frac{\partial u^t}{\partial W_{(in)}}\right]^T=\delta^t[x^t]^T,\\
\frac{\partial E}{\partial W_{(out)}}=\frac{\partial E}{\partial v^t}\left[\frac{\partial v^t}{\partial W_{(out)}}\right]^T=\delta^{out,t}[z^t]^T,\\
\frac{\partial E}{\partial W}=\frac{\partial E}{\partial u^t}\left[\frac{\partial u^t}{\partial W}\right]^T=\delta^t[z^{t-1}]^T,\\
\frac{\partial E}{\partial b}=\frac{\partial E}{\partial u^t}\frac{\partial u^t}{\partial b} = \delta^t,\\
\frac{\partial E}{\partial c}=\frac{\partial E}{\partial v^t}\frac{\partial v^t}{\partial c} = \delta^{out,t}
上記を用いると、パラメータ更新式は以下のように表現される。
W^{t+1}_{(in)}=W^{t}_{(in)}-\epsilon\frac{\partial E}{\partial W_{(in)}}=W^t_{(in)}-\epsilon\sum_{z=0}^{T_t}\delta^{t-z}[x^{t-z}]^T,\\
W^{t+1}_{(out)}=W^{t}_{(out)}-\epsilon\frac{\partial E}{\partial W_{(in)}}=W^t_{(out)}-\epsilon\delta^{out,t}[z^{t}]^T,\\
W^{t+1}=W^{t}-\epsilon\frac{\partial E}{\partial W}=W^t_{(in)}-\epsilon\sum_{z=0}^{T_t}\delta^{t-z}[z^{t-z-1}]^T,\\
b^{t+1}=b^t−\epsilon \frac{\partial E}{\partial b}=b^t−\epsilon \sum_{z=0}^{T_t} \delta^{t−z},\\
c^{t+1}=c^t−\epsilon\frac{\partial E}{\partial c} =c^t−\epsilon \delta^{out,t}
##1.2. 実装演習
import numpy as np
from common import functions
import matplotlib.pyplot as plt
def d_tanh(x):
return 1/(np.cosh(x) ** 2)
# データを用意
# 2進数の桁数
binary_dim = 8
# 最大値 + 1
largest_number = pow(2, binary_dim)
# largest_numberまで2進数を用意
binary = np.unpackbits(np.array([range(largest_number)],dtype=np.uint8).T,axis=1)
input_layer_size = 2
hidden_layer_size = 16
output_layer_size = 1
weight_init_std = 1
learning_rate = 0.1
iters_num = 10000
plot_interval = 100
# ウェイト初期化 (バイアスは簡単のため省略)
W_in = weight_init_std * np.random.randn(input_layer_size, hidden_layer_size)
W_out = weight_init_std * np.random.randn(hidden_layer_size, output_layer_size)
W = weight_init_std * np.random.randn(hidden_layer_size, hidden_layer_size)
# Xavier
# W_in = np.random.randn(input_layer_size, hidden_layer_size) / (np.sqrt(input_layer_size))
# W_out = np.random.randn(hidden_layer_size, output_layer_size) / (np.sqrt(hidden_layer_size))
# W = np.random.randn(hidden_layer_size, hidden_layer_size) / (np.sqrt(hidden_layer_size))
# He
# W_in = np.random.randn(input_layer_size, hidden_layer_size) / (np.sqrt(input_layer_size)) * np.sqrt(2)
# W_out = np.random.randn(hidden_layer_size, output_layer_size) / (np.sqrt(hidden_layer_size)) * np.sqrt(2)
# W = np.random.randn(hidden_layer_size, hidden_layer_size) / (np.sqrt(hidden_layer_size)) * np.sqrt(2)
# 勾配
W_in_grad = np.zeros_like(W_in)
W_out_grad = np.zeros_like(W_out)
W_grad = np.zeros_like(W)
u = np.zeros((hidden_layer_size, binary_dim + 1))
z = np.zeros((hidden_layer_size, binary_dim + 1))
y = np.zeros((output_layer_size, binary_dim))
delta_out = np.zeros((output_layer_size, binary_dim))
delta = np.zeros((hidden_layer_size, binary_dim + 1))
all_losses = []
for i in range(iters_num):
# A, B初期化 (a + b = d)
a_int = np.random.randint(largest_number/2)
a_bin = binary[a_int] # binary encoding
b_int = np.random.randint(largest_number/2)
b_bin = binary[b_int] # binary encoding
# 正解データ
d_int = a_int + b_int
d_bin = binary[d_int]
# 出力バイナリ
out_bin = np.zeros_like(d_bin)
# 時系列全体の誤差
all_loss = 0
# 時系列ループ
for t in range(binary_dim):
# 入力値
X = np.array([a_bin[ - t - 1], b_bin[ - t - 1]]).reshape(1, -1)
# 時刻tにおける正解データ
dd = np.array([d_bin[binary_dim - t - 1]])
u[:,t+1] = np.dot(X, W_in) + np.dot(z[:,t].reshape(1, -1), W)
z[:,t+1] = functions.sigmoid(u[:,t+1])
# z[:,t+1] = functions.relu(u[:,t+1])
# z[:,t+1] = np.tanh(u[:,t+1])
y[:,t] = functions.sigmoid(np.dot(z[:,t+1].reshape(1, -1), W_out))
#誤差
loss = functions.mean_squared_error(dd, y[:,t])
delta_out[:,t] = functions.d_mean_squared_error(dd, y[:,t]) * functions.d_sigmoid(y[:,t])
all_loss += loss
out_bin[binary_dim - t - 1] = np.round(y[:,t])
for t in range(binary_dim)[::-1]:
X = np.array([a_bin[-t-1],b_bin[-t-1]]).reshape(1, -1)
delta[:,t] = (np.dot(delta[:,t+1].T, W.T) + np.dot(delta_out[:,t].T, W_out.T)) * functions.d_sigmoid(u[:,t+1])
# delta[:,t] = (np.dot(delta[:,t+1].T, W.T) + np.dot(delta_out[:,t].T, W_out.T)) * functions.d_relu(u[:,t+1])
# delta[:,t] = (np.dot(delta[:,t+1].T, W.T) + np.dot(delta_out[:,t].T, W_out.T)) * d_tanh(u[:,t+1])
# 勾配更新
W_out_grad += np.dot(z[:,t+1].reshape(-1,1), delta_out[:,t].reshape(-1,1))
W_grad += np.dot(z[:,t].reshape(-1,1), delta[:,t].reshape(1,-1))
W_in_grad += np.dot(X.T, delta[:,t].reshape(1,-1))
# 勾配適用
W_in -= learning_rate * W_in_grad
W_out -= learning_rate * W_out_grad
W -= learning_rate * W_grad
W_in_grad *= 0
W_out_grad *= 0
W_grad *= 0
if(i % plot_interval == 0):
all_losses.append(all_loss)
print("iters:" + str(i))
print("Loss:" + str(all_loss))
print("Pred:" + str(out_bin))
print("True:" + str(d_bin))
out_int = 0
for index,x in enumerate(reversed(out_bin)):
out_int += x * pow(2, index)
print(str(a_int) + " + " + str(b_int) + " = " + str(out_int))
print("------------")
lists = range(0, iters_num, plot_interval)
plt.plot(lists, all_losses, label="loss")
plt.show()
以下の結果のように学習が進むにつれて、実際の値とRNNの予測結果が一致してくることがわかる。
iters:0
Loss:1.6938859799543602
Pred:[0 0 0 0 0 0 0 0]
True:[0 1 1 1 0 1 1 1]
12 + 107 = 0
------------
iters:100
Loss:1.0713609184549415
Pred:[1 1 1 1 1 1 0 1]
True:[1 0 0 1 1 0 0 1]
63 + 90 = 253
------------
iters:200
Loss:1.1100796140193345
Pred:[1 1 1 1 1 0 1 1]
True:[0 1 0 0 1 0 1 0]
14 + 60 = 251
------------
iters:300
Loss:0.9377709546625936
Pred:[0 0 0 1 1 0 0 1]
True:[1 0 0 0 1 1 0 1]
47 + 94 = 25
------------
iters:400
Loss:0.8738102608993707
Pred:[1 0 0 0 1 0 1 1]
True:[1 0 0 1 1 0 1 0]
69 + 85 = 139
------------
・
・
・
------------
iters:9500
Loss:0.002346530844460706
Pred:[0 1 1 0 1 0 1 0]
True:[0 1 1 0 1 0 1 0]
102 + 4 = 106
------------
iters:9600
Loss:0.0012131543058348268
Pred:[1 0 0 1 1 1 1 1]
True:[1 0 0 1 1 1 1 1]
75 + 84 = 159
------------
iters:9700
Loss:0.0007672512819978022
Pred:[0 1 0 1 0 0 1 1]
True:[0 1 0 1 0 0 1 1]
3 + 80 = 83
------------
iters:9800
Loss:0.002455167079922689
Pred:[0 1 1 1 1 1 1 0]
True:[0 1 1 1 1 1 1 0]
70 + 56 = 126
------------
iters:9900
Loss:0.003017986334835035
Pred:[1 1 0 0 1 0 0 0]
True:[1 1 0 0 1 0 0 0]
102 + 98 = 200
------------
##1.3. 確認テスト
RNNのネットワークには大きくわけて3つの重みがある。1つは入力から現在の中間層を定義する際にかけられる重み、1つは中間層から出力を定義する際にかけられる重みである。残り1つの重みについて説明せよ。
時刻$t-1$時点の中間層の出力を$t$時点の中間層へ入力する際にかけられる重み
$s_1$に至るまでの活性化関数を$f$とすると、以下のようになる。
s_1 = f(W_{(in)}x_1 + Ws_0 + b),\\
y_1 = g(W_{(out)}s_1 + c)
#2. LSTM
##2.1. 要点まとめ
RNNは時間軸を考えるため、長期の時間依存性を考慮したモデルを作るとなると、ネットワークの規模が大きくなるため、勾配消失（爆発）問題が起きる問題を抱えており、構造を変えてその問題を解決したのがLSTMとなる。
-
CEC
CECは記憶機能を持ち、勾配が1であれば勾配消失（爆発）問題が起きないため、勾配を1にするために導入された。しかしCECでは記憶のみに特化しているため、時間依存度に関係なく重みが一律でニューラルネットワークの学習特性が無いという課題がある。そのため、CECのまわりに学習機能を持たせる。（何を覚えるか、記憶しているもののうち何を使うか） -
入力ゲートと出力ゲート
入力ゲートは入力をどのくらいの重みで記憶を保存するか制御し、出力ゲートは出力する際にどう記憶を使えばよいか制御を行う。入力ゲート・出力ゲートを追加することで、学習機能を持たせることができ、CECの課題を解決している。 -
忘却ゲート
CECは過去の情報が全て保管されているため、過去の情報が要らなくなった場合、削除することはできず、保管され続けるという課題がある。過去の情報が要らなくなった場合、そのタイミングで情報を忘却する機能が必要で、その役割を担うのが忘却ゲートとなる。 -
覗き穴ゲート
入力ゲート・出力ゲートは判断材料として今回の入力と前回の出力を見ていたが、ついでにCECの今の記憶も見てみようというのが覗き穴ゲートの考え方となるが、あまり効果がなかったことが知られている。
##2.2. 確認テスト
以下の文章をLSTMに入力し空欄に当てはまる単語を予測したいとする。文中の「とても」という言葉は空欄の予測においてなくなっても影響を及ぼさないと考えられる。このような場合、どのゲートが作用すると考えられるか。
「映画おもしろかったね。ところで、とてもお腹が空いたから何か____。」
⇒忘却ゲート
#3. GRU
##3.1. 要点まとめ
LSTMでは、パラメータ数が多く、計算負荷が高くなる問題があったため、それを改善したGRUが考え出された。GRUではそのパラメータを大幅に削減し、精度は同等またはそれ以上が望める様になった構造となる。GRUのメリットとしては計算負荷が低いことにある。
##3.2. 実装演習
これまではnumpyを使って実装してきたが、以下は深層学習のフレームワークであるtensorflowを用いて実装した例となる。
import tensorflow as tf
import numpy as np
import re
import glob
import collections
import random
import pickle
import time
import datetime
import os
# logging levelを変更
tf.logging.set_verbosity(tf.logging.ERROR)
class Corpus:
def __init__(self):
self.unknown_word_symbol = "<???>" # 出現回数の少ない単語は未知語として定義しておく
self.unknown_word_threshold = 3 # 未知語と定義する単語の出現回数の閾値
self.corpus_file = "./corpus/**/*.txt"
self.corpus_encoding = "utf-8"
self.dictionary_filename = "./data_for_predict/word_dict.dic"
self.chunk_size = 5
self.load_dict()
words = []
for filename in glob.glob(self.corpus_file, recursive=True):
with open(filename, "r", encoding=self.corpus_encoding) as f:
# word breaking
text = f.read()
# 全ての文字を小文字に統一し、改行をスペースに変換
text = text.lower().replace("\n", " ")
# 特定の文字以外の文字を空文字に置換する
text = re.sub(r"[^a-z '\-]", "", text)
# 複数のスペースはスペース一文字に変換
text = re.sub(r"[ ]+", " ", text)
# 前処理： '-' で始まる単語は無視する
words = [ word for word in text.split() if not word.startswith("-")]
self.data_n = len(words) - self.chunk_size
self.data = self.seq_to_matrix(words)
def prepare_data(self):
"""
訓練データとテストデータを準備する。
data_n = ( text データの総単語数 ) - chunk_size
input: (data_n, chunk_size, vocabulary_size)
output: (data_n, vocabulary_size)
"""
# 入力と出力の次元テンソルを準備
all_input = np.zeros([self.chunk_size, self.vocabulary_size, self.data_n])
all_output = np.zeros([self.vocabulary_size, self.data_n])
# 準備したテンソルに、コーパスの one-hot 表現(self.data) のデータを埋めていく
# i 番目から ( i + chunk_size - 1 ) 番目までの単語が１組の入力となる
# このときの出力は ( i + chunk_size ) 番目の単語
for i in range(self.data_n):
all_output[:, i] = self.data[:, i + self.chunk_size] # (i + chunk_size) 番目の単語の one-hot ベクトル
for j in range(self.chunk_size):
all_input[j, :, i] = self.data[:, i + self.chunk_size - j - 1]
# 後に使うデータ形式に合わせるために転置を取る
all_input = all_input.transpose([2, 0, 1])
all_output = all_output.transpose()
# 訓練データ：テストデータを 4 : 1 に分割する
training_num = ( self.data_n * 4 ) // 5
return all_input[:training_num], all_output[:training_num], all_input[training_num:], all_output[training_num:]
def build_dict(self):
# コーパス全体を見て、単語の出現回数をカウントする
counter = collections.Counter()
for filename in glob.glob(self.corpus_file, recursive=True):
with open(filename, "r", encoding=self.corpus_encoding) as f:
# word breaking
text = f.read()
# 全ての文字を小文字に統一し、改行をスペースに変換
text = text.lower().replace("\n", " ")
# 特定の文字以外の文字を空文字に置換する
text = re.sub(r"[^a-z '\-]", "", text)
# 複数のスペースはスペース一文字に変換
text = re.sub(r"[ ]+", " ", text)
# 前処理： '-' で始まる単語は無視する
words = [word for word in text.split() if not word.startswith("-")]
counter.update(words)
# 出現頻度の低い単語を一つの記号にまとめる
word_id = 0
dictionary = {}
for word, count in counter.items():
if count <= self.unknown_word_threshold:
continue
dictionary[word] = word_id
word_id += 1
dictionary[self.unknown_word_symbol] = word_id
print("総単語数：", len(dictionary))
# 辞書を pickle を使って保存しておく
with open(self.dictionary_filename, "wb") as f:
pickle.dump(dictionary, f)
print("Dictionary is saved to", self.dictionary_filename)
self.dictionary = dictionary
print(self.dictionary)
def load_dict(self):
with open(self.dictionary_filename, "rb") as f:
self.dictionary = pickle.load(f)
self.vocabulary_size = len(self.dictionary)
self.input_layer_size = len(self.dictionary)
self.output_layer_size = len(self.dictionary)
print("総単語数: ", self.input_layer_size)
def get_word_id(self, word):
# print(word)
# print(self.dictionary)
# print(self.unknown_word_symbol)
# print(self.dictionary[self.unknown_word_symbol])
# print(self.dictionary.get(word, self.dictionary[self.unknown_word_symbol]))
return self.dictionary.get(word, self.dictionary[self.unknown_word_symbol])
# 入力された単語を one-hot ベクトルにする
def to_one_hot(self, word):
index = self.get_word_id(word)
data = np.zeros(self.vocabulary_size)
data[index] = 1
return data
def seq_to_matrix(self, seq):
print(seq)
data = np.array([self.to_one_hot(word) for word in seq]) # (data_n, vocabulary_size)
return data.transpose() # (vocabulary_size, data_n)
class Language:
"""
input layer: self.vocabulary_size
hidden layer: rnn_size = 30
output layer: self.vocabulary_size
"""
def __init__(self):
self.corpus = Corpus()
self.dictionary = self.corpus.dictionary
self.vocabulary_size = len(self.dictionary) # 単語数
self.input_layer_size = self.vocabulary_size # 入力層の数
self.hidden_layer_size = 30 # 隠れ層の RNN ユニットの数
self.output_layer_size = self.vocabulary_size # 出力層の数
self.batch_size = 128 # バッチサイズ
self.chunk_size = 5 # 展開するシーケンスの数。c_0, c_1, ..., c_(chunk_size - 1) を入力し、c_(chunk_size) 番目の単語の確率が出力される。
self.learning_rate = 0.005 # 学習率
self.epochs = 1000 # 学習するエポック数
self.forget_bias = 1.0 # LSTM における忘却ゲートのバイアス
self.model_filename = "./data_for_predict/predict_model.ckpt"
self.unknown_word_symbol = self.corpus.unknown_word_symbol
def inference(self, input_data, initial_state):
"""
:param input_data: (batch_size, chunk_size, vocabulary_size) 次元のテンソル
:param initial_state: (batch_size, hidden_layer_size) 次元の行列
:return:
"""
# 重みとバイアスの初期化
hidden_w = tf.Variable(tf.truncated_normal([self.input_layer_size, self.hidden_layer_size], stddev=0.01))
hidden_b = tf.Variable(tf.ones([self.hidden_layer_size]))
output_w = tf.Variable(tf.truncated_normal([self.hidden_layer_size, self.output_layer_size], stddev=0.01))
output_b = tf.Variable(tf.ones([self.output_layer_size]))
# BasicLSTMCell, BasicRNNCell は (batch_size, hidden_layer_size) が chunk_size 数ぶんつながったリストを入力とする。
# 現時点での入力データは (batch_size, chunk_size, input_layer_size) という３次元のテンソルなので
# tf.transpose や tf.reshape などを駆使してテンソルのサイズを調整する。
input_data = tf.transpose(input_data, [1, 0, 2]) # 転置。(chunk_size, batch_size, vocabulary_size)
input_data = tf.reshape(input_data, [-1, self.input_layer_size]) # 変形。(chunk_size * batch_size, input_layer_size)
input_data = tf.matmul(input_data, hidden_w) + hidden_b # 重みWとバイアスBを適用。 (chunk_size, batch_size, hidden_layer_size)
input_data = tf.split(input_data, self.chunk_size, 0) # リストに分割。chunk_size * (batch_size, hidden_layer_size)
# RNN のセルを定義する。RNN Cell の他に LSTM のセルや GRU のセルなどが利用できる。
cell = tf.nn.rnn_cell.BasicRNNCell(self.hidden_layer_size)
outputs, states = tf.nn.static_rnn(cell, input_data, initial_state=initial_state)
# 最後に隠れ層から出力層につながる重みとバイアスを処理する
# 最終的に softmax 関数で処理し、確率として解釈される。
# softmax 関数はこの関数の外で定義する。
output = tf.matmul(outputs[-1], output_w) + output_b
return output
def loss(self, logits, labels):
cost = tf.reduce_mean(tf.nn.softmax_cross_entropy_with_logits(logits=logits, labels=labels))
return cost
def training(self, cost):
# 今回は最適化手法として Adam を選択する。
# ここの AdamOptimizer の部分を変えることで、Adagrad, Adadelta などの他の最適化手法を選択することができる
optimizer = tf.train.AdamOptimizer(learning_rate=self.learning_rate).minimize(cost)
return optimizer
def train(self):
# 変数などの用意
input_data = tf.placeholder("float", [None, self.chunk_size, self.input_layer_size])
actual_labels = tf.placeholder("float", [None, self.output_layer_size])
initial_state = tf.placeholder("float", [None, self.hidden_layer_size])
prediction = self.inference(input_data, initial_state)
cost = self.loss(prediction, actual_labels)
optimizer = self.training(cost)
correct = tf.equal(tf.argmax(prediction, 1), tf.argmax(actual_labels, 1))
accuracy = tf.reduce_mean(tf.cast(correct, tf.float32))
# TensorBoard で可視化するため、クロスエントロピーをサマリーに追加
tf.summary.scalar("Cross entropy: ", cost)
summary = tf.summary.merge_all()
# 訓練・テストデータの用意
# corpus = Corpus()
trX, trY, teX, teY = self.corpus.prepare_data()
training_num = trX.shape[0]
# ログを保存するためのディレクトリ
timestamp = time.time()
dirname = datetime.datetime.fromtimestamp(timestamp).strftime("%Y%m%d%H%M%S")
# ここから実際に学習を走らせる
with tf.Session() as sess:
sess.run(tf.global_variables_initializer())
summary_writer = tf.summary.FileWriter("./log/" + dirname, sess.graph)
# エポックを回す
for epoch in range(self.epochs):
step = 0
epoch_loss = 0
epoch_acc = 0
# 訓練データをバッチサイズごとに分けて学習させる (= optimizer を走らせる)
# エポックごとの損失関数の合計値や（訓練データに対する）精度も計算しておく
while (step + 1) * self.batch_size < training_num:
start_idx = step * self.batch_size
end_idx = (step + 1) * self.batch_size
batch_xs = trX[start_idx:end_idx, :, :]
batch_ys = trY[start_idx:end_idx, :]
_, c, a = sess.run([optimizer, cost, accuracy],
feed_dict={input_data: batch_xs,
actual_labels: batch_ys,
initial_state: np.zeros([self.batch_size, self.hidden_layer_size])
}
)
epoch_loss += c
epoch_acc += a
step += 1
# コンソールに損失関数の値や精度を出力しておく
print("Epoch", epoch, "completed ouf of", self.epochs, "-- loss:", epoch_loss, " -- accuracy:",
epoch_acc / step)
# Epochが終わるごとにTensorBoard用に値を保存
summary_str = sess.run(summary, feed_dict={input_data: trX,
actual_labels: trY,
initial_state: np.zeros(
[trX.shape[0],
self.hidden_layer_size]
)
}
)
summary_writer.add_summary(summary_str, epoch)
summary_writer.flush()
# 学習したモデルも保存しておく
saver = tf.train.Saver()
saver.save(sess, self.model_filename)
# 最後にテストデータでの精度を計算して表示する
a = sess.run(accuracy, feed_dict={input_data: teX, actual_labels: teY,
initial_state: np.zeros([teX.shape[0], self.hidden_layer_size])})
print("Accuracy on test:", a)
def predict(self, seq):
"""
文章を入力したときに次に来る単語を予測する
:param seq: 予測したい単語の直前の文字列。chunk_size 以上の単語数が必要。
:return:
"""
# 最初に復元したい変数をすべて定義してしまいます
tf.reset_default_graph()
input_data = tf.placeholder("float", [None, self.chunk_size, self.input_layer_size])
initial_state = tf.placeholder("float", [None, self.hidden_layer_size])
prediction = tf.nn.softmax(self.inference(input_data, initial_state))
predicted_labels = tf.argmax(prediction, 1)
# 入力データの作成
# seq を one-hot 表現に変換する。
words = [word for word in seq.split() if not word.startswith("-")]
x = np.zeros([1, self.chunk_size, self.input_layer_size])
for i in range(self.chunk_size):
word = seq[len(words) - self.chunk_size + i]
index = self.dictionary.get(word, self.dictionary[self.unknown_word_symbol])
x[0][i][index] = 1
feed_dict = {
input_data: x, # (1, chunk_size, vocabulary_size)
initial_state: np.zeros([1, self.hidden_layer_size])
}
# tf.Session()を用意
with tf.Session() as sess:
# 保存したモデルをロードする。ロード前にすべての変数を用意しておく必要がある。
saver = tf.train.Saver()
saver.restore(sess, self.model_filename)
# ロードしたモデルを使って予測結果を計算
u, v = sess.run([prediction, predicted_labels], feed_dict=feed_dict)
keys = list(self.dictionary.keys())
# コンソールに文字ごとの確率を表示
for i in range(self.vocabulary_size):
c = self.unknown_word_symbol if i == (self.vocabulary_size - 1) else keys[i]
print(c, ":", u[0][i])
print("Prediction:", seq + " " + ("<???>" if v[0] == (self.vocabulary_size - 1) else keys[v[0]]))
return u[0]
def build_dict():
cp = Corpus()
cp.build_dict()
if __name__ == "__main__":
#build_dict()
ln = Language()
# 学習するときに呼び出す
#ln.train()
# 保存したモデルを使って単語の予測をする
ln.predict("some of them looks like")
??> : 2.919565e-05ストリーミング出力は最後の 5000 行に切り捨てられました。
beating : 1.3192031e-14
authentic : 1.4901815e-14
glow : 1.5293715e-14
oy : 1.4680216e-14
emotion : 1.4983698e-14
delight : 1.40041135e-14
nuclear : 1.4859978e-14
dropped : 1.4963249e-14
hiroshima : 1.3928354e-14
beings : 1.58225e-14
tens : 1.5364736e-14
burned : 1.3809686e-14
homeless : 1.564374e-14
albert : 1.4478778e-14
・
・
・
bmi : 3.1721086e-07
covariates : 2.834505e-06
yhl : 2.2330451e-07
yol : 9.9930106e-11
obesity : 1.3501609e-09
evgfp : 6.1830234e-09
unintended : 4.67851e-09
sizes : 2.5699424e-07
obese : 1.9368164e-07
Prediction: some of them looks like et
##3.3. 確認テスト
LSTMとCECが抱える課題について、それぞれ簡潔に述べよ。
LSTM：パラメータ数が多く、計算負荷が高い
CEC：時間依存度に関係なく重みが一律でニューラルネットワークの学習特性が無い
LSTMとGRUの違いを簡潔に述べよ。
LSTMは入力ゲート・出力ゲート・忘却ゲートの３つゲートとCECがある。GRUはCECがなく、更新ゲートとリセットゲートを持つ。GRUのほうがパラメータが少ないため、LSTMよりGRUは計算量が少ない。
#4. 双方向RNN
##4.1. 要点まとめ
双方向RNNは、過去の情報だけを利用して予測するだけでなく、未来の情報を利用することで、予測性能を向上させるためのモデルとなる。過去から未来への順伝播と未来から過去への逆伝播を組み合わせたものとなる。文章の推敲や、機械翻訳などで扱われる。
##4.2. 確認テスト
順方向と逆方向に伝播したときの中間層表現をあわせたものが特徴量となるので4となる。
#5. Seq2Seq
##5.1. 要点まとめ
Seq2Seqは時系列データを入力して時系列データを出力モデルで、入力を取るほうであるエンコーダーRNNと出力をするデコーダーRNNが構成される。エンコーダーRNNは単語の列からFinal Stateと呼ばれる文脈ベクトルを生成し、それをもとにデコーダーRNNは別の文脈を作り出そうとする。このSeq2Seqだが、１つの文に対して１つ回答しかできないという課題があったが、それを解決するために文脈に沿った意味ベクトルを作成することを目的に文脈自体も１つの単語のように扱えるように文の前の状態を次の状態に引き継ぐRNNのような構造を持たせたHREDが発明された。しかし複雑な構造を持ったが故にHREDは文脈に対して当たり障りのない回答しかしない課題が発生した。そこでオートエンコーダーの改良版であるVAEの考え方を取り入れたVHREDが考えられ、当たり障りのない回答以上の回答をするように改良された。ここで、オートエンコーダーは入力と出力が同じようになるように学習を進めることで、エンコーダーは元の画像に戻りやすいように潜在変数zを作り出し、デコーダーは作られたデータから元の画像に戻すという処理を学習する。VAEではこの潜在変数を作り出す際に正則化を行う。これを行うことで、元のデータが似ていれば潜在変数が同じような値になるように学習を進めることができる。
##5.2. 確認テスト
seq2seqの説明は２である。
seq2seq：1文の1問1答に対して処理ができるある時系列データからある時系列データを作り出すネットワーク
HRED：seq2seqの機構に対してそれまでの文脈を解釈に加えられるようにすることで、文脈の意味をくみ取った文の変換をできるようにしたもの
VHRED：HREDが文脈に対して当たり障りのない回答しかしなくなったことに対する解決策でオートエンコーダーであるVAEの考え方を取り入れて、当たり障りのない回答以上の回答をするように改良されたモデル
#6. Word2Vec
##6.1. 要点まとめ
Word2Vecはニューラルネットワークを通して、単語の分散表現を獲得するための手法となる。単語をone-hotベクトルで表現すると、次元数が多くメモリを大量に消費したり、似た単語の類似度を測ることができなかったりするなどの問題があるが、Word2Vecにより、より少ない次元数で単語の意味を数値で表現でき、現実的な計算量で省メモリ化を実現することができた。RNN系のモデルでは通常、Word2Vecにより単語を分散表現に変換したものを入力として扱う。
#7. Attention Mechanism
##7.1. 要点まとめ
seq2seqでは中間層の大きさが固定であるため、固定化された大きさの中で文の意味を表現することになるが、入力される単語数が多くなるほど、その中に意味を詰め込む必要が出てきてうまく文の意味を表現できなくなる。そこでそれを解決するために開発されたのがAttention Mechanismである。1文の中で重要な単語を自力で見つけられるようにする仕組みとなる。
##7.2. 確認テスト
RNNとword2vec、seq2seqとAttentionの違いを簡潔に述べよ。
RNN：時系列データを処理するのに適したネットワーク
Word2Vec：単語の分散表現を得る手法
seq2seq：一つの時系列データから別の時系列データを得るネットワーク
Attention：時系列データの中身に対して関連性の重みをつける手法
#8. 強化学習
強化学習は長期的に報酬を最大化できるように環境のなかで行動を選択できるエージェントを作ることを目標とする機械学習の一分野である。エージェントが環境に対して行動を起こし、環境が状態の更新を行い、状態と報酬をエージェントに与え、それをもとにエージェントは方策の更新を行い、また環境に対して行動を起こすということを繰り返していく。環境について事前に完璧な知識があれば、最適な行動を予測し決定することは可能だが、それは通常ありえないので、不完全な知識を元に行動しながら、データを収集し、最適な行動を見つけていく。
強化学習では方策関数と行動価値関数を学習させる。強化学習は計算量の問題で注目されていなかったが、学習を進める手法として、Q学習と関数近似法を組み合わせ手法が登場し、注目されるようになった。Q学習は行動価値関数を、行動する毎に更新することにより学習を進める方法で、関数近似法は価値関数や方策関数を関数近似する手法のことである。
方策関数を学習する際、方策関数をニューラルネットワークにしてあげることを考えることで方策勾配法が登場する。以下が方策勾配法でのパラメータ更新式となる。期待収益が大きくなるように学習させたいため、$\epsilon \nabla J(\theta)$が加算されている。
\theta^{(t+1)}=\theta^{(t)}+\epsilon \nabla J(\theta)
この$\nabla J(\theta)$は以下の式で表現できることが知られている。
\nabla_{\theta}J(\theta)=\mathbb{E}_{\pi_{\theta}}[\nabla_{\theta}\log \pi_{\theta}(a|s)Q^{\pi}(s,a)]
#9. AlphaGo
AlphaGoにAlphaGo LeeとAlphaGo Zeroがある。
AlphaGo LeeはPolicyNetとValueNetの２つのネットワークが登場し、ともに畳み込みニューラルネットワークになっている。PolicyNetが強化学習での方策関数で、ValueNetが価値関数である。PolicyNetでは、48チャンネルある19×19の盤面の特徴を入力し、19×19マスの着手予想確率が出力される。ValueNetでは、49チャンネルある19×19の盤面の特徴を入力し、現局面の勝率を-1～1で表したものが出力される。
AlphaGo Leeの学習は以下のステップで行われる。
- 教師あり学習によるRollOutPolicyとPolicyNetの学習
- 強化学習によるPolicyNetの学習
- 強化学習によるValueNetの学習
PolicyNetでは現在の盤面の状態から石をどこに置くか計算するのに3ミリ秒かかるため、何億回と繰り返す学習においては時間がかかる問題が発生する。そこで精度は落ちるがスピード重視で作られたものがRollOutPolicyで、こちらはPolicyNetより1000倍速い3マイクロ秒しかかからない。このRollOutPolicyを学習の最初のステップである教師あり学習で用いられる。価値関数を学習させる手法としてモンテカルロ木探索があるが、これを行う際に、RollOutPolicyを用いることで高速に学習を進めることができる。
AlphaGo ZeroはPolicyValueNetという１つのネットワークに統合され、Leeとは以下の違いがある。
- 教師あり学習を一切行わず、強化学習のみで作成
- 特徴入力からヒューリスティックな要素を排除し、石の配置のみにした
- PolicyNetとValueNetを１つのネットワークに統合した
- Residual Netを導入した
- モンテカルロ木探索からRollOutシミュレーションをなくした
Residual Netはネットワークにショートカット構造を追加して、勾配の爆発、消失を抑える効果を狙ったものである。
#10. 軽量化・高速化技術
##10.1. データ並列化
親モデルを各ワーカーに子モデルとしてコピー。その後、データを分割し、各ワーカーごとに計算させる。データ並列化は各モデルのパラメータの合わせ方で、同期型か非同期型か決まる。同期型では各ワーカーが計算が終わるのを待ち、全ワーカーの勾配が出たところで勾配の平均を計算し、親モデルのパラメータを更新する。非同期型では各ワーカーはお互いの計算を待たず、各子モデルごとに更新を行う。学習が終わった子モデルはパラメータサーバにPushされる。新たに学習を始める時は、パラメータサーバからPopしたモデルに対して学習していく。同期型の方が精度が良いことが多いので、主流となっている。
##10.2. モデル並列化
親モデルを各ワーカーに分割し、それぞれのモデルを学習させる。全てのデータで学習が終わった後で、一つのモデルに復元する。
##10.3. GPUによる高速化
GPUは比較的低性能なコアが多数あり、簡単な並列処理が得意である。ニューラルネットの学習は単純な行列演算が多いので、GPUによる高速化が可能。
##10.4. 量子化
量子化は通常のパラメータの64 bit浮動小数点を32 bitなど下位の精度に落とすことでメモリと演算処理の削減を行う。計算の高速化や省メモリ化といったメリットがある一方で精度の低下というデメリットがある。
##10.5. 蒸留
蒸留は規模の大きなモデルの知識を用いて軽量なモデルの作成を行う。蒸留は教師モデルと生徒モデルの2つで構成される。教師モデルは予測精度の高い、複雑なモデルやアンサンブルされたモデルで、生徒モデルは教師モデルをもとに作られる軽量なモデルである。
##10.6. プルーニング
プルーニングはモデルの精度に寄与の少ないニューロンを削除することでモデルの高速化や軽量化を行う。
#11. 応用モデル
##11.1. 要点まとめ
###11.1.1. MobileNet
MobileNetは画像認識モデルの一種。一般的な畳み込みレイヤーは計算量が多いが、MobileNetはDepthwise ConvolutionとPointwise Convolutionの組み合わせで軽量化を実現している。Depthwise Convolutionではフィルタの数は１で固定で入力マップのチャネルごとに畳み込みを実施し、Pointwise Convolutionではカーネルを１×１に固定して入力マップのポイントごとに畳み込みを実施する。これらを組み合わせることでMobileNetでは計算量を削減する。
###11.1.2. DenseNet
DenseNetは画像認識モデルの一種。DenseNetで特徴的なのはDenseBlockと呼ばれる部分となる。DenseBlockではその中のレイヤーを通過するごとにチャネルが増えていくような演算が行われている。DenseBlockではDenseBlockに入ってきた画像データにDenseBlock内の前のレイヤーでの計算結果が付け加えられたものが処理対象となる。それに対してBatch正規化やRelu関数による変換、3x3畳み込み層による処理などを行ってDenseBlock内の次のレイヤーに進んでいく。進むたびに計算結果が付け加えられていき、チャネルがどんどん増えていくのが特徴である。この増える量のことを成長率（ハイパーパラメータ）という。チャネル数が増えていくためのDenseBlockを通過すると、チャネルを減らすためのレイヤーであるTransiton Layerが用意されている。これは畳み込みとプーリングから構成される。
###11.1.3. Batch Normalization
Batch Normalizationはレイヤー間を流れるデータの分布をミニバッチ単位で平均が0・分散が1になるように正規化する。学習時間の短縮や初期値への依存低減、過学習の抑制など効果がある。ミニバッチ単位であるためバッチサイズに影響を受けやすいという問題点がある。バッチサイズが小さいと学習が収束しないことがあり、代わりにLayer Normalization（１画像に対して平均が0・分散が1になるようにする）やInstance Nrom（１画像中の１チャネルに対して平均が0・分散が1になるようにする）などの正規化手法が使われることが多い。
###11.1.4. Wavenet
Wavenetは音声生成モデルとなる。時系列データに対して畳み込み（Dilated convolution）を適用している。層が深くなるにつれて畳み込むリンクを離すということを行っている。これによってより長い範囲の情報をうまく使えるようにしている。
##11.2. 確認テスト
（い）$H×W×C×D_k×D_k$
（う）$H×W×C×M$
答えは「Dilated causal convolution」
答えは「パラメータ数に対する受容野が広い」
#12. Transformer
##12.1. 要点まとめ
Transformerは2017年6月に登場し、特徴としては従来のエンコーダーデコーダーモデルで使用されていたRNNは全く使われておらず、Attention機構のみを用いて作られた計算量の少ないモデルとなる。全体の構成としてエンコーダーとデコーダーに分かれており、エンコーダーではRNNを使用しないのでPositional Encodingで最初に単語ベクトルに単語の位置情報を付加し、それに対してAttention機構をかけて全結合層に流すことを行い、デコーダーでもRNNを使用しないため、未来の単語を見ないようにマスクをする機構が入っていたりする。エンコーダーもデコーダーもAttenntionはSelf-Attentionを用いている。
##12.2. 実装演習
##12.2.1. データセット
def load_data(file_path):
"""
テキストファイルからデータを読み込む
:param file_path: str, テキストファイルのパス
:return data: list, 文章（単語のリスト）のリスト
"""
data = []
for line in open(file_path, encoding='utf-8'):
words = line.strip().split() # スペースで単語を分割
data.append(words)
return data
train_X = load_data('./data/train.en')
train_Y = load_data('./data/train.ja')
# 訓練データと検証データに分割
train_X, valid_X, train_Y, valid_Y = train_test_split(train_X, train_Y, test_size=0.2, random_state=random_state)
# データセットの中身を確認
print('train_X:', train_X[:5])
print('train_Y:', train_Y[:5])
以下の結果のように英語と日本語が対になったデータセットであることがわかる。
train_X: [['where', 'shall', 'we', 'eat', 'tonight', '?'], ['i', 'made', 'a', 'big', 'mistake', 'in', 'choosing', 'my', 'wife', '.'], ['i', "'ll", 'have', 'to', 'think', 'about', 'it', '.'], ['it', 'is', 'called', 'a', 'lily', '.'], ['could', 'you', 'lend', 'me', 'some', 'money', 'until', 'this', 'weekend', '?']]
train_Y: [['今夜', 'は', 'どこ', 'で', '食事', 'を', 'し', 'よ', 'う', 'か', '。'], ['僕', 'は', '妻', 'を', '選', 'ぶ', 'の', 'に', '大変', 'な', '間違い', 'を', 'し', 'た', '。'], ['考え', 'と', 'く', 'よ', '。'], ['ｌｉｌｙ', 'と', '呼', 'ば', 'れ', 'て', 'い', 'ま', 'す', '。'], ['今週末', 'まで', 'いくら', 'か', '金', 'を', '貸', 'し', 'て', 'くれ', 'ま', 'せ', 'ん', 'か', '。']]
以下では文章を単語IDの列に変換することを行っている。
PAD = 0
UNK = 1
BOS = 2
EOS = 3
PAD_TOKEN = '<PAD>'
UNK_TOKEN = '<UNK>'
BOS_TOKEN = '<S>'
EOS_TOKEN = '</S>'
MIN_COUNT = 2 # 語彙に含める単語の最低出現回数
word2id = {
PAD_TOKEN: PAD,
BOS_TOKEN: BOS,
EOS_TOKEN: EOS,
UNK_TOKEN: UNK,
}
vocab_X = Vocab(word2id=word2id)
vocab_Y = Vocab(word2id=word2id)
vocab_X.build_vocab(train_X, min_count=MIN_COUNT)
vocab_Y.build_vocab(train_Y, min_count=MIN_COUNT)
vocab_size_X = len(vocab_X.id2word)
vocab_size_Y = len(vocab_Y.id2word)
def sentence_to_ids(vocab, sentence):
"""
単語のリストをインデックスのリストに変換する
:param vocab: Vocabのインスタンス
:param sentence: list of str
:return indices: list of int
"""
ids = [vocab.word2id.get(word, UNK) for word in sentence]
ids = [BOS] + ids + [EOS] # EOSを末尾に加える
return ids
train_X = [sentence_to_ids(vocab_X, sentence) for sentence in train_X]
train_Y = [sentence_to_ids(vocab_Y, sentence) for sentence in train_Y]
valid_X = [sentence_to_ids(vocab_X, sentence) for sentence in valid_X]
valid_Y = [sentence_to_ids(vocab_Y, sentence) for sentence in valid_Y]
DataLoaderの定義を行う。
class DataLoader(object):
def __init__(self, src_insts, tgt_insts, batch_size, shuffle=True):
"""
:param src_insts: list, 入力言語の文章（単語IDのリスト）のリスト
:param tgt_insts: list, 出力言語の文章（単語IDのリスト）のリスト
:param batch_size: int, バッチサイズ
:param shuffle: bool, サンプルの順番をシャッフルするか否か
"""
self.data = list(zip(src_insts, tgt_insts))
self.batch_size = batch_size
self.shuffle = shuffle
self.start_index = 0
self.reset()
def reset(self):
if self.shuffle:
self.data = shuffle(self.data, random_state=random_state)
self.start_index = 0
def __iter__(self):
return self
def __next__(self):
def preprocess_seqs(seqs):
# パディング
max_length = max([len(s) for s in seqs])
data = [s + [PAD] * (max_length - len(s)) for s in seqs]
# 単語の位置を表現するベクトルを作成
positions = [[pos+1 if w != PAD else 0 for pos, w in enumerate(seq)] for seq in data]
# テンソルに変換
data_tensor = torch.tensor(data, dtype=torch.long, device=device)
position_tensor = torch.tensor(positions, dtype=torch.long, device=device)
return data_tensor, position_tensor
# ポインタが最後まで到達したら初期化する
if self.start_index >= len(self.data):
self.reset()
raise StopIteration()
# バッチを取得して前処理
src_seqs, tgt_seqs = zip(*self.data[self.start_index:self.start_index+self.batch_size])
src_data, src_pos = preprocess_seqs(src_seqs)
tgt_data, tgt_pos = preprocess_seqs(tgt_seqs)
# ポインタを更新する
self.start_index += self.batch_size
return (src_data, src_pos), (tgt_data, tgt_pos)
##12.2.2. 各モジュールの定義
入出力の単語のEmbedding時に単語の位置情報を埋め込むPosition Encodingの定義を行う。
def position_encoding_init(n_position, d_pos_vec):
"""
Positional Encodingのための行列の初期化を行う
:param n_position: int, 系列長
:param d_pos_vec: int, 隠れ層の次元数
:return torch.tensor, size=(n_position, d_pos_vec)
"""
# PADがある単語の位置はpos=0にしておき、position_encも0にする
position_enc = np.array([
[pos / np.power(10000, 2 * (j // 2) / d_pos_vec) for j in range(d_pos_vec)]
if pos != 0 else np.zeros(d_pos_vec) for pos in range(n_position)])
position_enc[1:, 0::2] = np.sin(position_enc[1:, 0::2]) # dim 2i
position_enc[1:, 1::2] = np.cos(position_enc[1:, 1::2]) # dim 2i+1
return torch.tensor(position_enc, dtype=torch.float)
内積でAttentionを計算し、スケーリングを行うScaled Dot-Product Attentionの定義を行う。
class ScaledDotProductAttention(nn.Module):
def __init__(self, d_model, attn_dropout=0.1):
"""
:param d_model: int, 隠れ層の次元数
:param attn_dropout: float, ドロップアウト率
"""
super(ScaledDotProductAttention, self).__init__()
self.temper = np.power(d_model, 0.5) # スケーリング因子
self.dropout = nn.Dropout(attn_dropout)
self.softmax = nn.Softmax(dim=-1)
def forward(self, q, k, v, attn_mask):
"""
:param q: torch.tensor, queryベクトル,
size=(n_head*batch_size, len_q, d_model/n_head)
:param k: torch.tensor, key,
size=(n_head*batch_size, len_k, d_model/n_head)
:param v: torch.tensor, valueベクトル,
size=(n_head*batch_size, len_v, d_model/n_head)
:param attn_mask: torch.tensor, Attentionに適用するマスク,
size=(n_head*batch_size, len_q, len_k)
:return output: 出力ベクトル,
size=(n_head*batch_size, len_q, d_model/n_head)
:return attn: Attention
size=(n_head*batch_size, len_q, len_k)
"""
# QとKの内積でAttentionの重みを求め、スケーリングする
attn = torch.bmm(q, k.transpose(1, 2)) / self.temper # (n_head*batch_size, len_q, len_k)
# Attentionをかけたくない部分がある場合は、その部分を負の無限大に飛ばしてSoftmaxの値が0になるようにする
attn.data.masked_fill_(attn_mask, -float('inf'))
attn = self.softmax(attn)
attn = self.dropout(attn)
output = torch.bmm(attn, v)
return output, attn
Scaled Dot-Product Attentionを複数のヘッドで並列化するMulti-head Attentionの定義を行う。
class MultiHeadAttention(nn.Module):
def __init__(self, n_head, d_model, d_k, d_v, dropout=0.1):
"""
:param n_head: int, ヘッド数
:param d_model: int, 隠れ層の次元数
:param d_k: int, keyベクトルの次元数
:param d_v: int, valueベクトルの次元数
:param dropout: float, ドロップアウト率
"""
super(MultiHeadAttention, self).__init__()
self.n_head = n_head
self.d_k = d_k
self.d_v = d_v
# 各ヘッドごとに異なる重みで線形変換を行うための重み
# nn.Parameterを使うことで、Moduleのパラメータとして登録できる. TFでは更新が必要な変数はtf.Variableでラップするのでわかりやすい
self.w_qs = nn.Parameter(torch.empty([n_head, d_model, d_k], dtype=torch.float))
self.w_ks = nn.Parameter(torch.empty([n_head, d_model, d_k], dtype=torch.float))
self.w_vs = nn.Parameter(torch.empty([n_head, d_model, d_v], dtype=torch.float))
# nn.init.xavier_normal_で重みの値を初期化
nn.init.xavier_normal_(self.w_qs)
nn.init.xavier_normal_(self.w_ks)
nn.init.xavier_normal_(self.w_vs)
self.attention = ScaledDotProductAttention(d_model)
self.layer_norm = nn.LayerNorm(d_model) # 各層においてバイアスを除く活性化関数への入力を平均０、分散１に正則化
self.proj = nn.Linear(n_head*d_v, d_model) # 複数ヘッド分のAttentionの結果を元のサイズに写像するための線形層
# nn.init.xavier_normal_で重みの値を初期化
nn.init.xavier_normal_(self.proj.weight)
self.dropout = nn.Dropout(dropout)
def forward(self, q, k, v, attn_mask=None):
"""
:param q: torch.tensor, queryベクトル,
size=(batch_size, len_q, d_model)
:param k: torch.tensor, key,
size=(batch_size, len_k, d_model)
:param v: torch.tensor, valueベクトル,
size=(batch_size, len_v, d_model)
:param attn_mask: torch.tensor, Attentionに適用するマスク,
size=(batch_size, len_q, len_k)
:return outputs: 出力ベクトル,
size=(batch_size, len_q, d_model)
:return attns: Attention
size=(n_head*batch_size, len_q, len_k)
"""
d_k, d_v = self.d_k, self.d_v
n_head = self.n_head
# residual connectionのための入力 出力に入力をそのまま加算する
residual = q
batch_size, len_q, d_model = q.size()
batch_size, len_k, d_model = k.size()
batch_size, len_v, d_model = v.size()
# 複数ヘッド化
# torch.repeat または .repeatで指定したdimに沿って同じテンソルを作成
q_s = q.repeat(n_head, 1, 1) # (n_head*batch_size, len_q, d_model)
k_s = k.repeat(n_head, 1, 1) # (n_head*batch_size, len_k, d_model)
v_s = v.repeat(n_head, 1, 1) # (n_head*batch_size, len_v, d_model)
# ヘッドごとに並列計算させるために、n_headをdim=0に、batch_sizeをdim=1に寄せる
q_s = q_s.view(n_head, -1, d_model) # (n_head, batch_size*len_q, d_model)
k_s = k_s.view(n_head, -1, d_model) # (n_head, batch_size*len_k, d_model)
v_s = v_s.view(n_head, -1, d_model) # (n_head, batch_size*len_v, d_model)
# 各ヘッドで線形変換を並列計算(p16左側`Linear`)
q_s = torch.bmm(q_s, self.w_qs) # (n_head, batch_size*len_q, d_k)
k_s = torch.bmm(k_s, self.w_ks) # (n_head, batch_size*len_k, d_k)
v_s = torch.bmm(v_s, self.w_vs) # (n_head, batch_size*len_v, d_v)
# Attentionは各バッチ各ヘッドごとに計算させるためにbatch_sizeをdim=0に寄せる
q_s = q_s.view(-1, len_q, d_k) # (n_head*batch_size, len_q, d_k)
k_s = k_s.view(-1, len_k, d_k) # (n_head*batch_size, len_k, d_k)
v_s = v_s.view(-1, len_v, d_v) # (n_head*batch_size, len_v, d_v)
# Attentionを計算(p16.左側`Scaled Dot-Product Attention * h`)
outputs, attns = self.attention(q_s, k_s, v_s, attn_mask=attn_mask.repeat(n_head, 1, 1))
# 各ヘッドの結果を連結(p16左側`Concat`)
# torch.splitでbatch_sizeごとのn_head個のテンソルに分割
outputs = torch.split(outputs, batch_size, dim=0) # (batch_size, len_q, d_model) * n_head
# dim=-1で連結
outputs = torch.cat(outputs, dim=-1) # (batch_size, len_q, d_model*n_head)
# residual connectionのために元の大きさに写像(p16左側`Linear`)
outputs = self.proj(outputs) # (batch_size, len_q, d_model)
outputs = self.dropout(outputs)
outputs = self.layer_norm(outputs + residual)
return outputs, attns
単語列の位置ごとに独立して処理する2層のネットワークであるPosition-Wise Feed Forward Networkの定義を行う。
class PositionwiseFeedForward(nn.Module):
"""
:param d_hid: int, 隠れ層1層目の次元数
:param d_inner_hid: int, 隠れ層2層目の次元数
:param dropout: float, ドロップアウト率
"""
def __init__(self, d_hid, d_inner_hid, dropout=0.1):
super(PositionwiseFeedForward, self).__init__()
# window size 1のconv層を定義することでPosition wiseな全結合層を実現する.
self.w_1 = nn.Conv1d(d_hid, d_inner_hid, 1)
self.w_2 = nn.Conv1d(d_inner_hid, d_hid, 1)
self.layer_norm = nn.LayerNorm(d_hid)
self.dropout = nn.Dropout(dropout)
self.relu = nn.ReLU()
def forward(self, x):
"""
:param x: torch.tensor,
size=(batch_size, max_length, d_hid)
:return: torch.tensor,
size=(batch_size, max_length, d_hid)
"""
residual = x
output = self.relu(self.w_1(x.transpose(1, 2)))
output = self.w_2(output).transpose(2, 1)
output = self.dropout(output)
return self.layer_norm(output + residual)
Attentionに対して２つのマスクを定義する。
以下ではkey側の系列のPADトークンに対してAttentionを行わないようにするマスクを定義する。
def get_attn_padding_mask(seq_q, seq_k):
"""
keyのPADに対するattentionを0にするためのマスクを作成する
:param seq_q: tensor, queryの系列, size=(batch_size, len_q)
:param seq_k: tensor, keyの系列, size=(batch_size, len_k)
:return pad_attn_mask: tensor, size=(batch_size, len_q, len_k)
"""
batch_size, len_q = seq_q.size()
batch_size, len_k = seq_k.size()
pad_attn_mask = seq_k.data.eq(PAD).unsqueeze(1) # (N, 1, len_k) PAD以外のidを全て0にする
pad_attn_mask = pad_attn_mask.expand(batch_size, len_q, len_k) # (N, len_q, len_k)
return pad_attn_mask
_seq_q = torch.tensor([[1, 2, 3]])
_seq_k = torch.tensor([[4, 5, 6, 7, PAD]])
_mask = get_attn_padding_mask(_seq_q, _seq_k) # 行がquery、列がkeyに対応し、key側がPAD(=0)の時刻だけ1で他が0の行列ができる
print('query:\n', _seq_q)
print('key:\n', _seq_k)
print('mask:\n', _mask)
query:
tensor([[1, 2, 3]])
key:
tensor([[4, 5, 6, 7, 0]])
mask:
tensor([[[False, False, False, False, True],
[False, False, False, False, True],
[False, False, False, False, True]]])
Decoder側でSelf Attentionを行う際に、各時刻で未来の情報に対するAttentionを行わないようにするマスクを定義する。
def get_attn_subsequent_mask(seq):
"""
未来の情報に対するattentionを0にするためのマスクを作成する
:param seq: tensor, size=(batch_size, length)
:return subsequent_mask: tensor, size=(batch_size, length, length)
"""
attn_shape = (seq.size(1), seq.size(1))
# 上三角行列(diagonal=1: 対角線より上が1で下が0)
subsequent_mask = torch.triu(torch.ones(attn_shape, dtype=torch.uint8, device=device), diagonal=1)
subsequent_mask = subsequent_mask.repeat(seq.size(0), 1, 1)
return subsequent_mask
_seq = torch.tensor([[1,2,3,4]])
_mask = get_attn_subsequent_mask(_seq) # 行がquery、列がkeyに対応し、queryより未来のkeyの値が1で他は0の行列ができいる
print('seq:\n', _seq)
print('mask:\n', _mask)
seq:
tensor([[1, 2, 3, 4]])
mask:
tensor([[[0, 1, 1, 1],
[0, 0, 1, 1],
[0, 0, 0, 1],
[0, 0, 0, 0]]], device='cuda:0', dtype=torch.uint8)
##12.2.3. モデルの定義
以下ではEncoderの定義を行う。
class EncoderLayer(nn.Module):
"""Encoderのブロックのクラス"""
def __init__(self, d_model, d_inner_hid, n_head, d_k, d_v, dropout=0.1):
"""
:param d_model: int, 隠れ層の次元数
:param d_inner_hid: int, Position Wise Feed Forward Networkの隠れ層2層目の次元数
:param n_head: int, ヘッド数
:param d_k: int, keyベクトルの次元数
:param d_v: int, valueベクトルの次元数
:param dropout: float, ドロップアウト率
"""
super(EncoderLayer, self).__init__()
# Encoder内のSelf-Attention
self.slf_attn = MultiHeadAttention(
n_head, d_model, d_k, d_v, dropout=dropout)
# Postionwise FFN
self.pos_ffn = PositionwiseFeedForward(d_model, d_inner_hid, dropout=dropout)
def forward(self, enc_input, slf_attn_mask=None):
"""
:param enc_input: tensor, Encoderの入力,
size=(batch_size, max_length, d_model)
:param slf_attn_mask: tensor, Self Attentionの行列にかけるマスク,
size=(batch_size, len_q, len_k)
:return enc_output: tensor, Encoderの出力,
size=(batch_size, max_length, d_model)
:return enc_slf_attn: tensor, EncoderのSelf Attentionの行列,
size=(n_head*batch_size, len_q, len_k)
"""
# Self-Attentionのquery, key, valueにはすべてEncoderの入力（enc_input）が入る
enc_output, enc_slf_attn = self.slf_attn(
enc_input, enc_input, enc_input, attn_mask=slf_attn_mask)
enc_output = self.pos_ffn(enc_output)
return enc_output, enc_slf_attn
class Encoder(nn.Module):
"""EncoderLayerブロックからなるEncoderのクラス"""
def __init__(
self, n_src_vocab, max_length, n_layers=6, n_head=8, d_k=64, d_v=64,
d_word_vec=512, d_model=512, d_inner_hid=1024, dropout=0.1):
"""
:param n_src_vocab: int, 入力言語の語彙数
:param max_length: int, 最大系列長
:param n_layers: int, レイヤー数
:param n_head: int, ヘッド数
:param d_k: int, keyベクトルの次元数
:param d_v: int, valueベクトルの次元数
:param d_word_vec: int, 単語の埋め込みの次元数
:param d_model: int, 隠れ層の次元数
:param d_inner_hid: int, Position Wise Feed Forward Networkの隠れ層2層目の次元数
:param dropout: float, ドロップアウト率
"""
super(Encoder, self).__init__()
n_position = max_length + 1
self.max_length = max_length
self.d_model = d_model
# Positional Encodingを用いたEmbedding
self.position_enc = nn.Embedding(n_position, d_word_vec, padding_idx=PAD)
self.position_enc.weight.data = position_encoding_init(n_position, d_word_vec)
# 一般的なEmbedding
self.src_word_emb = nn.Embedding(n_src_vocab, d_word_vec, padding_idx=PAD)
# EncoderLayerをn_layers個積み重ねる
self.layer_stack = nn.ModuleList([
EncoderLayer(d_model, d_inner_hid, n_head, d_k, d_v, dropout=dropout)
for _ in range(n_layers)])
def forward(self, src_seq, src_pos):
"""
:param src_seq: tensor, 入力系列,
size=(batch_size, max_length)
:param src_pos: tensor, 入力系列の各単語の位置情報,
size=(batch_size, max_length)
:return enc_output: tensor, Encoderの最終出力,
size=(batch_size, max_length, d_model)
:return enc_slf_attns: list, EncoderのSelf Attentionの行列のリスト
"""
# 一般的な単語のEmbeddingを行う
enc_input = self.src_word_emb(src_seq)
# Positional EncodingのEmbeddingを加算する
enc_input += self.position_enc(src_pos)
enc_slf_attns = []
enc_output = enc_input
# key(=enc_input)のPADに対応する部分のみ1のマスクを作成
enc_slf_attn_mask = get_attn_padding_mask(src_seq, src_seq)
# n_layers個のEncoderLayerに入力を通す
for enc_layer in self.layer_stack:
enc_output, enc_slf_attn = enc_layer(
enc_output, slf_attn_mask=enc_slf_attn_mask)
enc_slf_attns += [enc_slf_attn]
return enc_output, enc_slf_attns
さらにDecoderの定義を以下では行う。
class DecoderLayer(nn.Module):
"""Decoderのブロックのクラス"""
def __init__(self, d_model, d_inner_hid, n_head, d_k, d_v, dropout=0.1):
"""
:param d_model: int, 隠れ層の次元数
:param d_inner_hid: int, Position Wise Feed Forward Networkの隠れ層2層目の次元数
:param n_head: int, ヘッド数
:param d_k: int, keyベクトルの次元数
:param d_v: int, valueベクトルの次元数
:param dropout: float, ドロップアウト率
"""
super(DecoderLayer, self).__init__()
# Decoder内のSelf-Attention
self.slf_attn = MultiHeadAttention(n_head, d_model, d_k, d_v, dropout=dropout)
# Encoder-Decoder間のSource-Target Attention
self.enc_attn = MultiHeadAttention(n_head, d_model, d_k, d_v, dropout=dropout)
# Positionwise FFN
self.pos_ffn = PositionwiseFeedForward(d_model, d_inner_hid, dropout=dropout)
def forward(self, dec_input, enc_output, slf_attn_mask=None, dec_enc_attn_mask=None):
"""
:param dec_input: tensor, Decoderの入力,
size=(batch_size, max_length, d_model)
:param enc_output: tensor, Encoderの出力,
size=(batch_size, max_length, d_model)
:param slf_attn_mask: tensor, Self Attentionの行列にかけるマスク,
size=(batch_size, len_q, len_k)
:param dec_enc_attn_mask: tensor, Soutce-Target Attentionの行列にかけるマスク,
size=(batch_size, len_q, len_k)
:return dec_output: tensor, Decoderの出力,
size=(batch_size, max_length, d_model)
:return dec_slf_attn: tensor, DecoderのSelf Attentionの行列,
size=(n_head*batch_size, len_q, len_k)
:return dec_enc_attn: tensor, DecoderのSoutce-Target Attentionの行列,
size=(n_head*batch_size, len_q, len_k)
"""
# Self-Attentionのquery, key, valueにはすべてDecoderの入力（dec_input）が入る
dec_output, dec_slf_attn = self.slf_attn(
dec_input, dec_input, dec_input, attn_mask=slf_attn_mask)
# Source-Target-AttentionのqueryにはDecoderの出力(dec_output), key, valueにはEncoderの出力（enc_output）が入る
dec_output, dec_enc_attn = self.enc_attn(
dec_output, enc_output, enc_output, attn_mask=dec_enc_attn_mask)
dec_output = self.pos_ffn(dec_output)
return dec_output, dec_slf_attn, dec_enc_attn
class Decoder(nn.Module):
"""DecoderLayerブロックからなるDecoderのクラス"""
def __init__(
self, n_tgt_vocab, max_length, n_layers=6, n_head=8, d_k=64, d_v=64,
d_word_vec=512, d_model=512, d_inner_hid=1024, dropout=0.1):
"""
:param n_tgt_vocab: int, 出力言語の語彙数
:param max_length: int, 最大系列長
:param n_layers: int, レイヤー数
:param n_head: int, ヘッド数
:param d_k: int, keyベクトルの次元数
:param d_v: int, valueベクトルの次元数
:param d_word_vec: int, 単語の埋め込みの次元数
:param d_model: int, 隠れ層の次元数
:param d_inner_hid: int, Position Wise Feed Forward Networkの隠れ層2層目の次元数
:param dropout: float, ドロップアウト率
"""
super(Decoder, self).__init__()
n_position = max_length + 1
self.max_length = max_length
self.d_model = d_model
# Positional Encodingを用いたEmbedding
self.position_enc = nn.Embedding(
n_position, d_word_vec, padding_idx=PAD)
self.position_enc.weight.data = position_encoding_init(n_position, d_word_vec)
# 一般的なEmbedding
self.tgt_word_emb = nn.Embedding(
n_tgt_vocab, d_word_vec, padding_idx=PAD)
self.dropout = nn.Dropout(dropout)
# DecoderLayerをn_layers個積み重ねる
self.layer_stack = nn.ModuleList([
DecoderLayer(d_model, d_inner_hid, n_head, d_k, d_v, dropout=dropout)
for _ in range(n_layers)])
def forward(self, tgt_seq, tgt_pos, src_seq, enc_output):
"""
:param tgt_seq: tensor, 出力系列,
size=(batch_size, max_length)
:param tgt_pos: tensor, 出力系列の各単語の位置情報,
size=(batch_size, max_length)
:param src_seq: tensor, 入力系列,
size=(batch_size, n_src_vocab)
:param enc_output: tensor, Encoderの出力,
size=(batch_size, max_length, d_model)
:return dec_output: tensor, Decoderの最終出力,
size=(batch_size, max_length, d_model)
:return dec_slf_attns: list, DecoderのSelf Attentionの行列のリスト
:return dec_slf_attns: list, DecoderのSelf Attentionの行列のリスト
"""
# 一般的な単語のEmbeddingを行う
dec_input = self.tgt_word_emb(tgt_seq)
# Positional EncodingのEmbeddingを加算する
dec_input += self.position_enc(tgt_pos)
# Self-Attention用のマスクを作成
# key(=dec_input)のPADに対応する部分が1のマスクと、queryから見たkeyの未来の情報に対応する部分が1のマスクのORをとる
dec_slf_attn_pad_mask = get_attn_padding_mask(tgt_seq, tgt_seq) # (N, max_length, max_length)
dec_slf_attn_sub_mask = get_attn_subsequent_mask(tgt_seq) # (N, max_length, max_length)
dec_slf_attn_mask = torch.gt(dec_slf_attn_pad_mask + dec_slf_attn_sub_mask, 0) # ORをとる
# key(=dec_input)のPADに対応する部分のみ1のマスクを作成
dec_enc_attn_pad_mask = get_attn_padding_mask(tgt_seq, src_seq) # (N, max_length, max_length)
dec_slf_attns, dec_enc_attns = [], []
dec_output = dec_input
# n_layers個のDecoderLayerに入力を通す
for dec_layer in self.layer_stack:
dec_output, dec_slf_attn, dec_enc_attn = dec_layer(
dec_output, enc_output,
slf_attn_mask=dec_slf_attn_mask,
dec_enc_attn_mask=dec_enc_attn_pad_mask)
dec_slf_attns += [dec_slf_attn]
dec_enc_attns += [dec_enc_attn]
return dec_output, dec_slf_attns, dec_enc_attns
class Transformer(nn.Module):
"""Transformerのモデル全体のクラス"""
def __init__(
self, n_src_vocab, n_tgt_vocab, max_length, n_layers=6, n_head=8,
d_word_vec=512, d_model=512, d_inner_hid=1024, d_k=64, d_v=64,
dropout=0.1, proj_share_weight=True):
"""
:param n_src_vocab: int, 入力言語の語彙数
:param n_tgt_vocab: int, 出力言語の語彙数
:param max_length: int, 最大系列長
:param n_layers: int, レイヤー数
:param n_head: int, ヘッド数
:param d_k: int, keyベクトルの次元数
:param d_v: int, valueベクトルの次元数
:param d_word_vec: int, 単語の埋め込みの次元数
:param d_model: int, 隠れ層の次元数
:param d_inner_hid: int, Position Wise Feed Forward Networkの隠れ層2層目の次元数
:param dropout: float, ドロップアウト率
:param proj_share_weight: bool, 出力言語の単語のEmbeddingと出力の写像で重みを共有する
"""
super(Transformer, self).__init__()
self.encoder = Encoder(
n_src_vocab, max_length, n_layers=n_layers, n_head=n_head,
d_word_vec=d_word_vec, d_model=d_model,
d_inner_hid=d_inner_hid, dropout=dropout)
self.decoder = Decoder(
n_tgt_vocab, max_length, n_layers=n_layers, n_head=n_head,
d_word_vec=d_word_vec, d_model=d_model,
d_inner_hid=d_inner_hid, dropout=dropout)
self.tgt_word_proj = nn.Linear(d_model, n_tgt_vocab, bias=False)
nn.init.xavier_normal_(self.tgt_word_proj.weight)
self.dropout = nn.Dropout(dropout)
assert d_model == d_word_vec # 各モジュールの出力のサイズは揃える
if proj_share_weight:
# 出力言語の単語のEmbeddingと出力の写像で重みを共有する
assert d_model == d_word_vec
self.tgt_word_proj.weight = self.decoder.tgt_word_emb.weight
def get_trainable_parameters(self):
# Positional Encoding以外のパラメータを更新する
enc_freezed_param_ids = set(map(id, self.encoder.position_enc.parameters()))
dec_freezed_param_ids = set(map(id, self.decoder.position_enc.parameters()))
freezed_param_ids = enc_freezed_param_ids | dec_freezed_param_ids
return (p for p in self.parameters() if id(p) not in freezed_param_ids)
def forward(self, src, tgt):
src_seq, src_pos = src
tgt_seq, tgt_pos = tgt
src_seq = src_seq[:, 1:]
src_pos = src_pos[:, 1:]
tgt_seq = tgt_seq[:, :-1]
tgt_pos = tgt_pos[:, :-1]
enc_output, *_ = self.encoder(src_seq, src_pos)
dec_output, *_ = self.decoder(tgt_seq, tgt_pos, src_seq, enc_output)
seq_logit = self.tgt_word_proj(dec_output)
return seq_logit
##12.2.4. 学習
以下では学習を実行する。
def compute_loss(batch_X, batch_Y, model, criterion, optimizer=None, is_train=True):
# バッチの損失を計算
model.train(is_train)
pred_Y = model(batch_X, batch_Y)
gold = batch_Y[0][:, 1:].contiguous()
# gold = batch_Y[0].contiguous()
loss = criterion(pred_Y.view(-1, pred_Y.size(2)), gold.view(-1))
if is_train: # 訓練時はパラメータを更新
optimizer.zero_grad()
loss.backward()
optimizer.step()
gold = gold.data.cpu().numpy().tolist()
pred = pred_Y.max(dim=-1)[1].data.cpu().numpy().tolist()
return loss.item(), gold, pred
MAX_LENGTH = 20
batch_size = 64
num_epochs = 15
lr = 0.001
ckpt_path = 'transformer.pth'
max_length = MAX_LENGTH + 2
model_args = {
'n_src_vocab': vocab_size_X,
'n_tgt_vocab': vocab_size_Y,
'max_length': max_length,
'proj_share_weight': True,
'd_k': 32,
'd_v': 32,
'd_model': 128,
'd_word_vec': 128,
'd_inner_hid': 256,
'n_layers': 3,
'n_head': 6,
'dropout': 0.1,
}
# DataLoaderやモデルを定義
train_dataloader = DataLoader(
train_X, train_Y, batch_size
)
valid_dataloader = DataLoader(
valid_X, valid_Y, batch_size,
shuffle=False
)
model = Transformer(**model_args).to(device)
optimizer = optim.Adam(model.get_trainable_parameters(), lr=lr)
criterion = nn.CrossEntropyLoss(ignore_index=PAD, size_average=False).to(device)
def calc_bleu(refs, hyps):
"""
BLEUスコアを計算する関数
:param refs: list, 参照訳。単語のリストのリスト (例： [['I', 'have', 'a', 'pen'], ...])
:param hyps: list, モデルの生成した訳。単語のリストのリスト (例： [['I', 'have', 'a', 'pen'], ...])
:return: float, BLEUスコア(0~100)
"""
refs = [[ref[:ref.index(EOS)]] for ref in refs]
hyps = [hyp[:hyp.index(EOS)] if EOS in hyp else hyp for hyp in hyps]
return 100 * bleu_score.corpus_bleu(refs, hyps)
# 訓練
best_valid_bleu = 0.
for epoch in range(1, num_epochs+1):
start = time.time()
train_loss = 0.
train_refs = []
train_hyps = []
valid_loss = 0.
valid_refs = []
valid_hyps = []
# train
for batch in train_dataloader:
batch_X, batch_Y = batch
loss, gold, pred = compute_loss(
batch_X, batch_Y, model, criterion, optimizer, is_train=True
)
train_loss += loss
train_refs += gold
train_hyps += pred
# valid
for batch in valid_dataloader:
batch_X, batch_Y = batch
loss, gold, pred = compute_loss(
batch_X, batch_Y, model, criterion, is_train=False
)
valid_loss += loss
valid_refs += gold
valid_hyps += pred
# 損失をサンプル数で割って正規化
train_loss /= len(train_dataloader.data)
valid_loss /= len(valid_dataloader.data)
# BLEUを計算
train_bleu = calc_bleu(train_refs, train_hyps)
valid_bleu = calc_bleu(valid_refs, valid_hyps)
# validationデータでBLEUが改善した場合にはモデルを保存
if valid_bleu > best_valid_bleu:
ckpt = model.state_dict()
torch.save(ckpt, ckpt_path)
best_valid_bleu = valid_bleu
elapsed_time = (time.time()-start) / 60
print('Epoch {} [{:.1f}min]: train_loss: {:5.2f} train_bleu: {:2.2f} valid_loss: {:5.2f} valid_bleu: {:2.2f}'.format(
epoch, elapsed_time, train_loss, train_bleu, valid_loss, valid_bleu))
print('-'*80)
Epoch 1 [0.3min]: train_loss: 77.21 train_bleu: 4.69 valid_loss: 41.17 valid_bleu: 10.81
--------------------------------------------------------------------------------
Epoch 2 [0.3min]: train_loss: 39.38 train_bleu: 12.27 valid_loss: 32.21 valid_bleu: 17.61
--------------------------------------------------------------------------------
Epoch 3 [0.3min]: train_loss: 32.01 train_bleu: 17.97 valid_loss: 28.10 valid_bleu: 22.06
--------------------------------------------------------------------------------
Epoch 4 [0.3min]: train_loss: 28.29 train_bleu: 21.68 valid_loss: 25.82 valid_bleu: 24.98
--------------------------------------------------------------------------------
Epoch 5 [0.3min]: train_loss: 25.81 train_bleu: 24.48 valid_loss: 24.19 valid_bleu: 27.15
--------------------------------------------------------------------------------
Epoch 6 [0.3min]: train_loss: 23.95 train_bleu: 26.80 valid_loss: 22.91 valid_bleu: 29.21
--------------------------------------------------------------------------------
Epoch 7 [0.3min]: train_loss: 22.51 train_bleu: 28.61 valid_loss: 22.05 valid_bleu: 30.79
--------------------------------------------------------------------------------
Epoch 8 [0.3min]: train_loss: 21.33 train_bleu: 30.19 valid_loss: 21.44 valid_bleu: 31.20
--------------------------------------------------------------------------------
Epoch 9 [0.3min]: train_loss: 20.33 train_bleu: 31.55 valid_loss: 21.00 valid_bleu: 31.71
--------------------------------------------------------------------------------
Epoch 10 [0.3min]: train_loss: 19.42 train_bleu: 32.90 valid_loss: 20.33 valid_bleu: 32.73
--------------------------------------------------------------------------------
Epoch 11 [0.3min]: train_loss: 18.69 train_bleu: 33.81 valid_loss: 19.95 valid_bleu: 33.62
--------------------------------------------------------------------------------
Epoch 12 [0.3min]: train_loss: 17.92 train_bleu: 35.16 valid_loss: 19.61 valid_bleu: 34.11
--------------------------------------------------------------------------------
Epoch 13 [0.3min]: train_loss: 17.28 train_bleu: 36.05 valid_loss: 19.29 valid_bleu: 34.87
--------------------------------------------------------------------------------
Epoch 14 [0.3min]: train_loss: 16.70 train_bleu: 36.92 valid_loss: 19.12 valid_bleu: 35.40
--------------------------------------------------------------------------------
Epoch 15 [0.4min]: train_loss: 16.17 train_bleu: 37.84 valid_loss: 18.88 valid_bleu: 35.65
--------------------------------------------------------------------------------
学習と検証でそれぞれbleuを求めており、学習が進行するにつれてbleuの値がよくなっていることがわかる。
##12.2.5. 評価
def test(model, src, max_length=20):
# 学習済みモデルで系列を生成する
model.eval()
src_seq, src_pos = src
batch_size = src_seq.size(0)
enc_output, enc_slf_attns = model.encoder(src_seq, src_pos)
tgt_seq = torch.full([batch_size, 1], BOS, dtype=torch.long, device=device)
tgt_pos = torch.arange(1, dtype=torch.long, device=device)
tgt_pos = tgt_pos.unsqueeze(0).repeat(batch_size, 1)
# 時刻ごとに処理
for t in range(1, max_length+1):
dec_output, dec_slf_attns, dec_enc_attns = model.decoder(
tgt_seq, tgt_pos, src_seq, enc_output)
dec_output = model.tgt_word_proj(dec_output)
out = dec_output[:, -1, :].max(dim=-1)[1].unsqueeze(1)
# 自身の出力を次の時刻の入力にする
tgt_seq = torch.cat([tgt_seq, out], dim=-1)
tgt_pos = torch.arange(t+1, dtype=torch.long, device=device)
tgt_pos = tgt_pos.unsqueeze(0).repeat(batch_size, 1)
return tgt_seq[:, 1:], enc_slf_attns, dec_slf_attns, dec_enc_attns
def ids_to_sentence(vocab, ids):
# IDのリストを単語のリストに変換する
return [vocab.id2word[_id] for _id in ids]
def trim_eos(ids):
# IDのリストからEOS以降の単語を除外する
if EOS in ids:
return ids[:ids.index(EOS)]
else:
return ids
# 学習済みモデルの読み込み
model = Transformer(**model_args).to(device)
ckpt = torch.load(ckpt_path)
model.load_state_dict(ckpt)
# テストデータの読み込み
test_X = load_data('./data/dev.en')
test_Y = load_data('./data/dev.ja')
test_X = [sentence_to_ids(vocab_X, sentence) for sentence in test_X]
test_Y = [sentence_to_ids(vocab_Y, sentence) for sentence in test_Y]
test_dataloader = DataLoader(
test_X, test_Y, 1,
shuffle=False
)
src, tgt = next(test_dataloader)
src_ids = src[0][0].cpu().numpy()
tgt_ids = tgt[0][0].cpu().numpy()
print('src: {}'.format(' '.join(ids_to_sentence(vocab_X, src_ids[1:-1]))))
print('tgt: {}'.format(' '.join(ids_to_sentence(vocab_Y, tgt_ids[1:-1]))))
preds, enc_slf_attns, dec_slf_attns, dec_enc_attns = test(model, src)
pred_ids = preds[0].data.cpu().numpy().tolist()
print('out: {}'.format(' '.join(ids_to_sentence(vocab_Y, trim_eos(pred_ids)))))
src: show your own business .
tgt: 自分 の 事 を しろ 。
out: 自分 の <UNK> を 持 っ て い た 。
他にもいくつか試してみたが結果は以下の通りだった。
src: she wrote to me to come at once .
tgt: 彼女 は 私 に すぐ 来 い と の 便り を よこ し た 。
out: 彼女 は 私 に すぐ に 来る 事 を し た 。
src: no . i 'm sorry , i 've got to go back early .
tgt: ごめん なさ い 。 早 く 帰 ら な く ちゃ 。
out: 私 は 散歩 に 行 く と 、 １ つ あ り ま せ ん 。
src: he lived a hard life .
tgt: 彼 は つら い 人生 を 送 っ た 。
out: 彼 は 一生 懸命 に 暮ら し た 。
結果に対する印象としては、一部近い翻訳結果もあるが、変な翻訳結果のほうが多いのかなという印象である。
# BLEUの評価
test_dataloader = DataLoader(
test_X, test_Y, 128,
shuffle=False
)
refs_list = []
hyp_list = []
for batch in test_dataloader:
batch_X, batch_Y = batch
preds, *_ = test(model, batch_X)
preds = preds.data.cpu().numpy().tolist()
refs = batch_Y[0].data.cpu().numpy()[:, 1:].tolist()
refs_list += refs
hyp_list += preds
bleu = calc_bleu(refs_list, hyp_list)
print(bleu)
23.377275753796077
これはBLUEの値だが、20～29だと「主旨は明白であるが、文法上の重大なエラーがある」という評価になると言われている。
#13. 物体検知・セグメンテーション
物体検知では入力画像に対して、物体検出位置であるBounding Box、ラベル、コンフィデンスの３つを併せて出力するのが一般的。
物体検知において代表的なデータセットにVOC12、ILSVRC17、MS COCO18、OICOD18などがある。この中でILSVRC17以外は物体個々にラベルが与えられている。データセットを選択する際の重要な指標に１画像あたりに映っている物体数があるが、これが小さいと日常とはかけ離れやすいデータになり、逆に大きいと物体の部分的な重なりもあり日常生活のコンテキストに近くなる。またクラス数については人が見たら同じ物体なのに違うラベルがつけられていることによりクラス数が多くなっているデータセットもあり、必ずしもクラス数が多いことがよいとは限らない。
クラス分類ではコンフィデンスの閾値を変化させても混合行列に入ってくるサンプル数に変化はないが、物体検知の場合は閾値を超えたものだけ出力されるため、閾値を変化させると混合行列に入ってくるサンプル数も変化する。
物体検知では物体位置の予測精度も評価したいが、そのための指標としてIoUが提案されている。
⇒IoU＝Area of Overlap/Area of Union.
物体検知ではコンフィデンスの閾値だけでなく、IoUの閾値も設定される。コンフィデンスが高い順に評価していくが、IoUが閾値より大きければTP、閾値より小さければFPと評価していき、もしIoUが閾値より大きくてもすでに検出済みのものであればFPと評価される。
コンフィデンスの閾値を変化させることを考えると、PR曲線を考えることができるが、このPR曲線の下側面積をAPと呼び、物体検知の評価指標で用いられる。このAPはクラスラベルごとに計算されるが、すべてのAPの平均をmAPという。
セグメンテーションではどこに何が写っているか各ピクセルに対して単一のクラスラベルを出力する。
CovolutionとPoolingを繰り返すことで解像度が落ちていくが、これがセグメンテーションでは問題となる。落ちた解像度をどのようにして入力と同じ解像度に戻すのかが肝となる。この解像度を戻すことをUp-Samplingという。セグメンテーションの手法にFCNがあるが、これは全結合層が畳み込み層に置き換えられており、これによって出力結果がヒートマップとして出力される。これに対してUp-Samplingを行っていくが、Up-Samplingを行う手法にDeconvolution/Transposed convolutionがあり、以下の手順で行われる。
- 特徴マップのpixel間隔をstrideだけ空ける
- 特徴マップのまわりに(kernel size - 1) - paddingだけ余白を作る
- 畳み込み演算を行う
ただUp-Samplingするだけでは輪郭がぼやけたものとして出力されるため、低レイヤーPooling層の出力をelement-wise addition することでローカルな情報を補完してからUp-samplingすることで輪郭の補完を行う。