Keras で超簡単 RNN を試してみる

• 概要
• 環境
• インストール
• サンプルコード
• 少し解説
• 結果
• 最後に
• 参考サイト

概要

これまでは単純なニューラルネットを組んでみました
今回は Keras に標準で付属している RNN レイヤーを使って RNN を実装してみます

コードも前回のを引き継いでいます
内容は参考サイトと同じですがコードのリファクタと執筆時点での最新の tenforflow を使って動作確認しています

環境

• macOS 14.0
• Python 3.11.6
  • keras 2.14.0
  • tensorflow 2.14.0
  • matplotlib 3.8.1

インストール

今回は結果をグラフで確認したほうがわかりやすいので matplotlib を追加でインストールします

• pipenv install matplotlib

サンプルコード

少しリファクタ不足です
解説は下に記載しています

from dataclasses import dataclass
from typing import Any

import matplotlib.pyplot as plt
import numpy as np
from tensorflow.keras.layers import Dense, SimpleRNN
from tensorflow.keras.models import Sequential

# RNN で使う一時系列のデータ数
NUM_RNN = 10


# 説明変数を管理するクラス
class IndependentVariable:
    # 3次元の行列データ (40, 10, 1)
    data: np.ndarray

    def __init__(self, num_rnn: int, y: np.ndarray) -> None:
        x_ = []
        # 50-10=40 回ループ
        data_size = y.size - num_rnn
        for i in range(data_size):
            x_.append(y[i : i + num_rnn])
        else:
            # RNN に入力できるように3次元配列にreshapeする
            self.data = np.array(x_).reshape(data_size, num_rnn, 1)


# 目的変数を管理するクラス
class DependentVariable:
    # 3次元の行列データ (40, 10, 1)
    # RNN では目的変数は説明変数の一つずれた配列データを作成するのがポイント
    labels: Any

    def __init__(self, num_rnn: int, y: np.ndarray) -> None:
        y_ = []
        # 50-10=40 回ループ
        data_size = y.size - num_rnn
        for i in range(data_size):
            # rnn では目的変数は説明変数の次の要素の値を設定するので一つずらした10列の配列を設定する
            y_.append(y[i + 1 : i + num_rnn + 1])
        else:
            # RNN に入力できるように3次元配列にreshapeする
            self.labels = np.array(y_).reshape(data_size, num_rnn, 1)


# テストデータを管理するクラス
class TestData:
    def __init__(self) -> None:
        # -4.9 から 4.9 まで 0.2 間隔の配列を作成する
        self.x = np.linspace(-4.9, 4.9)
        # data に対する sin を求めてラベルにする
        self.y = np.sin(self.x)
        self.independent = IndependentVariable(NUM_RNN, self.y)
        self.dependent = DependentVariable(NUM_RNN, self.y)


# RNN モデルを管理するクラス
class RNNModel:
    NUM_DIM = 8  # 中間層の次元数
    model: Sequential

    def __init__(self) -> None:
        self.model = Sequential()

    def compile(self) -> None:
        # SimpleRNNというレイヤーを使う、input_shepe は reshape したい際の後ろから2,3番目の値と同じにする
        self.model.add(
            SimpleRNN(self.NUM_DIM, input_shape=(NUM_RNN, 1), return_sequences=True)
        )
        # 出力は1次元で全結合層を使う
        self.model.add(Dense(1, activation="linear"))
        # 損失関数、最適化関数、評価関するを指定してモデル作成
        self.model.compile(loss="mean_squared_error", optimizer="sgd", metrics=["acc"])

    def show(self):
        self.model.summary()

    def train(
        self,
        independent: IndependentVariable,
        dependent: DependentVariable,
        epochs=20,
        batch_size=8,
    ):
        self.model.fit(
            independent.data, dependent.labels, epochs=epochs, batch_size=batch_size
        )

    def save(self, file_name="my_model"):
        self.model.save(file_name)


# 実際にモデルを評価するためのクラス
@dataclass
class Result:
    result: np.ndarray


class Test:
    def __init__(self, model: RNNModel, test_data: TestData) -> None:
        self.model = model
        self.test_data = test_data
        # 最初の要素を一次元に変換、結果もここに格納されていく
        self.test_data_and_result = self.test_data.independent.data[0].reshape(-1)

    def run(self) -> Result:
        # 学習した40データ分ループしてテスト
        for _ in range(self.test_data.x.size - NUM_RNN):
            # test_data の最後から10要素をテストに使う
            y = self.model.model.predict(
                self.test_data_and_result[-NUM_RNN:].reshape(1, NUM_RNN, 1)
            )
            # 結果をtest_dataに格納、y[0][NUM_RNN-1][0] が予測した値の場所
            # 10個の入力に対して10個の予測値が出てくる
            # ほしいのは前の9個ではなく新たに追加した予測値の分だけなので最後だけ取得して追加する
            self.test_data_and_result = np.append(
                self.test_data_and_result, y[0][NUM_RNN - 1][0]
            )
        return Result(result=self.test_data_and_result)


if __name__ == "__main__":
    # テストデータ作成
    test_data = TestData()
    # モデル生成
    model = RNNModel()
    model.compile()
    # モデルの訓練
    model.train(
        test_data.independent,
        test_data.dependent,
    )
    # モデルの評価
    test = Test(model, test_data)
    result = test.run()
    # 結果をグラフで表示
    plt.plot(test_data.x, test_data.y, label="Training data")
    plt.plot(test_data.x, result.result, label="Predicted")
    plt.legend()
    plt.show()

少し解説

最大のポイントは目的変数の作り方です
RNN は時系列や前の状態から次の値を予測します
なので目的変数は説明変数の次の状態を持つように作成する必要があります

モデルの作成、訓練は非常に簡単で keras に備わっている SimpleRNN というレイヤーを使うだけで完了します
各種関数、エポック、バッチサイズなどは参考サイトを元にしているのでご自身で調整してみてください (ハイパーパラメータ)

評価時にも少し工夫が必要で今回は結果から更に次の結果を取得する必要があるので test_data_and_result という感じでテストーデータに対して結果を結合して更にそれをテストデータとして使っています

もう少しリファクタできそうな感じはありますがとりあえず大まかな流れのクラスを変更することなく書けているので良しとしました

結果

実行ごとに結果がかわります

たまーにおかしな結果を出すこともありましたがほぼほぼそれっぽい結果になりました

最後に

Keras + RNN をやってみました
最大のポイントは目的変数の作り方かなと思います

今回のをベースに過去にやった文章生成のチュートリアルもリファクタしてみたいと思っています

参考サイト

• https://qiita.com/Ishotihadus/items/f8a2fd1d071047413302
• https://qiita.com/FukuharaYohei/items/25de4a0faf634ad34efc