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
如果未设置种子值（或使用不同的种子值），两个不同的对象将产生不同的结果。由于在 Notebook 的开头设置了随机种子，因此在顺序运行中结果将是相同的。这与 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 将不同，但当您再次运行 Notebook 时，result_3 将与上次运行中的值完全相同。result_4 也是如此。
模型训练过程中的复现性
如果您想复现模型训练过程的结果，则需要在训练过程中控制随机性源。为了展示一个实际示例，本节将使用并行映射和 shuffle 操作来利用 tf.data。
首先，让我们创建一个返回 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 参数。由于我们在 Notebook 的开头使用 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() 并且我们在 Notebook 的开头使用 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，可能仍然存在一些不一致性。
复现性取决于环境。如果您在具有相同环境的同一台机器上运行 Notebook 或代码，您将获得相同的结果。