TensorFlowで入力データの長さが可変なLSTMを実装する
以下の記事の続き k17trpsynth.hatenablog.com
目的
前回作ったLSTMは入力する文字列の長さをあらかじめ指定して学習し、予測する際も入力データの長さをそれに合わせるというもので、入力データの長さは不変であった。今回は予測時の入力データの長さが可変なLSTMを作りたい。
方法
利用した文字データなど、前回同様。
static_rnn（入力文字数が静的なRNN）をdynamic_rnn（入力文字数が動的なRNN）に変える。
hidden_to_output, final_state = tf.nn.dynamic_rnn(multi_cell, input_to_hidden, initial_state=initial_state, time_major=True)
この際、cell層に与えるデータは「入力文字数」×「バッチサイズ」×「隠れ層のユニット数」のサイズのTensorでなければならない。cellに渡す前に以下のようにインプットデータを整形する。
input = tf.reshape(tf.transpose(input, [1, 0, 2]), [-1, self.input_layer_size]) input_to_hidden = (tf.matmul(input, w_hidden) + b_hidden) input_to_hidden = tf.reshape(input_to_hidden, [self.max_word_length, -1, self.hidden_layer_size])
また、LSTMを用いる際、cellの初期状態は「cellの初期状態」と「参照する直前の出力の初期状態」をタプルで繋げる必要があった。static_rnnでは通常のタプルで繋げるだけで良かったが、dynamic_rnnでそれをすると以下のようなエラーがでる。
TypeError: The two structures don't have the same sequence type. First structure has type <class 'tuple'>, while second structure has type <class 'tensorflow.python.ops.rnn_cell_impl.LSTMStateTuple'>.
dynamic_rnnでは通常のタプルで繋げるのではなく、以下のようにtf.contrib.rnn.LSTMStateTupleで繋げる必要がある。
initial_state_c = tf.placeholder(tf.float64, [None, self.hidden_layer_size]) initial_state_h = tf.placeholder(tf.float64, [None, self.hidden_layer_size]) initial_state = tuple([tf.contrib.rnn.LSTMStateTuple(initial_state_c, initial_state_h)] * self.num_hidden_layer)
コードの全容は以下の通り。
import tensorflow as tf import numpy as np import re class Prepare_data: def __init__(self, file): self.max_word_length = 10 self.vocabulary_size = 27 self.file = open(file) def file_to_text(self): text = self.file.read() return text def text_to_array(self): text = self.file_to_text() text = text.lower() text = text.replace("\n", " ") text = re.sub(r"[^a-z ]", "", text) text = re.sub(r"[ ]+", " ", text) code_list = [] for i in range(len(text)): if text[i] == " ": code_list.append(self.vocabulary_size - 1) else: code_list.append(ord(text[i])-ord("a")) code_array = np.array(code_list) return code_array def array_to_one_hot(self): array = self.text_to_array() one_hot = np.eye(self.vocabulary_size)[array] return one_hot def make_data(self): one_hot = self.array_to_one_hot() data_num = one_hot.shape[0] - self.max_word_length input_data = np.zeros([data_num, self.max_word_length, self.vocabulary_size]) output_data = np.zeros([data_num, self.vocabulary_size]) for i in range(data_num): output_data[i, :] = one_hot[i + self.max_word_length, :] for j in range(self.max_word_length): input_data[i, j, :] = one_hot[i + j, :] training_num = data_num * 4 // 5 input_train = input_data[: training_num] output_train = output_data[: training_num] input_test = input_data[training_num :] output_test = output_data[training_num :] return input_train, output_train, input_test, output_test class Lstm: def __init__(self): self.input_layer_size = 27 self.hidden_layer_size = 30 self.num_hidden_layer = 1 self.output_layer_size = 27 self.max_word_length = 10 self.batch_size = 128 self.epoch = 100 def inference(self, input, initial_state): w_hidden = tf.Variable(tf.random_normal([self.input_layer_size, self.hidden_layer_size], dtype="float64")) b_hidden = tf.Variable(tf.random_normal([self.hidden_layer_size], dtype="float64")) w_output = tf.Variable(tf.random_normal([self.hidden_layer_size, self.output_layer_size], dtype="float64")) b_output = tf.Variable(tf.random_normal([self.output_layer_size], dtype="float64")) input = tf.reshape(tf.transpose(input, [1, 0, 2]), [-1, self.input_layer_size]) input_to_hidden = (tf.matmul(input, w_hidden) + b_hidden) input_to_hidden = tf.reshape(input_to_hidden, [self.max_word_length, -1, self.hidden_layer_size]) cell = tf.nn.rnn_cell.BasicLSTMCell(self.hidden_layer_size, reuse=tf.AUTO_REUSE) multi_cell = tf.nn.rnn_cell.MultiRNNCell([cell] * self.num_hidden_layer) hidden_to_output, final_state = tf.nn.dynamic_rnn(multi_cell, input_to_hidden, initial_state=initial_state, time_major=True) output = (tf.matmul(hidden_to_output[-1], w_output) + b_output) return output def cost(self, output, labels): cost = tf.reduce_mean(tf.nn.softmax_cross_entropy_with_logits(labels=labels, logits=output)) return cost def training(self, cost): optimizer = tf.train.AdamOptimizer() training = optimizer.minimize(cost) return training def train(self, file): test_data = Prepare_data(file) input_train, output_train, input_test, output_test = test_data.make_data() zero_state = np.zeros([self.batch_size, self.hidden_layer_size], dtype="float64") input_data = tf.placeholder(tf.float64, [None, self.max_word_length, self.input_layer_size]) labels = tf.placeholder(tf.float64, [None, self.output_layer_size]) initial_state_c = tf.placeholder(tf.float64, [None, self.hidden_layer_size]) initial_state_h = tf.placeholder(tf.float64, [None, self.hidden_layer_size]) initial_state = tuple([tf.contrib.rnn.LSTMStateTuple(initial_state_c, initial_state_h)] * self.num_hidden_layer) prediction = self.inference(input_data, initial_state) cost = self.cost(prediction, labels) training = self.training(cost) correct = tf.equal(tf.argmax(prediction, 1), tf.argmax(labels, 1)) accuracy = tf.reduce_mean(tf.cast(correct, dtype="float64")) with tf.Session() as sess: init = tf.global_variables_initializer() sess.run(init) for epoch in range(self.epoch): step = 1 sum_cost = 0 sum_acc = 0 while self.batch_size * step < input_train.shape[0]: input_batch = input_train[self.batch_size * (step - 1) : self.batch_size * step] output_batch = output_train[self.batch_size * (step - 1) : self.batch_size * step] c, _, a= sess.run([cost, training, accuracy], feed_dict = {input_data: input_batch, labels: output_batch, initial_state_c: zero_state, initial_state_h: zero_state}) sum_cost += c sum_acc += a step += 1 ave_cost = sum_cost / step epoch_acc = sum_acc / step print("epoch: {0}, cost: {1}, epoch_accuracy: {2}".format(epoch, ave_cost, epoch_acc)) print("Training finished") saver = tf.train.Saver() saver.save(sess, "./lstm_model") zero_state = np.zeros([input_test.shape[0], self.hidden_layer_size], dtype = "float64") a = sess.run(accuracy, feed_dict = {input_data: input_test, labels: output_test, initial_state_c: zero_state, initial_state_h: zero_state}) print("accuracy: {0}".format(a)) def predict(self, context): context = context.replace("\n", " ") context = re.sub(r"[^a-z ]", "", context) context = re.sub(r"[ ]+", " ", context) code_list = [] for i in range(len(context)): if context[i] == " ": code_list.append(self.input_layer_size - 1) else: code_list.append(ord(context[i])-ord("a")) code_array = np.array(code_list) one_hot = np.eye(self.input_layer_size)[code_array] input_pred = np.array([one_hot]) zero_state = np.zeros([1, self.hidden_layer_size], dtype="float64") input_data = tf.placeholder(tf.float64, [None, self.max_word_length, self.input_layer_size]) initial_state_c = tf.placeholder(tf.float64, [None, self.hidden_layer_size]) initial_state_h = tf.placeholder(tf.float64, [None, self.hidden_layer_size]) initial_state = tuple([tf.contrib.rnn.LSTMStateTuple(initial_state_c, initial_state_h)] * self.num_hidden_layer) prediction = tf.nn.softmax(self.inference(input_pred, initial_state)) labels_pred = tf.argmax(prediction, 1) with tf.Session() as sess: saver = tf.train.Saver() saver.restore(sess, "./lstm_model") p, l = sess.run([prediction, labels_pred], feed_dict = {input_data: input_pred, initial_state_c: zero_state, initial_state_h: zero_state}) for i in range(27): c = "_" if i == 26 else chr(i + ord("a")) print("{0}: {1}%".format(c, int(p[0][i] * 10000) / 100)) print("prediction: {0}{1}".format(context, "_" if l[0] == 26 else chr(l[0] + ord("a")))) if __name__=="__main__": test_lstm = Lstm() test_lstm.train("make_data.txt") test_lstm.predict("convenienc")
