在自定义训练循环中调整超参数
作者： Tom O'Malley, Haifeng Jin
创建日期 2019/10/28
最后修改日期 2022/01/12
描述： 使用 HyperModel.fit() 来调整训练超参数（例如批次大小）。
!pip install keras-tuner -q
简介
KerasTuner 中的 HyperModel 类提供了一种便捷的方式来在一个可重用对象中定义搜索空间。您可以重写 HyperModel.build() 来定义和超调模型本身。为了超调训练过程（例如，通过选择合适的批次大小、训练轮次数量或数据增强设置），您可以重写 HyperModel.fit()，在那里您可以访问
hp对象，它是keras_tuner.HyperParameters类的一个实例- 由
HyperModel.build()构建的模型
一个基本示例在 KerasTuner 入门指南 的“调整模型训练”部分有展示。
调整自定义训练循环
在本指南中，我们将继承 HyperModel 类，并通过重写 HyperModel.fit() 来编写自定义训练循环。关于如何使用 Keras 编写自定义训练循环，您可以参考指南 从头开始编写训练循环。
首先，我们导入所需的库，并创建用于训练和验证的数据集。这里，我们仅使用一些随机数据用于演示。
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))
然后，我们继承 HyperModel 类，命名为 MyHyperModel。在 MyHyperModel.build() 中，我们构建了一个简单的 Keras 模型来进行 10 种不同类别的图像分类。MyHyperModel.fit() 接受几个参数。其签名如下：
def fit(self, hp, model, x, y, validation_data, callbacks=None, **kwargs):
hp参数用于定义超参数。model参数是MyHyperModel.build()返回的模型。x,y, 和validation_data都是自定义定义的参数。稍后在调用tuner.search(x=x, y=y, validation_data=(x_val, y_val))时，我们将数据传递给它们。您可以定义任意数量的参数并赋予自定义名称。callbacks参数本打算与model.fit()一起使用。KerasTuner 在其中放置了一些有用的 Keras 回调，例如，用于在模型最佳轮次进行检查点的回调。
我们将在自定义训练循环中手动调用这些回调。在调用它们之前，我们需要通过以下代码将我们的模型分配给它们，以便它们可以访问模型进行检查点操作。
for callback in callbacks:
callback.model = model
在这个例子中，我们只调用了回调的 on_epoch_end() 方法来帮助我们进行模型检查点。如果需要，您也可以调用其他回调方法。如果您不需要保存模型，则不需要使用回调。
在自定义训练循环中，我们调整了数据集的批次大小，将 NumPy 数据包装成 tf.data.Dataset。请注意，您也可以在此处调整任何预处理步骤。我们还调整了优化器的学习率。
我们将使用验证损失作为模型的评估指标。为了计算平均验证损失，我们将使用 keras.metrics.Mean()，它会跨批次平均验证损失。我们需要返回验证损失供调优器记录。
class MyHyperModel(keras_tuner.HyperModel):
def build(self, hp):
"""Builds a convolutional model."""
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):
# Convert the datasets to 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
)
# Define the optimizer.
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)
# The metric to track validation loss.
epoch_loss_metric = keras.metrics.Mean()
# Function to run the train step.
@tf.function
def run_train_step(images, labels):
with tf.GradientTape() as tape:
logits = model(images)
loss = loss_fn(labels, logits)
# Add any regularization losses.
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))
# Function to run the validation step.
@tf.function
def run_val_step(images, labels):
logits = model(images)
loss = loss_fn(labels, logits)
# Update the metric.
epoch_loss_metric.update_state(loss)
# Assign the model to the callbacks.
for callback in callbacks:
callback.set_model(model)
# Record the best validation loss value
best_epoch_loss = float("inf")
# The custom training loop.
for epoch in range(2):
print(f"Epoch: {epoch}")
# Iterate the training data to run the training step.
for images, labels in train_ds:
run_train_step(images, labels)
# Iterate the validation data to run the validation step.
for images, labels in validation_data:
run_val_step(images, labels)
# Calling the callbacks after epoch.
epoch_loss = float(epoch_loss_metric.result().numpy())
for callback in callbacks:
# The "my_metric" is the objective passed to the tuner.
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 the evaluation metric value.
return best_epoch_loss
现在，我们可以初始化调优器。这里，我们使用 Objective("my_metric", "min") 作为我们要最小化的指标。目标名称应与您在传递给回调的 logs 中用作键的名称一致。回调需要使用 logs 中的此值来找到最佳轮次以进行模型检查点。
tuner = keras_tuner.RandomSearch(
objective=keras_tuner.Objective("my_metric", "min"),
max_trials=2,
hypermodel=MyHyperModel(),
directory="results",
project_name="custom_training",
overwrite=True,
)
我们通过将 MyHyperModel.fit() 签名中定义的参数传递给 tuner.search() 来开始搜索。
tuner.search(x=x_train, y=y_train, validation_data=(x_val, y_val))
Trial 2 Complete [00h 00m 02s]
my_metric: 2.3025283813476562
Best my_metric So Far: 2.3025283813476562
Total elapsed time: 00h 00m 04s
最后，我们可以检索结果。
best_hps = tuner.get_best_hyperparameters()[0]
print(best_hps.values)
best_model = tuner.get_best_models()[0]
best_model.summary()
{'units': 128, 'batch_size': 32, 'learning_rate': 0.0034272591820215972}
Model: "functional_1"
┏━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━┳━━━━━━━━━━━━━━━━━━━━━━━━━━━┳━━━━━━━━━━━━┓ ┃ Layer (type) ┃ Output Shape ┃ Param # ┃ ┡━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━╇━━━━━━━━━━━━━━━━━━━━━━━━━━━╇━━━━━━━━━━━━┩ │ input_layer (InputLayer) │ (None, 28, 28, 1) │ 0 │ ├─────────────────────────────────┼───────────────────────────┼────────────┤ │ flatten (Flatten) │ (None, 784) │ 0 │ ├─────────────────────────────────┼───────────────────────────┼────────────┤ │ dense (Dense) │ (None, 128) │ 100,480 │ ├─────────────────────────────────┼───────────────────────────┼────────────┤ │ dense_1 (Dense) │ (None, 10) │ 1,290 │ └─────────────────────────────────┴───────────────────────────┴────────────┘
Total params: 101,770 (397.54 KB)
Trainable params: 101,770 (397.54 KB)
Non-trainable params: 0 (0.00 B)
总之，要在自定义训练循环中调整超参数，您只需重写 HyperModel.fit() 来训练模型并返回评估结果。利用提供的回调，您可以轻松地在模型最佳轮次保存训练好的模型，并在之后加载最佳模型。
要了解更多关于 KerasTuner 的基础知识，请参阅 KerasTuner 入门指南。