Keras 模型的可重复性
作者： Frightera
创建日期 2023/05/05
最后修改日期 2023/05/05
描述：演示 Keras 模型中随机权重初始化和可重复性。
简介
本示例演示如何控制 Keras 模型中的随机性。有时，您可能希望在运行之间获得完全相同的结果，以进行实验或调试问题。
设置
import json
import numpy as np
import tensorflow as tf
import keras
from keras import layers
from keras import initializers
# Set the seed using keras.utils.set_random_seed. This will set:
# 1) `numpy` seed
# 2) backend random seed
# 3) `python` random seed
keras.utils.set_random_seed(812)
# If using TensorFlow, this will make GPU ops as deterministic as possible,
# but it will affect the overall performance, so be mindful of that.
tf.config.experimental.enable_op_determinism()
Keras 中的权重初始化
大多数 Keras 层都有 kernel_initializer 和 bias_initializer 参数。这些参数允许您指定用于初始化层变量权重的策略。以下内置初始化器可作为 keras.initializers 的一部分使用。
initializers_list = [
initializers.RandomNormal,
initializers.RandomUniform,
initializers.TruncatedNormal,
initializers.VarianceScaling,
initializers.GlorotNormal,
initializers.GlorotUniform,
initializers.HeNormal,
initializers.HeUniform,
initializers.LecunNormal,
initializers.LecunUniform,
initializers.Orthogonal,
]
在可重复的模型中，模型的权重应在后续运行中使用相同的值进行初始化。首先，我们将检查初始化器在使用相同的 seed 值多次调用时是如何表现的。
for initializer in initializers_list:
print(f"Running {initializer}")
for iteration in range(2):
# In order to get same results across multiple runs from an initializer,
# you can specify a seed value.
result = float(initializer(seed=42)(shape=(1, 1)))
print(f"\tIteration --> {iteration} // Result --> {result}")
print("\n")
Running <class 'keras.src.initializers.random_initializers.RandomNormal'>
Iteration --> 0 // Result --> 0.000790853810030967
Iteration --> 1 // Result --> 0.000790853810030967
Running <class 'keras.src.initializers.random_initializers.RandomUniform'>
Iteration --> 0 // Result --> -0.02175668440759182
Iteration --> 1 // Result --> -0.02175668440759182
Running <class 'keras.src.initializers.random_initializers.TruncatedNormal'>
Iteration --> 0 // Result --> 0.000790853810030967
Iteration --> 1 // Result --> 0.000790853810030967
Running <class 'keras.src.initializers.random_initializers.VarianceScaling'>
Iteration --> 0 // Result --> 0.017981600016355515
Iteration --> 1 // Result --> 0.017981600016355515
Running <class 'keras.src.initializers.random_initializers.GlorotNormal'>
Iteration --> 0 // Result --> 0.017981600016355515
Iteration --> 1 // Result --> 0.017981600016355515
Running <class 'keras.src.initializers.random_initializers.GlorotUniform'>
Iteration --> 0 // Result --> -0.7536736726760864
Iteration --> 1 // Result --> -0.7536736726760864
Running <class 'keras.src.initializers.random_initializers.HeNormal'>
Iteration --> 0 // Result --> 0.025429822504520416
Iteration --> 1 // Result --> 0.025429822504520416
Running <class 'keras.src.initializers.random_initializers.HeUniform'>
Iteration --> 0 // Result --> -1.065855622291565
Iteration --> 1 // Result --> -1.065855622291565
Running <class 'keras.src.initializers.random_initializers.LecunNormal'>
Iteration --> 0 // Result --> 0.017981600016355515
Iteration --> 1 // Result --> 0.017981600016355515
Running <class 'keras.src.initializers.random_initializers.LecunUniform'>
Iteration --> 0 // Result --> -0.7536736726760864
Iteration --> 1 // Result --> -0.7536736726760864
Running <class 'keras.src.initializers.random_initializers.OrthogonalInitializer'>
Iteration --> 0 // Result --> 1.0
Iteration --> 1 // Result --> 1.0
现在，让我们检查两个不同的初始化器对象在具有相同种子值时的行为。
# Setting the seed value for an initializer will cause two different objects
# to produce same results.
glorot_normal_1 = keras.initializers.GlorotNormal(seed=42)
glorot_normal_2 = keras.initializers.GlorotNormal(seed=42)
input_dim, neurons = 3, 5
# Call two different objects with same shape
result_1 = glorot_normal_1(shape=(input_dim, neurons))
result_2 = glorot_normal_2(shape=(input_dim, neurons))
# Check if the results are equal.
equal = np.allclose(result_1, result_2)
print(f"Are the results equal? {equal}")
Are the results equal? True
如果未设置种子值（或使用不同的种子值），则两个不同的对象将产生不同的结果。由于随机种子是在笔记本开始时设置的，因此结果将在后续运行中保持一致。这与 keras.utils.set_random_seed 相关。
glorot_normal_3 = keras.initializers.GlorotNormal()
glorot_normal_4 = keras.initializers.GlorotNormal()
# Let's call the initializer.
result_3 = glorot_normal_3(shape=(input_dim, neurons))
# Call the second initializer.
result_4 = glorot_normal_4(shape=(input_dim, neurons))
equal = np.allclose(result_3, result_4)
print(f"Are the results equal? {equal}")
Are the results equal? False
result_3 和 result_4 将不同，但当您再次运行笔记本时，result_3 将与之前的运行具有相同的值。result_4 也是如此。
模型训练过程中的可重复性
如果您想重现模型训练过程的结果，您需要在训练过程中控制随机性来源。为了展示一个真实的示例，本节将使用 tf.data，使用并行映射和 shuffle 操作。
为了开始，让我们创建一个简单的函数，它返回 Keras 模型的历史对象。
def train_model(train_data: tf.data.Dataset, test_data: tf.data.Dataset) -> dict:
model = keras.Sequential(
[
layers.Conv2D(32, (3, 3), activation="relu"),
layers.MaxPooling2D((2, 2)),
layers.Dropout(0.2),
layers.Conv2D(32, (3, 3), activation="relu"),
layers.MaxPooling2D((2, 2)),
layers.Dropout(0.2),
layers.Conv2D(32, (3, 3), activation="relu"),
layers.GlobalAveragePooling2D(),
layers.Dense(64, activation="relu"),
layers.Dropout(0.2),
layers.Dense(10, activation="softmax"),
]
)
model.compile(
optimizer="adam",
loss="sparse_categorical_crossentropy",
metrics=["accuracy"],
jit_compile=False,
)
# jit_compile's default value is "auto" which will cause some problems in some
# ops, therefore it's set to False.
# model.fit has a `shuffle` parameter which has a default value of `True`.
# If you are using array-like objects, this will shuffle the data before
# training. This argument is ignored when `x` is a generator or
# [`tf.data.Dataset`](https://tensorflowcn.cn/api_docs/python/tf/data/Dataset).
history = model.fit(train_data, epochs=2, validation_data=test_data)
print(f"Model accuracy on test data: {model.evaluate(test_data)[1] * 100:.2f}%")
return history.history
# Load the MNIST dataset
(train_images, train_labels), (
test_images,
test_labels,
) = keras.datasets.mnist.load_data()
# Construct tf.data.Dataset objects
train_ds = tf.data.Dataset.from_tensor_slices((train_images, train_labels))
test_ds = tf.data.Dataset.from_tensor_slices((test_images, test_labels))
Downloading data from https://storage.googleapis.com/tensorflow/tf-keras-datasets/mnist.npz
11490434/11490434 ━━━━━━━━━━━━━━━━━━━━ 0s 0us/step
请记住我们在函数开始时调用了 tf.config.experimental.enable_op_determinism()。这使得 tf.data 操作变得确定性。但是，使 tf.data 操作确定性会带来性能损失。如果您想了解更多信息，请查看此 官方指南。
这里简要概述一下发生了什么。模型具有 kernel_initializer 和 bias_initializer 参数。由于我们在笔记本开头使用 keras.utils.set_random_seed 设置了随机种子，因此初始化器将在后续运行中产生相同的结果。此外，TensorFlow 操作现在已成为确定性。通常，您将使用具有数千个硬件线程的 GPU，这会导致出现非确定性行为。
def prepare_dataset(image, label):
# Cast and normalize the image
image = tf.cast(image, tf.float32) / 255.0
# Expand the channel dimension
image = tf.expand_dims(image, axis=-1)
# Resize the image
image = tf.image.resize(image, (32, 32))
return image, label
tf.data.Dataset 对象有一个 shuffle 方法，用于对数据进行混洗。此方法有一个 buffer_size 参数，用于控制缓冲区的尺寸。如果您将此值设置为 len(train_images)，则整个数据集将被混洗。如果缓冲区大小等于数据集的长度，则元素将以完全随机的顺序进行混洗。
将缓冲区大小设置为数据集长度的主要缺点是，填充缓冲区可能需要一段时间，具体取决于数据集的大小。
这里简要概述一下发生了什么：1) shuffle() 方法创建一个指定大小的缓冲区。2) 数据集的元素被随机混洗并放入缓冲区。3) 然后从缓冲区中以随机顺序返回元素。
由于 tf.config.experimental.enable_op_determinism() 已启用，并且我们在笔记本开头使用 keras.utils.set_random_seed 设置了随机种子，因此 shuffle() 方法将在后续运行中产生相同的结果。
# Prepare the datasets, batch-map --> vectorized operations
train_data = (
train_ds.shuffle(buffer_size=len(train_images))
.batch(batch_size=64)
.map(prepare_dataset, num_parallel_calls=tf.data.AUTOTUNE)
.prefetch(buffer_size=tf.data.AUTOTUNE)
)
test_data = (
test_ds.batch(batch_size=64)
.map(prepare_dataset, num_parallel_calls=tf.data.AUTOTUNE)
.prefetch(buffer_size=tf.data.AUTOTUNE)
)
首次训练模型。
history = train_model(train_data, test_data)
Epoch 1/2
938/938 ━━━━━━━━━━━━━━━━━━━━ 73s 73ms/step - accuracy: 0.5726 - loss: 1.2175 - val_accuracy: 0.9401 - val_loss: 0.1924
Epoch 2/2
938/938 ━━━━━━━━━━━━━━━━━━━━ 89s 81ms/step - accuracy: 0.9105 - loss: 0.2885 - val_accuracy: 0.9630 - val_loss: 0.1131
157/157 ━━━━━━━━━━━━━━━━━━━━ 3s 17ms/step - accuracy: 0.9553 - loss: 0.1353
Model accuracy on test data: 96.30%
让我们将结果保存到 JSON 文件中，然后重新启动内核。重新启动内核后，我们应该看到与之前运行相同的结果，包括训练和测试数据的指标和损失值。
# Save the history object into a json file
with open("history.json", "w") as fp:
json.dump(history, fp)
为了不覆盖结果，请不要运行上面的单元格。再次执行模型训练单元格并比较结果。
with open("history.json", "r") as fp:
history_loaded = json.load(fp)
逐一比较结果。您会发现它们是相等的。
for key in history.keys():
for i in range(len(history[key])):
if not np.allclose(history[key][i], history_loaded[key][i]):
print(f"{key} not equal")
结论
在本教程中，您学习了如何在 Keras 和 TensorFlow 中控制随机性来源。您还学习了如何重现模型训练过程的结果。
如果您想每次都使用相同的权重初始化模型，则需要设置层的 kernel_initializer 和 bias_initializer 参数，并为初始化器提供 seed 值。
由于数值误差累积，例如在 RNN 层中使用 recurrent_dropout，仍然可能存在一些不一致。
可重复性取决于环境。如果您在具有相同环境的同一台机器上运行笔记本或代码，您将获得相同的结果。