Keras で mnist の画像学習を CNN を使ってやってみる

• 概要
• 環境
• サンプルコード
• Conv2D のレイヤーを増やしてみる
• 最後に
• 参考サイト

概要

前回 Keras + mnist のチュートリアルをやりました
今回は CNN を使った方法を紹介します
チュートリアルはこちらになります

勉強のために Conv2D のレイヤーを増やしたバージョンも紹介します

環境

• macOS 14.2.1
• Python 3.11.6
• tenforflow 2.15.0
• keras 2.15.0

サンプルコード

import numpy as np
import tensorflow as tf

mnist = tf.keras.datasets.mnist
# 60000個の学習データ
(x_train, y_train), (x_test, y_test) = mnist.load_data()
# 学習データとテストデータを3次元から4次元に変換、Conv2D を使う場合は大抵の場合4次元必要、前回 https://hawksnowlog.blogspot.com/2024/01/keras-mnist-tutorial.html Dense のみは3次元のデータのままいける
x_train = x_train.reshape((60000, 28, 28, 1))
x_test = x_test.reshape((10000, 28, 28, 1))
# ピクセル情報は0-255の範囲なので255で割ることで0-1の範囲に変換する
x_train, x_test = x_train / 255.0, x_test / 255.0

model = tf.keras.models.Sequential([
    tf.keras.layers.Conv2D(32, (3, 3), activation='relu', input_shape=(28, 28, 1)),  # 3x3 の複数のフィルタを32枚使って画像にフィルタをかけ新たな行列を作成
    tf.keras.layers.MaxPooling2D((2, 2)),  # 作成した新たな行列に対して 2x2 の領域で最大の値を取得、更にそれを新たな行列とする
    # ここで更に Conv2D + MaxPooling2D のレイヤーを増やしても OK、ただし学習時間は長くなる
    tf.keras.layers.Conv2D(64, (3, 3), activation='relu'),
    tf.keras.layers.MaxPooling2D((2, 2)),
    tf.keras.layers.Conv2D(64, (3, 3), activation='relu'),
    # ここから出力用のレイヤー
    tf.keras.layers.Flatten(),  # 3次元->1次元への変換
    tf.keras.layers.Dense(64, activation='relu'),  # 64ノードの全結合->10ノードの全結合
    tf.keras.layers.Dense(10, activation='softmax'),  # 最後はソフトマックスで10個ある出力の合計が1になるように変換する (定番
])

# モデルコンパイル
model.compile(optimizer='adam',  # 最適化関数は Adaptive moment
              loss='sparse_categorical_crossentropy',  # 損失関数は交差エントロピー誤差
              metrics=['accuracy'])  # 評価関数は正解率

# # 学習
model.fit(x_train, y_train, epochs=5)


# 評価
model.evaluate(x_test, y_test)
# 予測と結果表示
results = model.predict(x_test[:5])
# 結果はDense(10)なので10個の配列の予測値から成り立っている
# その配列10個中で一番値の大きいインデックスが予測した手書きの数字になる
# np.argmax はそれを一発でやってくれる便利メソッド
print(np.argmax(results, axis=1))  # [7 2 1 0 4] になるはず
print(y_test[:5])  # [7 2 1 0 4] こっちは答え

Conv2D のレイヤーを増やしてみる

前のコードのレイヤーを増やしてみます
最後の Conv2D(64, (3, 3), activation='relu') で (3, 3, 64) となりこれに更に conv2D でフィルタをかけると (1, 1, 128) などになり

ValueError: One of the dimensions in the output is <= 0 due to downsampling in conv2d_3. Consider increasing the input size. Received input shape [None, 1, 1, 128] which would produce output shape with a zero or negative value in a dimension.

というエラーになるので少しフィルタの次元数を調整します
具体的に (3, 3) を (2, 2) のフィルタにして次のレイヤに行く際の変換した画像の縦横の次元数を少しずつ減らすようにしています

import numpy as np
import tensorflow as tf

mnist = tf.keras.datasets.mnist
# 60000個の学習データ
(x_train, y_train), (x_test, y_test) = mnist.load_data()
# 学習データとテストデータを3次元から4次元に変換、Conv2D を使う場合は大抵の場合4次元必要、前回 https://hawksnowlog.blogspot.com/2024/01/keras-mnist-tutorial.html Dense のみは3次元のデータのままいける
x_train = x_train.reshape((60000, 28, 28, 1))
x_test = x_test.reshape((10000, 28, 28, 1))
# ピクセル情報は0-255の範囲なので255で割ることで0-1の範囲に変換する
x_train, x_test = x_train / 255.0, x_test / 255.0

model = tf.keras.models.Sequential([
    tf.keras.layers.Conv2D(32, (2, 2), activation='relu', input_shape=(28, 28, 1)),  # 3x3 の複数のフィルタを32枚使って画像にフィルタをかけ新たな行列を作成
    tf.keras.layers.MaxPooling2D((2, 2)),  # 作成した新たな行列に対して 2x2 の領域で最大の値を取得、更にそれを新たな行列とする
    # ここで更に Conv2D + MaxPooling2D のレイヤーを増やしても OK、ただし学習時間は長くなる
    # 以下では 128 個のフィルタで Conv2D するレイヤーを増やしている
    tf.keras.layers.Conv2D(64, (2, 2), activation='relu'),
    tf.keras.layers.MaxPooling2D((2, 2)),
    tf.keras.layers.Conv2D(128, (2, 2), activation='relu'),  # レイヤーを増やす場合変換した行列の縦横の次元数が途中のレイヤで (1, 1) にならないように Conv2D のフィルタの次元数を調整する必要がある
    tf.keras.layers.MaxPooling2D((2, 2)),
    tf.keras.layers.Conv2D(128, (2, 2), activation='relu'),
    # ここから出力用のレイヤー
    tf.keras.layers.Flatten(),  # 3次元->1次元への変換
    tf.keras.layers.Dense(64, activation='relu'),  # 64ノードの全結合->10ノードの全結合
    tf.keras.layers.Dense(10, activation='softmax'),  # 最後はソフトマックスで10個ある出力の合計が1になるように変換する (定番
])


# モデルコンパイル
model.compile(optimizer='adam',  # 最適化関数は Adaptive moment
              loss='sparse_categorical_crossentropy',  # 損失関数は交差エントロピー誤差
              metrics=['accuracy'])  # 評価関数は正解率

# # 学習
model.fit(x_train, y_train, epochs=5)


# 評価
model.evaluate(x_test, y_test)
# 予測と結果表示
results = model.predict(x_test[:5])
# 結果はDense(10)なので10個の配列の予測値から成り立っている
# その配列10個中で一番値の大きいインデックスが予測した手書きの数字になる
# np.argmax はそれを一発でやってくれる便利メソッド
print(np.argmax(results, axis=1))  # [7 2 1 0 4] になるはず
print(y_test[:5])  # [7 2 1 0 4] こっちは答え

model.summary()

 Layer (type)                Output Shape              Param #   
=================================================================
 conv2d (Conv2D)             (None, 27, 27, 32)        160       
                                                                 
 max_pooling2d (MaxPooling2  (None, 13, 13, 32)        0         
 D)                                                              
                                                                 
 conv2d_1 (Conv2D)           (None, 12, 12, 64)        8256      
                                                                 
 max_pooling2d_1 (MaxPoolin  (None, 6, 6, 64)          0         
 g2D)                                                            
                                                                 
 conv2d_2 (Conv2D)           (None, 5, 5, 128)         32896     
                                                                 
 max_pooling2d_2 (MaxPoolin  (None, 2, 2, 128)         0         
 g2D)                                                            
                                                                 
 conv2d_3 (Conv2D)           (None, 1, 1, 128)         65664     
                                                                 
 flatten (Flatten)           (None, 128)               0         
                                                                 
 dense (Dense)               (None, 64)                8256      
                                                                 
 dense_1 (Dense)             (None, 10)                650       
                                                                 
=================================================================
Total params: 115882 (452.66 KB)
Trainable params: 115882 (452.66 KB)
Non-trainable params: 0 (0.00 Byte)
_________________________________________________________________

最後に

Keras + CNN で画像解析を試してみました
前回はただの Dense レイヤーだけでしたが今回は CNN という方法を使って判定しています
精度はどちらも良かったのですが画像解析の主流は CNN かなと思います
単純にレイヤーを増やすだけで精度がどんどんよくなるのが摩訶不思議な感じなしますがそれが DeepLearning のすごいところです

CNN 自体の説明は以下の動画が非常に参考になりました
https://www.youtube.com/watch?v=xzzTYL90M8s

参考サイト

• https://www.tensorflow.org/tutorials/images/intro_to_cnns?hl=ja
• https://qiita.com/kenichiro-yamato/items/60affeb7ca9f67c87a17

Keras で mnist のチュートリアルをやってみる

• 概要
• 環境
• サンプルコード
• 学習データについて
• 学習時間
• 最後に
• 参考サイト

概要

これまでは文章を学習させてきました
今回は画像認識を keras を使ってやってみました
コードの解説や疑問点はコメントで記載しています

mnist + Keras のチュートリアルをやってみました
コードはリファクタしていません

環境

• macOS 14.2.1
• Python 3.11.6
• tenforflow 2.15.0
• keras 2.15.0

サンプルコード

各種何をやっているかの説明はコメントに記載しています

import numpy as np
import tensorflow as tf

mnist = tf.keras.datasets.mnist
# 60000個の学習データ
(x_train, y_train), (x_test, y_test) = mnist.load_data()
# ピクセル情報は0-255の範囲なので255で割ることで0-1の範囲に変換する
x_train, x_test = x_train / 255.0, x_test / 255.0

model = tf.keras.models.Sequential([
    tf.keras.layers.Flatten(input_shape=(28, 28)),  # 1つの画像データが28x28のピクセルデータ
    tf.keras.layers.Dense(128, activation='relu'),  # ここが128なのが不明64や256でも動作はする (ハイパーパラメータか？
    tf.keras.layers.Dropout(0.2),  # ここの0.2なのも不明、そもそもDropout がなくても動作する (ハイパーパラメータか？
    tf.keras.layers.Dense(10)  # 結果は0-9の手書きの数字のどれかなので10出力
])

# 損失関数の作成 (交差エントロピー誤差)
loss_fn = tf.keras.losses.SparseCategoricalCrossentropy(from_logits=True)

# モデルコンパイル
model.compile(optimizer="adam",
              loss=loss_fn,
              metrics=["accuracy"])

# 学習
model.fit(x_train, y_train, epochs=5)
# 評価
model.evaluate(x_test, y_test)
# 予測と結果表示
results = model.predict(x_test[:5])
# 結果はDense(10)なので10個の配列の予測値から成り立っている
# その配列10個中で一番値の大きいインデックスが予測した手書きの数字になる
# np.argmax はそれを一発でやってくれる便利メソッド
print(np.argmax(results, axis=1))  # [7 2 1 0 4] になるはず
print(y_test[:5])  # [7 2 1 0 4] こっちは答え

学習データについて

• 60000 枚の画像のデータが入っている
  • 1枚の画像情報は 28 x 28 のデータで格納されている
  • それぞれの情報は0-255までのピクセル情報が格納されている
• x_train が画像のピクセルデータで y_train がピクセルデータの実際の答えの数字が格納されている一次元の配列 (当然 60000 個のデータ

学習時間

M2 mac mini の GPU ありででだいたい 5 分程度で完了します

最後に

やはりレイヤーを作る部分が職人技のような気がします

参考サイト

• https://zenn.dev/mi_ztyanya/books/9bcdf19bb90504/viewer/f19568
• https://weblabo.oscasierra.net/python/keras-mnist-sample.html

Keras 学習コンテンツ一覧

• 概要
• 環境
• 記事一覧
• その他記事
• 感じたこと

概要

Keras の使い方を学びたかったので簡単な使い方から RNN を使った文章生成までやってみました
それぞれ記事にしたので一覧をまとめておきます
また学習した感想など記載します

環境

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

記事一覧

• 第0回: RNNのチュートリアルをやってみた
• 第1回: Keras 超入門
• 第2回: Keras でモデルを保存する
• 第3回: Keras で Tensorboard を使う
• 第4回: Keras で独自の評価関数を作る
• 第5回: Keras で超簡単 RNN を試してみる
• 第6回: Keras で2値判定のRNNをやってみる
• 第7回: Keras + RNN の文章生成チュートリアルをリファクタしてみた

その他記事

• Import “tensorflow.keras” could not be resolved 対策

感じたこと

• Keras の使い方もそうだが numpy の使い方のほうが重要だと感じた
  • numpy で作成したベクトルの扱い方
  • 多次元ベクトル、shape の情報をイメージできるか
  • reshape など
• Keras のレイヤーを理解し使えるようになるまでは大変そう
  • 入力の数と出力の数を考慮してレイヤーを結合できるか
  • 学習時と評価時でちゃんと適切なテストデータを作成することができるか
• 学習よりもデータを生成、クリーニングする段階のほうが重要であり難しいコードになりがちな気がする
  • numpy や tf.data を使って学習に必要なフォーマットに落とせるか
  • RNN の場合は「一つずらす」必要があるのでこれをコードで表現できるか
  • numpy や tf.data は最悪使わないでもできるので無理に使わなくてもいい
• コードをリファクタリングしたほうが理解が深まる
  • なぜかすべての紹介されているコードがすべて一枚ペラのコードばかりでクラス化されていたり冗長部分の排除ができていなかった
  • TestData, Model, Test など必要な機能、役割ごとに分割して管理すると何をやっているかもわかりやすい気がした
  • 自分が紹介してるコードはクラス化されています
• ハイパーパラメータについて
  • こればかりは数学の知識や経験が大事になってくる
  • サンプルの値をそのまま使ってもいいがハイパーパラメータの変化による効果や仕組みを理解してないと Keras を理解したとは言えないと思う
  • それでも王道を使っておけば深層学習自体のポテンシャルの高さである程度の精度は出すことができる (それ以上を求める場合には数学の知識を含めハイパーパラメータの扱いが必要な気がする
• 評価関数、損失関数など
  • ここも経験や数学の知識によって精度に大きく影響する
  • 一流を求めるなら自作の関数の作成と理解が必要
  • それでもデフォルトの関数を使えばそれなりの評価が出てしまう

Keras + RNN の文章生成チュートリアルをリファクタしてみた

• 概要
• 環境
• サンプルコード
• ちょっと解説
• リファクタリング不足点
• 最後に
• 参考サイト

概要

過去にやった文章生成のチュートリアルをリファクタして更に理解を深めてみました

環境

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

サンプルコード

import os
from dataclasses import dataclass
from typing import Optional

import numpy as np
import tensorflow as tf
from keras.src.callbacks import History
from tensorflow.keras.callbacks import ModelCheckpoint
from tensorflow.keras.layers import GRU, Dense, Embedding
from tensorflow.keras.losses import sparse_categorical_crossentropy
from tensorflow.keras.models import Sequential
from tensorflow.keras.utils import get_file


# テストデータを管理するクラス
class TestData:
    # 文章を分割する長さ
    SEQ_LENGTH = 100
    # シャッフル用のバッファサイズ
    BUFFER_SIZE = 10000
    # シャッフル後の分割単位
    BATCH_SIZE = 64

    def __init__(self) -> None:
        # テキストデータのダウンロード
        self.__download()
        # 各種文字と文字インデックスの組み合わせの辞書と配列を作成
        self.char2idx = self.__gen_char2idx()
        self.idx2char = self.__gen_idx2char()
        # テキスト情報を数字情報のインデックス番号の配列に変換
        self.text_as_int = np.array([self.char2idx[c] for c in self.text])
        # TensorSliceDataset の作成、配列で要素は tf.Tensor に変換されたテキスト情報を数字情報のインデックス番号がある
        self.char_dataset = tf.data.Dataset.from_tensor_slices(self.text_as_int)
        # 100 文字ごとに分割して TensorSliceDataset を再度作成、BatchDataset になる
        # ここで説明変数と目的変数が生成される
        self.sequences = self.char_dataset.batch(
            self.SEQ_LENGTH + 1, drop_remainder=True
        )
        # 各 100 文字ごとの文章情報を1文字づらした [input: tf.Tensor, target: tf.Tensor] 配列を作ります、全文章分 map で繰り返して作ります
        dataset = self.sequences.map(self.shift)
        # シャルフルし再度 64 個ごとにまとめます、BatchDataset になる
        self.dataset = dataset.shuffle(self.BUFFER_SIZE).batch(
            self.BATCH_SIZE, drop_remainder=True
        )

    @property
    def vocab(self) -> list:
        # 文字を分割して辞書順に並び替えたリストを生成
        return sorted(set(self.text))

    def __download(self):
        path_to_file = get_file(
            "shakespeare.txt",
            "https://storage.googleapis.com/download.tensorflow.org/data/shakespeare.txt",
        )
        self.text = open(path_to_file, "rb").read().decode(encoding="utf-8")

    def __gen_char2idx(self) -> dict:
        # 文字に対応した数字のインデックス番号の組み合わせを生成
        return {u: i for i, u in enumerate(self.vocab)}

    def __gen_idx2char(self) -> np.ndarray:
        # 文字を分割して辞書順に並び替えたリストを ndarray 形式に変換
        return np.array(self.vocab)

    def shift(self, chunk: tf.raw_ops.BatchDataset):
        # RNN 学習のために1文字づらしたシーケンスを作成します
        # :-1 は最後から1文字抜いた先頭からすべての文字 (説明変数)
        independent = chunk[:-1]
        # 1: は先頭から1文字抜いた最後までのすべての文字 (目的変数)
        dependent = chunk[1:]
        return independent, dependent


# RNN モデルを管理するクラス
class RNNModel:
    EMBEDDING_DIM = 256  # 埋め込み層の次元数
    RUN_UNITS = 1024  # RNN ユニット数
    model: Sequential

    def __init__(self, vocab_size: int, batch_size) -> None:
        self.model = Sequential()
        self.vocab_size = vocab_size
        self.batch_size = batch_size

    def build(self, batch_size: Optional[int] = None, new: bool = False):
        if batch_size is None:
            bs = self.batch_size
        else:
            bs = batch_size
        if new:
            self.model = Sequential()
        # 埋め込み層、固定次元の密ベクトルに変換するレイヤー
        self.model.add(
            Embedding(
                self.vocab_size,
                self.EMBEDDING_DIM,
                batch_input_shape=[bs, None],
            )
        )
        # RNN 層、今回は GRU レイヤーを使う
        self.model.add(
            GRU(
                self.RUN_UNITS,
                return_sequences=True,
                stateful=True,
                recurrent_initializer="glorot_uniform",
            )
        )
        # 出力層
        self.model.add(Dense(self.vocab_size))

    def compile(self) -> None:
        self.model.compile(optimizer="adam", loss=Loss().keras_scc)

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

    def train(
        self,
        dataset: tf.raw_ops.BatchDataset,
        epochs=10,
        callbacks=[],
    ) -> History:
        history = self.model.fit(dataset, epochs=epochs, callbacks=callbacks)
        return history

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

    def rebuild(self, checkpoint_dir: str = "./training_checkpoints"):
        # テスト時の入力のバッチサイズは1になるので batch_size=1 で再度ビルドする
        self.build(batch_size=1, new=True)
        self.model.load_weights(tf.train.latest_checkpoint(checkpoint_dir))
        self.model.build(tf.TensorShape([1, None]))


# 損失関数を管理するクラス
class Loss:
    def keras_scc(self, y_true, y_pred):
        # sparse_categorical_crossentropy をカスタム
        return sparse_categorical_crossentropy(y_true, y_pred, from_logits=True)


# モデルを再度構築するので1回目のモデルを保存するためのコールバック
class Callback:
    @classmethod
    def save_model(cls) -> ModelCheckpoint:
        checkpoint_dir = "./training_checkpoints"
        checkpoint_prefix = os.path.join(checkpoint_dir, "ckpt_{epoch}")
        return ModelCheckpoint(filepath=checkpoint_prefix, save_weights_only=True)


# 実際にモデルを評価するためのクラス
@dataclass
class Result:
    text: str

    def show(self):
        print(self.text)


class Test:
    # 1.0 より大きいほど意外なテキストを生成する、1.0 は予測したそのままの値になる
    TEMPERRATURE = 1.0
    # 生成する文字数
    NUM_GENERATE_CHAR = 1000

    def __init__(
        self, model: RNNModel, test_data: TestData, start_string: str = "ROMEO: "
    ) -> None:
        self.model = model
        self.test_data = test_data
        self.start_string = start_string
        # start_string を数字のインデックス配列に変換
        input_eval = [self.test_data.char2idx[char] for char in start_string]
        self.input_eval = tf.expand_dims(input_eval, 0)

    def run(self) -> Result:
        self.model.model.reset_states()
        generated_text = []
        for _ in range(self.NUM_GENERATE_CHAR):
            # 文字の予測
            predictions = self.model.model(self.input_eval)
            # サイズが1の次元を削除し次元数を減らす
            predictions = tf.squeeze(predictions, 0)
            # カテゴリー分布をつかってモデルから返された文字のインデックス番号を予測
            predictions = predictions / self.TEMPERRATURE
            # 最後の要素が入力に対して予測された文字のインデックス番号
            predicted_id = tf.random.categorical(predictions, num_samples=1)[
                -1, 0
            ].numpy()
            # 過去の隠れ状態とともに予測された文字をモデルへのつぎの入力として渡す
            self.input_eval = tf.expand_dims([predicted_id], 0)
            # インデックス番号から文字への変換と結果の格納
            generated_text.append(self.test_data.idx2char[predicted_id])
        return Result(text=self.start_string + "".join(generated_text))


if __name__ == "__main__":
    # テストデータ作成
    test_data = TestData()
    # モデル生成
    model = RNNModel(len(test_data.vocab), TestData.BATCH_SIZE)
    model.build()
    model.compile()
    # # モデルの訓練
    model.train(test_data.dataset, callbacks=[Callback.save_model()])
    # モデルの評価
    model.rebuild()
    model.show()
    test = Test(model, test_data)
    result = test.run()
    # 結果を表示
    result.show()

ちょっと解説

テストデータは単純なテキストを使います
テキスト情報を1文字ずつ分解し数字とのマッピング情報を作成しテキストをすべて数値化することで学習データを作成します
文字情報の数値化は自然言語ではよくある手法になります

データをシャッフルしたりバッチサイズで分割するのは Keras で学習させる際のフォーマットに変換していると理解していますがこの辺りの手法を自然に使いこなせるレベルにならないとダメかなと思います

モデルの生成は Embedding -> GRU -> Dense とシンプルです
Embedding は自然言語の学習ではほぼ必須です
GRU (Gated Recurrent Unit) は RNN の一種です
モデル学習時にチェックポイントを保存しています
評価する際に入力をシンプルにしたいので学習時のバッチサイズ64を再度チェックポイントからモデルをビルドしてバッチサイズを1にするためです

予測した値はそのまま使わずカテゴリー分布という手法を使って再度計算させています (正直理由は不明です

リファクタリング不足点

• TestData をもう少しリファクタリングしたほうがいい
  • データの生成を init ではなくそれぞれの関数にしたほうがいい
  • 説明変数と目的変数を別のクラスで管理した方がいい
• Model の model の管理をリファクタリングしたほうがいい
  • build で新規 or 既存の書き換えを制御するよりかは ModelFactory を作って model を個別に生成できるような仕組みにするといいかも
• Test を柔軟にしたほうがいい
  • いろいろなデータでテストできるようにしたほうがいい
  • 生成文字列サイズなども可変にできるといい
  • 何をやっているかわからないところがあるのでコメントで補完したい

最後に

Keras を使った学習方法や流れはだいたい把握できましたがフルスクラッチでゼロから numpy や keras を使ってモデルを作るにはまだまだ学習が足りないと感じました

numpy で生成したベクトルの扱い方や keras のレイヤーの生成方法やレイヤーに対する入出力の制御、最適化あたりを駆使できるようにならないとオリジナルなモデルを作るのは難しいなと感じました

参考サイト

• https://keras.io/ja/
• https://www.tensorflow.org/api_docs/python/tf/raw_ops/BatchDataset
• https://www.tensorflow.org/tutorials/text/text_generation?hl=ja

Keras で2値判定のRNNをやってみる

• 概要
• 環境
• サンプルコード
• ちょっと解説
• 最後に
• 参考サイト

概要

前回 RNN を使って正弦波の予測を行ってみました
10 個の入力に対して 10 個の予測値を出力するようなモデルを組みました
今回は複数の入力に対して 0 or 1 を判定するようなモデルを組んでみたいと思います

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

環境

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

サンプルコード

from dataclasses import dataclass
from random import randint
from typing import Any

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

# RNN で使う一時系列のデータ数
NUM_RNN = 10
# 学習させるデータの数
NUM_DATA = 200


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

    def __init__(self, num_rnn: int) -> None:
        # 今回は説明変数と目的変数を別で作るのが面倒なので説明変数側で作成してしまう
        # 空の器を作成
        x_train = np.empty((0, num_rnn))
        y_train = np.empty((0, 1))
        for i in range(NUM_DATA):
            num_random = randint(-20, 20)
            if i % 2 == 1:  # 奇数の場合
                # linespaceは等差数列
                x_train = np.append(
                    x_train,
                    np.linspace(
                        num_random, num_random + num_rnn - 1, num=num_rnn
                    ).reshape(1, num_rnn),
                    axis=0,
                )
                # 減少する数列の場合は1
                y_train = np.append(y_train, np.zeros(1).reshape(1, 1), axis=0)
            else:  # 偶数の場合
                x_train = np.append(
                    x_train,
                    np.linspace(
                        num_random, num_random - num_rnn + 1, num=num_rnn
                    ).reshape(1, num_rnn),
                    axis=0,
                )
                # 増加する数列の場合は0
                y_train = np.append(y_train, np.ones(1).reshape(1, 1), axis=0)
        # RNN 学習用に整形
        self.data = x_train.reshape(NUM_DATA, num_rnn, 1)
        self.labels = y_train.reshape(NUM_DATA, 1)


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

    def __init__(self, labels: np.ndarray) -> None:
        self.labels = labels


# テストデータを管理するクラス
class TestData:
    def __init__(self) -> None:
        self.independent = IndependentVariable(NUM_RNN)
        self.dependent = DependentVariable(self.independent.labels)


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

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

    def compile(self) -> None:
        # RNN層
        self.model.add(
            SimpleRNN(
                self.NUM_DIM,
                batch_input_shape=(None, NUM_RNN, 1),
                return_sequences=False,
            )
        )
        # 最終出力が1つの結合層を作成
        self.model.add(Dense(1, activation="sigmoid"))
        self.model.compile(loss="binary_crossentropy", optimizer="adam")

    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:
    y_pred: int
    test_data: np.ndarray

    def show(self):
        if self.y_pred > 0.9:
            print(f"減算する数列です ({self.y_pred}) -> ", self.test_data)
        elif self.y_pred < 0.1:
            print(f"増加する数列です ({self.y_pred}) -> ", self.test_data)
        else:
            print(f"未知の数列です ({self.y_pred}) -> ", self.test_data)


class Test:
    def __init__(self, model: RNNModel, test_data: TestData) -> None:
        self.model = model
        # 学習データの一部をそのまま学習データに利用する
        self.test_data = test_data

    def run(self) -> list[Result]:
        # 10回テスト
        result: list[Result] = []
        for i in range(10):
            y_pred = self.model.model.predict(
                self.test_data.independent.data[i].reshape(1, NUM_RNN, 1)
            )
            r = Result(
                y_pred=y_pred,
                test_data=self.test_data.independent.data[i].reshape(NUM_RNN),
            )
            result.append(r)
        return 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()
    # 結果を表示
    for r in result:
        r.show()

ちょっと解説

今回テストデータの生成はすべて説明変数内で行っています
インタフェースの関係で目的変数用のクラスは残しています

学習させる際は最後に出力 1 の Dense レイヤーを作成するのがポイントです
結果の表示は 0.9 以上 0.1 以下でやっていますがたまーに 0.7 や 0.3 などが出てくる可能性があるのでここは適当に変更してください
RNN の 2 値判定の場合は必ず 1 or 0 が出力されるわけではなく与えられた値から 1 っぽい 0 っぽい値が出てくるのでこの判定は自分で行う必要があります

最後に

2値判定の RNN をやってみました
ほぼ前回のコードと流れも変わっていない感じです
少しリファクタをさぼった感じがするのでもう少しいい感じに書ける気がします

参考サイト

• https://qiita.com/FukuharaYohei/items/39f865bb53cdd5052179

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