作者: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")
作為我們將要最小化的指標。目標名稱應與您在傳遞給回呼的 'on_epoch_end()' 方法的 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 入門指南。