2023年11月8日水曜日

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

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 をやってみました
ほぼ前回のコードと流れも変わっていない感じです
少しリファクタをさぼった感じがするのでもう少しいい感じに書ける気がします

参考サイト

2023年11月7日火曜日

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

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 をやってみました
最大のポイントは目的変数の作り方かなと思います

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

参考サイト

2023年11月6日月曜日

Keras で独自の評価関数を作る

Keras で独自の評価関数を作る

概要

Keras で model.compile 時に独自の評価関数を指定する方法を紹介します

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

環境

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

サンプルコード

import datetime
from dataclasses import dataclass
from typing import Any, Optional, Union

import numpy as np
from tensorflow.keras import backend
from tensorflow.keras.callbacks import ModelCheckpoint, TensorBoard
from tensorflow.keras.layers import Dense
from tensorflow.keras.models import Sequential


# 説明変数を管理するクラス
class IndependentVariable:
    NUM_TRAIN_COL = 2

    # 128行2列の行列データ
    data: np.ndarray[Any, np.dtype[np.float64]]

    def __init__(self, row: int) -> None:
        self.data = np.random.rand(row, self.NUM_TRAIN_COL)


# 目的変数を管理するクラス
class DependentVariable:
    NUM_TRAIN_COL = 1

    # 128行1列の行列データ、説明変数に対する答えのラベルを管理
    labels: Any

    def __init__(self, data: np.ndarray[Any, np.dtype[np.float64]], row: int) -> None:
        # np.sum(data, axis=1) で各行ごとの合計を計算
        # それが 1.0 以上なら True で True(1) or False(0) * 1 をするので最終的には128行の 0 or 1 の配列になる
        self.raw_labels = (np.sum(data, axis=1) > 1.0) * 1
        self.labels = self.raw_labels.reshape(row, self.NUM_TRAIN_COL)


# テストデータを管理するクラス
class TestData:
    NUM_TRAIN_ROW = 128

    def __init__(self) -> None:
        self.independent = IndependentVariable(self.NUM_TRAIN_ROW)
        self.dependent = DependentVariable(self.independent.data, self.NUM_TRAIN_ROW)


# モデルを管理するクラス
CallbackType = Union[ModelCheckpoint, TensorBoard]


class Model:
    model: Sequential

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

    def compile(self) -> None:
        # 結合層(2層->4層)
        self.model.add(Dense(4, input_dim=2, activation="tanh"))
        # 結合層(4層->1層):入力次元を省略すると自動的に前の層の出力次元数を引き継ぐ
        self.model.add(Dense(1, activation="sigmoid"))
        self.model.compile(
            # accuracy は acc, binary_accuracy と同じ
            loss="binary_crossentropy",
            optimizer="sgd",
            # metrics=["accuracy", Metrics().mean_pred],
            metrics=[Metrics().mean_pred],
        )

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

    def train(
        self,
        independent: IndependentVariable,
        dependent: DependentVariable,
        epochs=300,
        validation_split=0.2,
        callbacks: Optional[list[CallbackType]] = [],
    ):
        self.model.fit(
            independent.data,
            dependent.labels,
            epochs=epochs,
            validation_split=validation_split,
            callbacks=callbacks,
        )

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


# model.fit の callback で使用する関数を管理するクラス
class Callback:
    @classmethod
    def save_model(cls, file_name: str = "./model.hdf5") -> ModelCheckpoint:
        return ModelCheckpoint(file_name, save_best_only=True)

    @classmethod
    def tensorboard(cls, log_dir: str = "logs/fit/") -> TensorBoard:
        log_dir = log_dir + datetime.datetime.now().strftime("%Y%m%d-%H%M%S")
        return TensorBoard(log_dir=log_dir, histogram_freq=1)


# 評価関数を管理するクラス
class Metrics:
    def mean_pred(self, _, y_pred):
        return backend.mean(y_pred)


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


class Test:
    NUM_TEST_ROW = 50

    def __init__(self, model: Model) -> None:
        self.model = model
        # テストデータは Independent, DependentVariable を使える
        self.independent = IndependentVariable(self.NUM_TEST_ROW)
        self.dependent = DependentVariable(self.independent.data, self.NUM_TEST_ROW)

    def run(self) -> Result:
        predict = ((self.model.model.predict(self.independent.data) > 0.5) * 1).reshape(
            self.NUM_TEST_ROW
        )
        # 50個中何個正解だったか
        accuracy = sum(predict == self.dependent.raw_labels) / self.NUM_TEST_ROW
        return Result(accuracy=accuracy)


if __name__ == "__main__":
    # テストデータ作成
    test_data = TestData()
    # モデル生成
    model = Model()
    model.compile()
    # モデルの訓練
    model.train(
        test_data.independent,
        test_data.dependent,
        callbacks=[Callback.save_model(), Callback.tensorboard()],
    )
    model.save()
    # モデルの評価
    test = Test(model)
    result = test.run()
    print(result)

ちょっと解説

評価関数は compile 時に metrics で指定します
標準で付属の評価関数もいくつかあります
今までは acc = accuracy = binary_accuracy を指定していました

今回は独自の評価関数を作成して指定します
評価関数は Metrics クラスで管理します
独自の評価関数は 2 つの引数を取ります
y_true (正解ラベル) と y_pred (予測値) が必須になります

今回は backend.mean は平均値を計算するメソッドです

最後に

acc などデフォルトで用意されている評価関数でも精度に大きな問題は出ませんが独自の評価関数にすることで精度が大きく向上する場合もあるので使えて損はない技術だと思います

参考サイト

2023年11月5日日曜日

Keras で Tensorboard を使う

Keras で Tensorboard を使う

概要

tensorboard はモデルや関数、データセットの内容などいろいろな情報を可視化できるツールです
今回は tensorboard を動かしてみました

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

環境

  • macOS 14.0
  • Python 3.11.6
    • keras 2.14.0
    • tensorflow 2.14.0
    • tensorboard 2.14.1

インストール

  • pipenv install tensorboard

サンプルコード

tensorboard は学習の際に出力されるログファイルを元に起動します
まずは学習時のコードを修正します
callbacks を使って model.fit 時に tensorboard に必要なログ情報を出力するようにします

import datetime
from dataclasses import dataclass
from typing import Any, Optional, Union

import numpy as np
from tensorflow.keras.callbacks import ModelCheckpoint, TensorBoard
from tensorflow.keras.layers import Dense
from tensorflow.keras.models import Sequential


# 説明変数を管理するクラス
class IndependentVariable:
    NUM_TRAIN_COL = 2

    # 128行2列の行列データ
    data: np.ndarray[Any, np.dtype[np.float64]]

    def __init__(self, row: int) -> None:
        self.data = np.random.rand(row, self.NUM_TRAIN_COL)


# 目的変数を管理するクラス
class DependentVariable:
    NUM_TRAIN_COL = 1

    # 128行1列の行列データ、説明変数に対する答えのラベルを管理
    labels: Any

    def __init__(self, data: np.ndarray[Any, np.dtype[np.float64]], row: int) -> None:
        # np.sum(data, axis=1) で各行ごとの合計を計算
        # それが 1.0 以上なら True で True(1) or False(0) * 1 をするので最終的には128行の 0 or 1 の配列になる
        self.raw_labels = (np.sum(data, axis=1) > 1.0) * 1
        self.labels = self.raw_labels.reshape(row, self.NUM_TRAIN_COL)


# テストデータを管理するクラス
class TestData:
    NUM_TRAIN_ROW = 128

    def __init__(self) -> None:
        self.independent = IndependentVariable(self.NUM_TRAIN_ROW)
        self.dependent = DependentVariable(self.independent.data, self.NUM_TRAIN_ROW)


# モデルを管理するクラス
CallbackType = Union[ModelCheckpoint, TensorBoard]


class Model:
    model: Sequential

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

    def compile(self) -> None:
        # 結合層(2層->4層)
        self.model.add(Dense(4, input_dim=2, activation="tanh"))
        # 結合層(4層->1層):入力次元を省略すると自動的に前の層の出力次元数を引き継ぐ
        self.model.add(Dense(1, activation="sigmoid"))
        self.model.compile(
            loss="binary_crossentropy", optimizer="sgd", metrics=["accuracy"]
        )

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

    def train(
        self,
        independent: IndependentVariable,
        dependent: DependentVariable,
        epochs=300,
        validation_split=0.2,
        callbacks: Optional[list[CallbackType]] = [],
    ):
        self.model.fit(
            independent.data,
            dependent.labels,
            epochs=epochs,
            validation_split=validation_split,
            callbacks=callbacks,
        )

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


class Callback:
    @classmethod
    def save_model(cls, file_name: str = "./model.hdf5") -> ModelCheckpoint:
        return ModelCheckpoint(file_name, save_best_only=True)

    @classmethod
    def tensorboard(cls, log_dir: str = "logs/fit") -> TensorBoard:
        log_dir = log_dir + datetime.datetime.now().strftime("%Y%m%d-%H%M%S")
        return TensorBoard(log_dir=log_dir, histogram_freq=1)


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


class Test:
    NUM_TEST_ROW = 50

    def __init__(self, model: Model) -> None:
        self.model = model
        # テストデータは Independent, DependentVariable を使える
        self.independent = IndependentVariable(self.NUM_TEST_ROW)
        self.dependent = DependentVariable(self.independent.data, self.NUM_TEST_ROW)

    def run(self) -> Result:
        predict = ((self.model.model.predict(self.independent.data) > 0.5) * 1).reshape(
            self.NUM_TEST_ROW
        )
        # 50個中何個正解だったか
        accuracy = sum(predict == self.dependent.raw_labels) / self.NUM_TEST_ROW
        return Result(accuracy=accuracy)


if __name__ == "__main__":
    # テストデータ作成
    test_data = TestData()
    # モデル生成
    model = Model()
    model.compile()
    # モデルの訓練
    model.train(
        test_data.independent,
        test_data.dependent,
        callbacks=[Callback.save_model(), Callback.tensorboard()],
    )
    model.save()
    # モデルの評価
    test = Test(model)
    result = test.run()
    print(result)

実行後に logs/fit/20231101-101337/ という日付付きのフォルダがあることを確認しましょう

tensorboard 起動

  • pipenv run tensorboard --logdir logs/fit
Serving TensorBoard on localhost; to expose to the network, use a proxy or pass --bind_all
TensorBoard 2.14.1 at http://localhost:6006/ (Press CTRL+C to quit)

あとは localhost:6006 にアクセスすればいろいろな情報が確認できます

最後に

起動までは簡単ですが tensorboard 自体が複雑で使い方がわかるまでに学習が必要そうです

参考サイト

2023年11月4日土曜日

Keras でモデルを保存する

Keras でモデルを保存する

概要

前回 の続きでモデルを保存する方法を学びました
model.fit で callbacks を使って保存する方法と SavedModel を使う方法があるようです
今回はどちらも紹介します

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

環境

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

サンプルコード1 callbacks を使う方法

from dataclasses import dataclass
from typing import Any, Callable, Optional

import numpy as np
from tensorflow.keras.callbacks import ModelCheckpoint
from tensorflow.keras.layers import Dense  # type: ignore
from tensorflow.keras.models import Sequential  # type: ignore


# 説明変数を管理するクラス
class IndependentVariable:
    NUM_TRAIN_COL = 2

    # 128行2列の行列データ
    data: np.ndarray[Any, np.dtype[np.float64]]

    def __init__(self, row: int) -> None:
        self.data = np.random.rand(row, self.NUM_TRAIN_COL)


# 目的変数を管理するクラス
class DependentVariable:
    NUM_TRAIN_COL = 1

    # 128行1列の行列データ、説明変数に対する答えのラベルを管理
    labels: Any

    def __init__(self, data: np.ndarray[Any, np.dtype[np.float64]], row: int) -> None:
        # np.sum(data, axis=1) で各行ごとの合計を計算
        # それが 1.0 以上なら True で True(1) or False(0) * 1 をするので最終的には128行の 0 or 1 の配列になる
        self.raw_labels = (np.sum(data, axis=1) > 1.0) * 1
        self.labels = self.raw_labels.reshape(row, self.NUM_TRAIN_COL)


# テストデータを管理するクラス
class TestData:
    NUM_TRAIN_ROW = 128

    def __init__(self) -> None:
        self.independent = IndependentVariable(self.NUM_TRAIN_ROW)
        self.dependent = DependentVariable(self.independent.data, self.NUM_TRAIN_ROW)


# モデルを管理するクラス
class Model:
    model: Sequential

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

    def compile(self) -> None:
        # 結合層(2層->4層)
        self.model.add(Dense(4, input_dim=2, activation="tanh"))
        # 結合層(4層->1層):入力次元を省略すると自動的に前の層の出力次元数を引き継ぐ
        self.model.add(Dense(1, activation="sigmoid"))
        self.model.compile(
            loss="binary_crossentropy", optimizer="sgd", metrics=["accuracy"]
        )

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

    def train(
        self,
        independent: IndependentVariable,
        dependent: DependentVariable,
        epochs=300,
        validation_split=0.2,
        callbacks: Optional[list[ModelCheckpoint]] = [],
    ):
        self.model.fit(
            independent.data,
            dependent.labels,
            epochs=epochs,
            validation_split=validation_split,
            callbacks=callbacks,
        )


class Callback:
    @classmethod
    def save_model(cls, file_name: str = "./model.hdf5") -> ModelCheckpoint:
        return ModelCheckpoint(file_name, save_best_only=True)


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


class Test:
    NUM_TEST_ROW = 50

    def __init__(self, model: Model) -> None:
        self.model = model
        # テストデータは Independent, DependentVariable を使える
        self.independent = IndependentVariable(self.NUM_TEST_ROW)
        self.dependent = DependentVariable(self.independent.data, self.NUM_TEST_ROW)

    def run(self) -> Result:
        predict = ((self.model.model.predict(self.independent.data) > 0.5) * 1).reshape(
            self.NUM_TEST_ROW
        )
        # 50個中何個正解だったか
        accuracy = sum(predict == self.dependent.raw_labels) / self.NUM_TEST_ROW
        return Result(accuracy=accuracy)


if __name__ == "__main__":
    # テストデータ作成
    test_data = TestData()
    # モデル生成
    model = Model()
    model.compile()
    # モデルの訓練
    model.train(
        test_data.independent, test_data.dependent, callbacks=[Callback.save_model()]
    )
    # モデルの評価
    test = Test(model)
    result = test.run()
    print(result)

実行後にカレントを確認すると model.hdf5 というファイルが保存されているのが確認できます

Callback というクラスを用意してそこで model.fit 用のコールバック関数を管理するようにしました

サンプルコード2 SavedModel を使う方法

from dataclasses import dataclass
from typing import Any, Callable, Optional

import numpy as np
from tensorflow.keras.callbacks import ModelCheckpoint
from tensorflow.keras.layers import Dense  # type: ignore
from tensorflow.keras.models import Sequential  # type: ignore


# 説明変数を管理するクラス
class IndependentVariable:
    NUM_TRAIN_COL = 2

    # 128行2列の行列データ
    data: np.ndarray[Any, np.dtype[np.float64]]

    def __init__(self, row: int) -> None:
        self.data = np.random.rand(row, self.NUM_TRAIN_COL)


# 目的変数を管理するクラス
class DependentVariable:
    NUM_TRAIN_COL = 1

    # 128行1列の行列データ、説明変数に対する答えのラベルを管理
    labels: Any

    def __init__(self, data: np.ndarray[Any, np.dtype[np.float64]], row: int) -> None:
        # np.sum(data, axis=1) で各行ごとの合計を計算
        # それが 1.0 以上なら True で True(1) or False(0) * 1 をするので最終的には128行の 0 or 1 の配列になる
        self.raw_labels = (np.sum(data, axis=1) > 1.0) * 1
        self.labels = self.raw_labels.reshape(row, self.NUM_TRAIN_COL)


# テストデータを管理するクラス
class TestData:
    NUM_TRAIN_ROW = 128

    def __init__(self) -> None:
        self.independent = IndependentVariable(self.NUM_TRAIN_ROW)
        self.dependent = DependentVariable(self.independent.data, self.NUM_TRAIN_ROW)


# モデルを管理するクラス
class Model:
    model: Sequential

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

    def compile(self) -> None:
        # 結合層(2層->4層)
        self.model.add(Dense(4, input_dim=2, activation="tanh"))
        # 結合層(4層->1層):入力次元を省略すると自動的に前の層の出力次元数を引き継ぐ
        self.model.add(Dense(1, activation="sigmoid"))
        self.model.compile(
            loss="binary_crossentropy", optimizer="sgd", metrics=["accuracy"]
        )

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

    def train(
        self,
        independent: IndependentVariable,
        dependent: DependentVariable,
        epochs=300,
        validation_split=0.2,
        callbacks: Optional[list[ModelCheckpoint]] = [],
    ):
        self.model.fit(
            independent.data,
            dependent.labels,
            epochs=epochs,
            validation_split=validation_split,
            callbacks=callbacks,
        )

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


class Callback:
    @classmethod
    def save_model(cls, file_name: str = "./model.hdf5") -> ModelCheckpoint:
        return ModelCheckpoint(file_name, save_best_only=True)


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


class Test:
    NUM_TEST_ROW = 50

    def __init__(self, model: Model) -> None:
        self.model = model
        # テストデータは Independent, DependentVariable を使える
        self.independent = IndependentVariable(self.NUM_TEST_ROW)
        self.dependent = DependentVariable(self.independent.data, self.NUM_TEST_ROW)

    def run(self) -> Result:
        predict = ((self.model.model.predict(self.independent.data) > 0.5) * 1).reshape(
            self.NUM_TEST_ROW
        )
        # 50個中何個正解だったか
        accuracy = sum(predict == self.dependent.raw_labels) / self.NUM_TEST_ROW
        return Result(accuracy=accuracy)


if __name__ == "__main__":
    # テストデータ作成
    test_data = TestData()
    # モデル生成
    model = Model()
    model.compile()
    # モデルの訓練
    model.train(
        test_data.independent, test_data.dependent, callbacks=[Callback.save_model()]
    )
    model.save()
    # モデルの評価
    test = Test(model)
    result = test.run()
    print(result)

Model クラスに save メソッドを作成してそれを訓練後に呼び出すことでモデルを保存するようにしました

実行後に my_model フォルダが作成されその配下に各種モデル情報が保存されます

ls -l my_model/
total 136
drwxr-xr-x  2 hawk  staff     64 11  1 09:10 assets
-rw-r--r--  1 hawk  staff     56 11  1 09:10 fingerprint.pb
-rw-r--r--  1 hawk  staff   6537 11  1 09:10 keras_metadata.pb
-rw-r--r--  1 hawk  staff  56108 11  1 09:10 saved_model.pb
drwxr-xr-x  4 hawk  staff    128 11  1 09:10 variables

ちょっと解説

モデル全体を保存したい場合は SavedModel を使ったほうが良さそうです
callbacks を使う方法はチェックポイントを作成しているためどちらかというとモデルを丸々保存する用途というよりかは訓練を途中でやめて再度訓練を再開したい場合などに使う感じかなと思います

最後に

モデルを保存する方法を紹介しました
モデルを公開したり別のツールやアプリでモデルを使いたいときに利用する感じかなと思います

参考サイト