在自定义训练循环中调整超参数
作者：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 回调，例如，在最佳 epoch 处检查模型的回调。
我们将在自定义训练循环中手动调用回调。在调用它们之前，我们需要使用以下代码将模型分配给它们，以便它们可以访问该模型进行检查。
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") 作为我们要最小化的指标。目标名称应与您在传递给回调的 'on_epoch_end()' 方法的 logs 中用作键的名称一致。回调需要使用 logs 中的此值来查找检查模型的最佳 epoch。
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() 来训练模型并返回评估结果。使用提供的回调，您可以轻松地在最佳 epoch 保存训练的模型，并在以后加载最佳模型。
要了解有关 KerasTuner 的基本知识，请参阅 KerasTuner 入门指南。