代码示例 / Keras 快速教程 / Keras 模型的可复现性

Keras 模型的可复现性

作者: Frightera
创建时间 2023/05/05
上次修改时间 2023/05/05
描述:演示 Keras 模型中的随机权重初始化和可复现性。

ⓘ 此示例使用 Keras 3

在 Colab 中查看 GitHub 源码


简介

此示例演示如何在 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_initializerbias_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_3result_4 将不同,但当您再次运行笔记本时,result_3 将具有与先前运行中相同的值。result_4 也是如此。


模型训练过程中的可复现性

如果要重现模型训练过程的结果,则需要在训练过程中控制随机性来源。为了展示一个真实的例子,本节利用了使用并行映射和混洗操作的 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_initializerbias_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_initializerbias_initializer 参数,并向初始化器提供 seed 值。

由于数值误差累积(例如在 RNN 层中使用 recurrent_dropout)等原因,仍然可能存在一些不一致。

可复现性取决于环境。如果您在具有相同环境的同一台机器上运行笔记本或代码,则将获得相同的结果。