# -*- coding: utf-8 -*-
# 参考 TensorFlowの活用ガイド(技術評論社)
# 参考 https://deepinsider.jp/tutor/introtensorflow/whatisrnn
import argparse
import os
import sys
import numpy as np
import pandas as pd
import tensorflow as tf
DEFAULT_DATA_DIR = 'data/prepared/aozorabunko/classify/morph'
VOCAB_SIZE = 50000 # 語彙の数
BATCH_SIZE = 100 #128
MAX_LEN = 64 # 入力層の入力数（このサイズの要素数で、訓練、評価、テストのデータリストを作ることになる。）
HIDDEN_SIZE = 96 #128
EMB_SIZE = 128 #
EMB_SIZE = 64 #
STEPS = 50000
STEPS = 1000
LOG_DIR = 'log'
MODEL_DIR = 'models/classify_rnn'
ENCODING = 'utf-8'
TRAIN_FILE_NAME = 'train.csv'
VALID_FILE_NAME = 'valid.csv'
TEST_FILE_NAME = 'test.csv'
args_output_dir="data/prepared/aozorabunko/classify/morph"
train_file = os.path.join(args_output_dir, TRAIN_FILE_NAME)
valid_file = os.path.join(args_output_dir, VALID_FILE_NAME)
test_file = os.path.join(args_output_dir, TEST_FILE_NAME)
print('\n以前にダウンロード(青空文庫)したファイルは次の通り')
print("train_file:" + train_file)
print("valid_file:" + valid_file)
print("test_file:" + test_file)
print('\n上記のデータファイルからpandasオブジェクト生成')
train_df = pd.read_csv(train_file)
valid_df = pd.read_csv(valid_file)
test_df = pd.read_csv(test_file)
train_df.info()
print('----------以上が、訓練データの情報です。\n')
label_names = sorted(train_df.label.unique())
print('Labels: {}'.format(label_names))
sys.path.append('..') # 親ディレクトリのモージュールを利用
import utils
print("train_df.textのデータでtext_encoderに情報構築")
text_encoder = utils.TextEncoder()
print("次で、出現単語を出現個数の大きい順に並べて最大要素数がVOCAB_SIZE個のリスト語彙を作っている。");
text_encoder.build_vocab(train_df.text, VOCAB_SIZE)
print("語彙Counter情報" ,text_encoder.counter)
print("語彙の総数：" ,text_encoder.word_count )
print("text_encoder.vocabの情報" ,text_encoder.vocab)
print()
# 訓練やテストに使うデータと正解情報を得る関数定義
def build_datamart(df, labels, text_encoder, max_len):
n_samples = len(df)
x = np.zeros((n_samples, max_len), dtype=np.int32)
for i, text in enumerate(df.text):
encoded = text_encoder.encode(text)[:max_len]
x[i][:len(encoded)] = encoded
y = df.label.map(labels.index).astype(np.int32).values
return x, y
print("各データと、そのラベルの情報を取得")
train_x, train_y = build_datamart(train_df, label_names, text_encoder, MAX_LEN)
print("訓練データと、その教師ラベル:", train_x,train_y)
valid_x, valid_y = build_datamart(valid_df, label_names, text_encoder, MAX_LEN)
print("検証データと、その検証ラベル:", valid_x, valid_y)
test_x, test_y = build_datamart(test_df, label_names, text_encoder, MAX_LEN)
print("テストデータと、そのテストラベル:", test_x, test_y)
# バッチ情報を得るためのジェネレータ取得関数定義
def build_generator(x, y, batch_size):
while True:
n_samples = len(x)
indices = np.random.permutation(range(n_samples))
for i in range(0, n_samples, batch_size):
batch_indices = [indices[(i + j) % n_samples] for j in range(batch_size)]
batch_x = x[batch_indices]
batch_y = y[batch_indices]
yield batch_x, batch_y
train_gen = build_generator(train_x, train_y, BATCH_SIZE)
print("訓練用のデータ（教師データを含む）のジェネレータを生成結果:",train_gen)
print("placeholderの用意");
inputs = tf.placeholder(tf.int32, (None, MAX_LEN), name='inputs')
labels = tf.placeholder(tf.int32, (None,), name='labels')
print("ＲＮＮ「Recurrent Neural Network」 ＣＥＬＬの用意");
rnn_cell = tf.contrib.rnn.GRUCell(HIDDEN_SIZE)
def rnn_model(inputs, labels, rnn_cell, n_classes, emb_size, vocab_size, hidden_size):
"""
RNNによる分類モデルで、各引数の概要を示す。
inputs (Tensor): [batch_size, num_steps] の入力データ.
labels (Tensor): [batch_size] の正解ラベル.
rnn_cell (RNNCell): RNN（Recurrent Neural Network）セル.
n_classes (int): 分類クラス数.
emb_size (int): 単語の埋め込み空間のサイズ.
vocab_size (int): 語彙数(単語IDの最大値).
hidden_size (int): 全結合層のサイズ
Returns:
予測結果、確率スコア、誤差関数、正解率、学習演算
"""
# 単語の埋め込み
with tf.variable_scope('emb'):
emb_w = tf.get_variable(
'w',
shape=(vocab_size, emb_size),
initializer=tf.contrib.layers.xavier_initializer()
)
emb = tf.nn.embedding_lookup(emb_w, inputs)
#
# rnn_cell から RNN を構築
with tf.variable_scope('rnn'):
outputs, state = tf.nn.dynamic_rnn(rnn_cell, emb, dtype=tf.float32)
last_outputs = outputs[:, -1] # 分類問題では最後の出力のみを使う
#
# RNN の出力から作者を推定する。ここでは2層のネットワークを使う
with tf.variable_scope('output'):
x = tf.layers.dense(last_outputs, hidden_size, activation=tf.nn.relu)
logits = tf.layers.dense(x, n_classes)
#
one_hot_labels = tf.one_hot(labels, n_classes)
loss = tf.reduce_mean(
tf.nn.softmax_cross_entropy_with_logits(labels=one_hot_labels, logits=logits)
)
#
train_op = tf.train.AdamOptimizer().minimize(loss)
softmax = tf.nn.softmax(logits)
#
pred = tf.cast(tf.argmax(logits, axis=-1, name='pred'), tf.int32)
acc = tf.reduce_mean(tf.cast(tf.equal(pred, labels), tf.float32))
return pred, softmax, loss, acc, train_op
# RNNによる分類モデルの生成
pred, softmax, loss, acc, train_op = rnn_model(
inputs,#入力層のplaceholder（MAX_LENの入力数）。バッチ情報がジェネレータ（train_gen）より与えられる前提
labels,#上記入力に対する正解ラベル。出力層と比較予定バッチ情報がジェネレータ（train_gen）より与えられる前提
rnn_cell,
len(label_names),
EMB_SIZE,
text_encoder.vocab_size,
HIDDEN_SIZE
)
checkpoint_path = os.path.join(MODEL_DIR, 'model.ckpt')
print( "学習データの保存ファイル：",checkpoint_path )
saver = tf.train.Saver() # "学習データの保存準備
with tf.Session() as sess:
sess.run(tf.global_variables_initializer())
loss_sum_ = 0
for step in range(1, STEPS + 1):
# 学習用バッチを取得
batch_x, batch_y = next(train_gen )
loss_, _ = sess.run(
[loss, train_op],
feed_dict={
inputs: batch_x,
labels: batch_y
}
)
loss_sum_ += loss_
#
# 100ステップごとに検証用データで精度を確認
if step % 100 == 0:
pred_, valid_loss_, acc_ = sess.run(
[pred, loss, acc],
feed_dict={
inputs: valid_x,
labels: valid_y
}
)
# 学習用データについてのロスは過去100ステップの平均をとる
avg_loss_ = loss_sum_/100
loss_sum_ = 0
print('step: {}, train loss: {}, valid loss: {}, acc: {}'.format(
step, avg_loss_, valid_loss_, acc_
))
saver.save(sess, checkpoint_path, global_step=step) # "学習データの保存
# Test
ckpt = tf.train.get_checkpoint_state(MODEL_DIR)
with tf.Session() as sess:
saver.restore(sess=sess, save_path=ckpt.model_checkpoint_path)
pred_, loss_, acc_ = sess.run(
[pred, loss, acc],
feed_dict={
inputs: test_x,
labels: test_y
}
)
print('loss: {}, acc: {}'.format(loss_, acc_))