概要
前回 の続きでモデルを保存する方法を学びました
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 を使う方法はチェックポイントを作成しているためどちらかというとモデルを丸々保存する用途というよりかは訓練を途中でやめて再度訓練を再開したい場合などに使う感じかなと思います
最後に
モデルを保存する方法を紹介しました
モデルを公開したり別のツールやアプリでモデルを使いたいときに利用する感じかなと思います
0 件のコメント:
コメントを投稿