당신의 커스텀 트레이닝 루프에서 하이퍼파라미터 튜닝

!pip install keras-tuner -q

import keras_tuner
import tensorflow as tf
import keras
import numpy as np


x_train = np.random.rand(1000, 28, 28, 1)
y_train = np.random.randint(0, 10, (1000, 1))
x_val = np.random.rand(1000, 28, 28, 1)
y_val = np.random.randint(0, 10, (1000, 1))

def fit(self, hp, model, x, y, validation_data, callbacks=None, **kwargs):

for callback in callbacks:
    callback.model = model

class MyHyperModel(keras_tuner.HyperModel):
    def build(self, hp):
        """컨볼루션 모델을 빌드합니다."""
        inputs = keras.Input(shape=(28, 28, 1))
        x = keras.layers.Flatten()(inputs)
        x = keras.layers.Dense(
            units=hp.Choice("units", [32, 64, 128]), activation="relu"
        )(x)
        outputs = keras.layers.Dense(10)(x)
        return keras.Model(inputs=inputs, outputs=outputs)

    def fit(self, hp, model, x, y, validation_data, callbacks=None, **kwargs):
        # 데이터셋을 tf.data.Dataset으로 변환합니다.
        batch_size = hp.Int("batch_size", 32, 128, step=32, default=64)
        train_ds = tf.data.Dataset.from_tensor_slices((x_train, y_train)).batch(
            batch_size
        )
        validation_data = tf.data.Dataset.from_tensor_slices(validation_data).batch(
            batch_size
        )

        # 옵티마이저 정의.
        optimizer = keras.optimizers.Adam(
            hp.Float("learning_rate", 1e-4, 1e-2, sampling="log", default=1e-3)
        )
        loss_fn = keras.losses.SparseCategoricalCrossentropy(from_logits=True)

        # 검증 손실을 추적할 메트릭.
        epoch_loss_metric = keras.metrics.Mean()

        # 트레이닝 단계를 실행하는 함수.
        @tf.function
        def run_train_step(images, labels):
            with tf.GradientTape() as tape:
                logits = model(images)
                loss = loss_fn(labels, logits)
                # 정규화 손실 추가.
                if model.losses:
                    loss += tf.math.add_n(model.losses)
            gradients = tape.gradient(loss, model.trainable_variables)
            optimizer.apply_gradients(zip(gradients, model.trainable_variables))

        # 검증 단계를 실행하는 함수.
        @tf.function
        def run_val_step(images, labels):
            logits = model(images)
            loss = loss_fn(labels, logits)
            # 메트릭 업데이트.
            epoch_loss_metric.update_state(loss)

        # 모델을 콜백에 할당합니다.
        for callback in callbacks:
            callback.set_model(model)

        # 최상의 검증 손실 값을 기록.
        best_epoch_loss = float("inf")

        # 커스텀 트레이닝 루프.
        for epoch in range(2):
            print(f"Epoch: {epoch}")

            # 트레이닝 데이터를 반복하면서, 트레이닝 단계를 실행.
            for images, labels in train_ds:
                run_train_step(images, labels)

            # 검증 데이터를 반복하면서, 검증 단계를 실행.
            for images, labels in validation_data:
                run_val_step(images, labels)

            # 에포크가 끝난 후 콜백 호출.
            epoch_loss = float(epoch_loss_metric.result().numpy())
            for callback in callbacks:
                # "my_metric"은 튜너에 전달된 objective 입니다.
                callback.on_epoch_end(epoch, logs={"my_metric": epoch_loss})
            epoch_loss_metric.reset_state()

            print(f"Epoch loss: {epoch_loss}")
            best_epoch_loss = min(best_epoch_loss, epoch_loss)

        # 평가 메트릭 값을 반환.
        return best_epoch_loss

tuner = keras_tuner.RandomSearch(
    objective=keras_tuner.Objective("my_metric", "min"),
    max_trials=2,
    hypermodel=MyHyperModel(),
    directory="results",
    project_name="custom_training",
    overwrite=True,
)

tuner.search(x=x_train, y=y_train, validation_data=(x_val, y_val))

best_hps = tuner.get_best_hyperparameters()[0]
print(best_hps.values)

best_model = tuner.get_best_models()[0]
best_model.summary()