RNNでテキスト生成やってみた！（自然言語処理）【図解速習DeepLearning】#015

こんにちは！こーたろーです。

本日もやってまいりましょう！【図解速習DEEP LEARNING】の自然言語処理の課題です。

今回は、テキストの生成ということで、RNNを用いて作成していきます。

今回は、シェイクスピア作品のデータセットを用いて、データの文字列”Shakesper”という入力に対して、文字列中の次の文字　"e"　を予測するようなモデルを訓練していきます。

このモデルを繰り返し実行（呼び出し）することによって、長い文章を生成していきます。

それでは早速参りましょう！

必要なライブラリーのインポート

import tensorflow as tf
import numpy as np
import os
import time

データセットのダウンロード

シェイクスピアのテキストをダウンロードしていきます。

path_to_file = tf.keras.utils.get_file('shakespeare.txt', 'https://storage.googleapis.com/download.tensorflow.org/data/shakespeare.txt')

text = open(path_to_file, 'rb').read().decode(encoding='utf-8')

print ('Length of text: {} characters'.format(len(text)))

先頭の文字を表示してみましょう。

print(text[:250])

ファイル中の文字のユニーク値の数を調べる。

vocab = sorted(set(text))
print ('{} unique characters'.format(len(vocab)))

データの前処理

データセットは数値として扱う必要があるため、テキストの１文字１文字を数値に割り当てていきます。

65種類の文字に対して、整数を割り当てていきます。

char2idx = {u:i for i, u in enumerate(vocab)}
idx2char = np.array(vocab)

text_as_int = np.array([char2idx[c] for c in text])



print('{')
for char,_ in zip(char2idx, range(20)):
    print('  {:4s}: {:3d},'.format(repr(char), char2idx[char]))
print('  ...\n}')

テキストが数値と入れ替わりました。

表示してみましょう。

print ('{} ---- characters mapped to int ---- > {}'.format(repr(text[:13]), text_as_int[:13]))

tf.data.Dataset.from_tensor_slicesを使用して、テキストベクトルの文字インデックスを連続に変換します。

これは、Helloという文字を例にすると、「H」を入力したら「e」が出力、「e」が入力なら「l」が出力のように、スライジングするデータへと変換するメソッドです。

seq_length = 100
examples_per_epoch = len(text)//(seq_length+1)

char_dataset = tf.data.Dataset.from_tensor_slices(text_as_int)

for i in char_dataset.take(5):
  print(idx2char[i.numpy()])

データセットを変換しています。

始めの5文字を表示させてみました。

「batch」メソッドを使って、個々の文字を、求める長さのシーケンスに変換します。

mapメソッドを使って、各バッチでデータの複製とシフトを行い、インプットデータと正解ラベルを作成していきます。

def split_input_target(chunk):
    input_text = chunk[:-1]
    target_text = chunk[1:]
    return input_text, target_text

dataset = sequences.map(split_input_target)

どのようにスライドしているか確認してみましょう。

sequences = char_dataset.batch(seq_length+1, drop_remainder=True)

for item in sequences.take(5):
  print(repr(''.join(idx2char[item.numpy()])))

モデルの構築

訓練用のバッチを定義していきます。

テキストを分割して、扱い易いようにバッチにまとめていきます。
訓練時は、バッチサイズ単位で、入力をシャッフルしたものを用います。

BATCH_SIZE = 64

BUFFER_SIZE = 10000

dataset = dataset.shuffle(BUFFER_SIZE).batch(BATCH_SIZE, drop_remainder=True)

dataset

モデルの引数を定義します。

vocab_size = len(vocab)

embedding_dim = 256

rnn_units = 1024

モデルは、kerasのSequentialを使って定義していきます。

今回は、繰り返し呼び出し・出力させるということを繰り返すため、関数定義しておきます。

def build_model(vocab_size, embedding_dim, rnn_units, batch_size):
  model = tf.keras.Sequential([
    tf.keras.layers.Embedding(vocab_size, embedding_dim,
                              batch_input_shape=[batch_size, None]),
    tf.keras.layers.GRU(rnn_units,
                        return_sequences=True,
                        stateful=True,
                        recurrent_initializer='glorot_uniform'),
    tf.keras.layers.Dense(vocab_size)
  ])
  return model




model = build_model(
  vocab_size = len(vocab),
  embedding_dim=embedding_dim,
  rnn_units=rnn_units,
  batch_size=BATCH_SIZE)

モデルを訓練する

オプティマイザーと損失関数を定義します。

損失関数は「sparse_categorical_crossentropy」を使用します。

def loss(labels, logits):
  return tf.keras.losses.sparse_categorical_crossentropy(labels, logits, from_logits=True)

example_batch_loss  = loss(target_example_batch, example_batch_predictions)
print("Prediction shape: ", example_batch_predictions.shape, " # (batch_size, sequence_length, vocab_size)")
print("scalar_loss:      ", example_batch_loss.numpy().mean())

オプティマイザーは「adam」としています。

model.compile(optimizer='adam', loss=loss)

「callbacks.ModelCheckpoint」を使って、訓練中のチェックポイントを保存するようにします。

checkpoint_dir = './training_checkpoints'

checkpoint_prefix = os.path.join(checkpoint_dir, "ckpt_{epoch}")

checkpoint_callback=tf.keras.callbacks.ModelCheckpoint(
    filepath=checkpoint_prefix,
    save_weights_only=True)

訓練を実施していきます。

EPOCHS=10

history = model.fit(dataset, epochs=EPOCHS, callbacks=[checkpoint_callback])

テキストを生成する

バッチサイズを１を使用します。

RNNは、状態をタイムステップからタイムステップへと渡すようなアルゴリズムのため、モデルは一度構築すると、固定されたバッチサイズしか受け付けられません。

モデルを異なるバッチサイズで実行するためには、モデルを再構築し、チェックポイントから重みを復元する必要があります。

tf.train.latest_checkpoint(checkpoint_dir)

最終段のチェックポイントデータから重みを新しいモデルにロードします。

model = build_model(vocab_size, embedding_dim, rnn_units, batch_size=1)

model.load_weights(tf.train.latest_checkpoint(checkpoint_dir))

model.build(tf.TensorShape([1, None]))


model.summary()

テキスト生成の関数を定義します。

テキスト生成は、次の手順で行われます。

1. 開始文字列を選択し、RNN の状態を初期化。生成する文字数を設定します。
2. 開始文字列と RNN の状態を使って、次の文字の予測分布出力します。
3. カテゴリー分布を使用して、予測された文字のインデックスを計算します。
4. 予測された文字をモデルの次の入力にします。

上記て順で次の文字を予測した後、更新された RNN の状態が再びモデルにフィードバックされます。
こうやってモデルは前に予測した文字から次々と文字とを生成していきます。。

def generate_text(model, start_string):

  num_generate = 1000

  input_eval = [char2idx[s] for s in start_string]
  input_eval = tf.expand_dims(input_eval, 0)

  text_generated = []

  temperature = 1.0

  model.reset_states()
  for i in range(num_generate):
      predictions = model(input_eval)
      predictions = tf.squeeze(predictions, 0)

      predictions = predictions / temperature
      predicted_id = tf.random.categorical(predictions, num_samples=1)[-1,0].numpy()

      input_eval = tf.expand_dims([predicted_id], 0)

      text_generated.append(idx2char[predicted_id])

  return (start_string + ''.join(text_generated))




print(generate_text(model, start_string=u"ROMEO: "))

生成されたテキストは、単語になっていない部分もありますが、何とか英語を表現しようとしている様子は伺えます。

なんとなく赤ちゃんが大人の真似をしてしゃべってる様子にも似ている気がします。

今回は、RNNで文字生成を行いましたが、まだまだ学習・訓練に改良の余地があるようです。

もっと滑らかにテキストを出力できるように、腕を上げていきたいものです。

ではでは。。