► 开发者指南 / 使用内置方法进行训练与评估

使用内置方法进行训练与评估

作者: fchollet
创建日期: 2019/03/01
最后修改: 2023/06/25
描述: 使用 fit() 和 evaluate() 的完整训练与评估指南。

设置

# 我们导入torch和TF，以便使用torch Dataloaders和tf.data.Datasets。
import torch
import tensorflow as tf

import os
import numpy as np
import keras
from keras import layers
from keras import ops

介绍

本指南涵盖了使用内置API进行训练、评估和预测（推断）模型的内容（如 Model.fit()，Model.evaluate() 和 Model.predict()）。

如果您有兴趣在指定自己的训练步骤函数时利用 fit()，请参阅有关自定义 fit() 中发生的内容的指南：

如果您有兴趣从头开始编写自己的训练与评估循环，请参阅有关编写训练循环的指南：

一般来说，无论您是使用内置循环还是编写自己的循环，模型训练与评估在每种类型的Keras模型中严格以相同的方式进行—— 序列模型、使用功能API构建的模型，以及通过模型子类化编写的模型。

API 概述：第一个端到端示例

在将数据传递给模型的内置训练循环时，您应该使用：

NumPy数组（如果您的数据较小并且可以放入内存中）
keras.utils.PyDataset 的子类
tf.data.Dataset 对象
PyTorch DataLoader 实例

在接下来的几个段落中，我们将使用MNIST数据集作为NumPy数组，以演示如何使用优化器、损失和指标。之后，我们将仔细查看其他选项。

我们考虑以下模型（在这里，我们使用功能API构建，但它也可以是序列模型或子类化模型）：

inputs = keras.Input(shape=(784,), name="digits")
x = layers.Dense(64, activation="relu", name="dense_1")(inputs)
x = layers.Dense(64, activation="relu", name="dense_2")(x)
outputs = layers.Dense(10, activation="softmax", name="predictions")(x)

model = keras.Model(inputs=inputs, outputs=outputs)

典型的端到端工作流如下，包括：

训练
在从原始训练数据生成的保留集上进行验证
在测试数据上进行评估

我们将使用MNIST数据作为这个示例。

(x_train, y_train), (x_test, y_test) = keras.datasets.mnist.load_data()

# 预处理数据（这些是NumPy数组）
x_train = x_train.reshape(60000, 784).astype("float32") / 255
x_test = x_test.reshape(10000, 784).astype("float32") / 255

y_train = y_train.astype("float32")
y_test = y_test.astype("float32")

# 为验证保留10,000个样本
x_val = x_train[-10000:]
y_val = y_train[-10000:]
x_train = x_train[:-10000]
y_train = y_train[:-10000]

我们指定训练配置（优化器、损失、指标）：

model.compile(
    optimizer=keras.optimizers.RMSprop(),  # 优化器
    # 要最小化的损失函数
    loss=keras.losses.SparseCategoricalCrossentropy(),
    # 监控的指标列表
    metrics=[keras.metrics.SparseCategoricalAccuracy()],
)

我们调用 fit()，它将通过将数据切片为大小为 batch_size 的“批次”，并在给定的 epochs 次数上重复遍历整个数据集来训练模型。

print("在训练数据上拟合模型")
history = model.fit(
    x_train,
    y_train,
    batch_size=64,
    epochs=2,
    # 我们传递一些验证信息以监控
    # 在每个epoch结束时的验证损失和指标
    validation_data=(x_val, y_val),
)

在训练数据上拟合模型
第 1/2 轮次
 782/782 ━━━━━━━━━━━━━━━━━━━━ 1s 955us/步 - 损失: 0.5740 - 稀疏分类准确率: 0.8368 - 验证损失: 0.2040 - 验证稀疏分类准确率: 0.9420
第 2/2 轮次
 782/782 ━━━━━━━━━━━━━━━━━━━━ 0s 390us/步 - 损失: 0.1745 - 稀疏分类准确率: 0.9492 - 验证损失: 0.1415 - 验证稀疏分类准确率: 0.9581

返回的 history 对象记录了训练过程中的损失值和指标值：

print(history.history)

{'loss': [0.34448376297950745, 0.16419583559036255], 'sparse_categorical_accuracy': [0.9008600115776062, 0.9509199857711792], 'val_loss': [0.20404714345932007, 0.14145156741142273], 'val_sparse_categorical_accuracy': [0.9419999718666077, 0.9581000208854675]}

我们通过 evaluate() 在测试数据上评估模型：

# 使用 `evaluate` 在测试数据上评估模型
print("Evaluate on test data")
results = model.evaluate(x_test, y_test, batch_size=128)
print("test loss, test acc:", results)

# 使用 `predict` 在新数据上生成预测（概率 -- 最后一个层的输出）
print("Generate predictions for 3 samples")
predictions = model.predict(x_test[:3])
print("predictions shape:", predictions.shape)

Evaluate on test data
 79/79 ━━━━━━━━━━━━━━━━━━━━ 0s 271us/step - loss: 0.1670 - sparse_categorical_accuracy: 0.9489
test loss, test acc: [0.1484374850988388, 0.9550999999046326]
Generate predictions for 3 samples
 1/1 ━━━━━━━━━━━━━━━━━━━━ 0s 33ms/step
predictions shape: (3, 10)

现在，让我们详细回顾这个工作流程的每个部分。

`compile()` 方法：指定损失、指标和优化器

要使用 fit() 训练模型，您需要指定一个损失函数、一个优化器，以及可选的监控指标。

您将这些作为参数传递给模型的 compile() 方法：

model.compile(
    optimizer=keras.optimizers.RMSprop(learning_rate=1e-3),
    loss=keras.losses.SparseCategoricalCrossentropy(),
    metrics=[keras.metrics.SparseCategoricalAccuracy()],
)

metrics 参数应该是一个列表 — 您的模型可以有任意数量的指标。

如果您的模型有多个输出，您可以为每个输出指定不同的损失和指标，您还可以调节每个输出对模型总损失的贡献。您可以在 将数据传递给多输入、多输出模型 部分找到更多细节。

请注意，如果您对默认设置满意，在许多情况下，优化器、损失和指标可以通过字符串标识符作为快捷方式指定：

model.compile(
    optimizer="rmsprop",
    loss="sparse_categorical_crossentropy",
    metrics=["sparse_categorical_accuracy"],
)

为了方便后续重用，让我们将模型定义和编译步骤放入函数中；我们将在本指南的不同示例中多次调用它们。

def get_uncompiled_model():
    inputs = keras.Input(shape=(784,), name="digits")
    x = layers.Dense(64, activation="relu", name="dense_1")(inputs)
    x = layers.Dense(64, activation="relu", name="dense_2")(x)
    outputs = layers.Dense(10, activation="softmax", name="predictions")(x)
    model = keras.Model(inputs=inputs, outputs=outputs)
    return model


def get_compiled_model():
    model = get_uncompiled_model()
    model.compile(
        optimizer="rmsprop",
        loss="sparse_categorical_crossentropy",
        metrics=["sparse_categorical_accuracy"],
    )
    return model

有许多内置的优化器、损失和指标可用

一般来说，您不需要从头创建自己的损失、指标或优化器，因为您所需要的很可能已经是 Keras API 的一部分：

优化器：

SGD()（有或没有动量）
RMSprop()
Adam()
等等。

损失：

MeanSquaredError()
KLDivergence()
CosineSimilarity()
等等。

指标：

AUC()
Precision()
Recall()
等等。

自定义损失

如果您需要创建自定义损失，Keras 提供三种方法来做到这一点。

第一种方法是创建一个接受输入 y_true 和 y_pred 的函数。以下示例显示了一个计算真实数据与预测之间均方误差的损失函数：

def custom_mean_squared_error(y_true, y_pred):
    return ops.mean(ops.square(y_true - y_pred), axis=-1)


model = get_uncompiled_model()
model.compile(optimizer=keras.optimizers.Adam(), loss=custom_mean_squared_error)

# 我们需要将标签进行独热编码以使用 MSE
y_train_one_hot = ops.one_hot(y_train, num_classes=10)
model.fit(x_train, y_train_one_hot, batch_size=64, epochs=1)

 782/782 ━━━━━━━━━━━━━━━━━━━━ 1s 525us/step - loss: 0.0277

<keras.src.callbacks.history.History at 0x2e5dde350>

如果您需要一个损失函数，它接受除了 y_true 和 y_pred 之外的参数，您可以子类化 keras.losses.Loss 类并实现以下两个方法：

__init__(self)：接受在调用您的损失函数时传递的参数
call(self, y_true, y_pred)：使用目标（y_true）和模型预测（y_pred）计算模型的损失假设你想使用均方误差（mean squared error），但添加一个额外项，以降低预测值远离0.5的情况（我们假设分类目标是独热编码，值在0和1之间）。这为模型创造了不那么自信的激励，这可能有助于减少过拟合（我们在尝试之前无法知道它是否有效！）。

以下是你可以这样做的方式：

class CustomMSE(keras.losses.Loss):
    def __init__(self, regularization_factor=0.1, name="custom_mse"):
        super().__init__(name=name)
        self.regularization_factor = regularization_factor

    def call(self, y_true, y_pred):
        mse = ops.mean(ops.square(y_true - y_pred), axis=-1)
        reg = ops.mean(ops.square(0.5 - y_pred), axis=-1)
        return mse + reg * self.regularization_factor


model = get_uncompiled_model()
model.compile(optimizer=keras.optimizers.Adam(), loss=CustomMSE())

y_train_one_hot = ops.one_hot(y_train, num_classes=10)
model.fit(x_train, y_train_one_hot, batch_size=64, epochs=1)

 782/782 ━━━━━━━━━━━━━━━━━━━━ 1s 532us/step - loss: 0.0492

<keras.src.callbacks.history.History at 0x2e5d0d360>

自定义指标

如果你需要一个不在API中的指标，可以通过继承keras.metrics.Metric类轻松创建自定义指标。你需要实现4个方法：

__init__(self)，在此方法中你将为你的指标创建状态变量。
update_state(self, y_true, y_pred, sample_weight=None)，该方法使用目标y_true和模型预测y_pred来更新状态变量。
result(self)，该方法使用状态变量计算最终结果。
reset_state(self)，该方法重新初始化指标的状态。

状态更新和结果计算被分开保留（在update_state()和result()中），因为在某些情况下，结果计算可能非常昂贵，并且只会定期进行。

以下是一个简单的示例，展示如何实现一个CategoricalTruePositives指标，用于计算多少样本被正确分类为属于给定类别：

class CategoricalTruePositives(keras.metrics.Metric):
    def __init__(self, name="categorical_true_positives", **kwargs):
        super().__init__(name=name, **kwargs)
        self.true_positives = self.add_variable(
            shape=(), name="ctp", initializer="zeros"
        )

    def update_state(self, y_true, y_pred, sample_weight=None):
        y_pred = ops.reshape(ops.argmax(y_pred, axis=1), (-1, 1))
        values = ops.cast(y_true, "int32") == ops.cast(y_pred, "int32")
        values = ops.cast(values, "float32")
        if sample_weight is not None:
            sample_weight = ops.cast(sample_weight, "float32")
            values = ops.multiply(values, sample_weight)
        self.true_positives.assign_add(ops.sum(values))

    def result(self):
        return self.true_positives.value

    def reset_state(self):
        # 指标的状态将在每个epoch开始时重置。
        self.true_positives.assign(0.0)


model = get_uncompiled_model()
model.compile(
    optimizer=keras.optimizers.RMSprop(learning_rate=1e-3),
    loss=keras.losses.SparseCategoricalCrossentropy(),
    metrics=[CategoricalTruePositives()],
)
model.fit(x_train, y_train, batch_size=64, epochs=3)

Epoch 1/3
 782/782 ━━━━━━━━━━━━━━━━━━━━ 1s 568us/step - categorical_true_positives: 180967.9219 - loss: 0.5876
Epoch 2/3
 782/782 ━━━━━━━━━━━━━━━━━━━━ 0s 377us/step - categorical_true_positives: 182141.9375 - loss: 0.1733
Epoch 3/3
 782/782 ━━━━━━━━━━━━━━━━━━━━ 0s 377us/step - categorical_true_positives: 182303.5312 - loss: 0.1180

<keras.src.callbacks.history.History at 0x2e5f02d10>

处理不符合标准签名的损失和指标

绝大多数的损失和指标可以从y_true和y_pred中计算，其中y_pred是模型的输出——但并非全部。例如，正则化损失可能只需要一个层的激活（在这种情况下没有目标），而这种激活可能不是模型输出。

在这种情况下，你可以在自定义层的调用方法中调用self.add_loss(loss_value)。以这种方式添加的损失将被添加到训练期间的“主”损失中（传递给compile()的那个）。以下是一个简单的示例，展示如何添加活动正则化（请注意，所有Keras层内置了活动正则化——此层仅用于提供一个具体的示例）：

class ActivityRegularizationLayer(layers.Layer):
    def call(self, inputs):
        self.add_loss(ops.sum(inputs) * 0.1)
        return inputs  # 透传层.


inputs = keras.Input(shape=(784,), name="digits")
x = layers.Dense(64, activation="relu", name="dense_1")(inputs)

# 将活动正则化插入为一层
x = ActivityRegularizationLayer()(x)

x = layers.Dense(64, activation="relu", name="dense_2")(x)
outputs = layers.Dense(10, name="predictions")(x)

model = keras.Model(inputs=inputs, outputs=outputs)
model.compile(
    optimizer=keras.optimizers.RMSprop(learning_rate=1e-3),
    loss=keras.losses.SparseCategoricalCrossentropy(from_logits=True),
)

# 显示的损失将比之前高得多
# 这是由于正则化组件的影响。
model.fit(x_train, y_train, batch_size=64, epochs=1)

 782/782 ━━━━━━━━━━━━━━━━━━━━ 1s 505us/step - loss: 3.4083

<keras.src.callbacks.history.History at 0x2e60226b0>

注意，当你通过 add_loss() 传递损失值时，可以在没有损失函数的情况下调用 compile()，因为模型已经有一个损失值要最小化。

考虑以下 LogisticEndpoint 层：它接收目标和 logits 作为输入，并通过 add_loss() 跟踪交叉熵损失。

class LogisticEndpoint(keras.layers.Layer):
    def __init__(self, name=None):
        super().__init__(name=name)
        self.loss_fn = keras.losses.BinaryCrossentropy(from_logits=True)

    def call(self, targets, logits, sample_weights=None):
        # 计算训练时的损失值并将其添加到
        # 层中，使用 `self.add_loss()`。
        loss = self.loss_fn(targets, logits, sample_weights)
        self.add_loss(loss)

        # 返回推断时的预测张量（用于 `.predict()`）。
        return ops.softmax(logits)

你可以在一个具有两个输入（输入数据和目标）的模型中使用它，编译时不带 loss 参数，如下所示：

inputs = keras.Input(shape=(3,), name="inputs")
targets = keras.Input(shape=(10,), name="targets")
logits = keras.layers.Dense(10)(inputs)
predictions = LogisticEndpoint(name="predictions")(targets, logits)

model = keras.Model(inputs=[inputs, targets], outputs=predictions)
model.compile(optimizer="adam")  # 没有损失参数！

data = {
    "inputs": np.random.random((3, 3)),
    "targets": np.random.random((3, 10)),
}
model.fit(data)

 1/1 ━━━━━━━━━━━━━━━━━━━━ 0s 89ms/step - loss: 0.6982

<keras.src.callbacks.history.History at 0x2e5cc91e0>

有关训练多输入模型的更多信息，请参见 将数据传递给多输入、多输出模型 部分。

自动划分验证保留集

在你所看到的第一个端到端示例中，我们使用 validation_data 参数将 NumPy 数组的元组 (x_val, y_val) 传递给模型，以便在每个周期结束时评估验证损失和验证指标。

这是另一个选项：参数 validation_split 允许你自动保留一部分训练数据用于验证。参数值表示要保留用于验证的数据的比例，因此它应该设置为大于 0 且小于 1 的数字。例如，validation_split=0.2 意味着“使用 20% 的数据进行验证”，而 validation_split=0.6 则意味着“使用 60% 的数据进行验证”。

验证计算的方式是通过取 fit() 调用接收到的数组的最后 x% 样本，在任何洗牌之前。

请注意，只有在使用 NumPy 数据进行训练时，你才能使用 validation_split。

model = get_compiled_model()
model.fit(x_train, y_train, batch_size=64, validation_split=0.2, epochs=1)

 625/625 ━━━━━━━━━━━━━━━━━━━━ 1s 563us/step - loss: 0.6161 - sparse_categorical_accuracy: 0.8259 - val_loss: 0.2379 - val_sparse_categorical_accuracy: 0.9302

<keras.src.callbacks.history.History at 0x2e6007610>

使用 `tf.data` 数据集进行训练与评估

在过去的几段中，你已经看到了如何处理损失、指标和优化器，以及如何在数据作为 NumPy 数组传递时使用 validation_data 和 validation_split 参数。

另一种选择是使用类似迭代器的方式，比如 tf.data.Dataset、PyTorch DataLoader 或 Keras PyDataset。让我们看看前者。

tf.data API 是 TensorFlow 2.0 中的一组实用工具，用于以快速和可扩展的方式加载和预处理数据。有关创建 Datasets 的完整指南，请参见 tf.data 文档。

你可以使用 tf.data 来训练你的 Keras 模型，无论你使用的是哪个后端 – 无论是 JAX、PyTorch 还是 TensorFlow。 你可以将 Dataset 实例直接传递给 fit()、evaluate() 和 predict() 方法：

model = get_compiled_model()

# 首先，我们创建一个训练数据集实例。
# 为了我们的示例，我们将使用与之前相同的 MNIST 数据。
train_dataset = tf.data.Dataset.from_tensor_slices((x_train, y_train))
# 随机打乱并切片数据集。
train_dataset = train_dataset.shuffle(buffer_size=1024).batch(64)

# 现在我们获取一个测试数据集。
test_dataset = tf.data.Dataset.from_tensor_slices((x_test, y_test))
test_dataset = test_dataset.batch(64)

# 由于数据集已经处理了批量化，
# 我们不传递 `batch_size` 参数。
model.fit(train_dataset, epochs=3)

# 你也可以在数据集上进行评估或预测。
print("Evaluate")
result = model.evaluate(test_dataset)
dict(zip(model.metrics_names, result))

Epoch 1/3
 782/782 ━━━━━━━━━━━━━━━━━━━━ 1s 688us/step - loss: 0.5631 - sparse_categorical_accuracy: 0.8458
Epoch 2/3
 782/782 ━━━━━━━━━━━━━━━━━━━━ 0s 512us/step - loss: 0.1703 - sparse_categorical_accuracy: 0.9484
Epoch 3/3
 782/782 ━━━━━━━━━━━━━━━━━━━━ 0s 506us/step - loss: 0.1187 - sparse_categorical_accuracy: 0.9640
Evaluate
 157/157 ━━━━━━━━━━━━━━━━━━━━ 0s 622us/step - loss: 0.1380 - sparse_categorical_accuracy: 0.9582

{'loss': 0.11913617700338364, 'compile_metrics': 0.965399980545044}

注意，数据集在每个 epoch 结束时会重置，因此可以在下一个 epoch 中重复使用。

如果您只想在该数据集中运行特定数量的批次进行训练，可以传递 steps_per_epoch 参数，该参数指定模型在进行到下一个 epoch 之前，应该使用该数据集运行多少个训练步骤。

model = get_compiled_model()

# 准备训练数据集
train_dataset = tf.data.Dataset.from_tensor_slices((x_train, y_train))
train_dataset = train_dataset.shuffle(buffer_size=1024).batch(64)

# 每个 epoch 只使用 100 个批次（共 64 * 100 个样本）
model.fit(train_dataset, epochs=3, steps_per_epoch=100)

Epoch 1/3
 100/100 ━━━━━━━━━━━━━━━━━━━━ 0s 508us/step - loss: 1.2000 - sparse_categorical_accuracy: 0.6822 
Epoch 2/3
 100/100 ━━━━━━━━━━━━━━━━━━━━ 0s 481us/step - loss: 0.4004 - sparse_categorical_accuracy: 0.8827
Epoch 3/3
 100/100 ━━━━━━━━━━━━━━━━━━━━ 0s 471us/step - loss: 0.3546 - sparse_categorical_accuracy: 0.8968

<keras.src.callbacks.history.History at 0x2e64df400>

您还可以将 Dataset 实例作为 fit() 中的 validation_data 参数传递：

model = get_compiled_model()

# 准备训练数据集
train_dataset = tf.data.Dataset.from_tensor_slices((x_train, y_train))
train_dataset = train_dataset.shuffle(buffer_size=1024).batch(64)

# 准备验证数据集
val_dataset = tf.data.Dataset.from_tensor_slices((x_val, y_val))
val_dataset = val_dataset.batch(64)

model.fit(train_dataset, epochs=1, validation_data=val_dataset)

 782/782 ━━━━━━━━━━━━━━━━━━━━ 1s 837us/step - loss: 0.5569 - sparse_categorical_accuracy: 0.8508 - val_loss: 0.1711 - val_sparse_categorical_accuracy: 0.9527

<keras.src.callbacks.history.History at 0x2e641e920>

在每个 epoch 结束时，模型将迭代验证数据集并计算验证损失和验证指标。

如果您只想在此数据集中运行特定数量的批次进行验证，您可以传递 validation_steps 参数，该参数指定模型应该在中断验证并继续到下一个 epoch 之前，使用验证数据集运行多少个验证步骤：

model = get_compiled_model()

# 准备训练数据集
train_dataset = tf.data.Dataset.from_tensor_slices((x_train, y_train))
train_dataset = train_dataset.shuffle(buffer_size=1024).batch(64)

# 准备验证数据集
val_dataset = tf.data.Dataset.from_tensor_slices((x_val, y_val))
val_dataset = val_dataset.batch(64)

model.fit(
    train_dataset,
    epochs=1,
    # 仅使用数据集的前 10 个批次运行验证
    # 使用 `validation_steps` 参数
    validation_data=val_dataset,
    validation_steps=10,
)

 782/782 ━━━━━━━━━━━━━━━━━━━━ 1s 771us/step - loss: 0.5562 - sparse_categorical_accuracy: 0.8436 - val_loss: 0.3345 - val_sparse_categorical_accuracy: 0.9062

<keras.src.callbacks.history.History at 0x2f9542e00>

注意，验证数据集在每次使用后会重置（这样您将始终在每个 epoch 上评估相同的样本）。

在使用 Dataset 对象进行训练时，不支持 validation_split 参数（从训练数据中生成保留集），因为此功能需要能够索引数据集的样本，而在一般情况下，使用 Dataset API 是不可能的。

使用 `PyDataset` 实例进行训练和评估

keras.utils.PyDataset 是一个实用工具，您可以通过子类化它来获得具有两个重要属性的 Python 生成器：

它在多进程中表现良好。
它可以进行洗牌（例如，在 fit() 中传递 shuffle=True 时）。

PyDataset 必须实现两个方法：

__getitem__
__len__

__getitem__ 方法应返回一个完整的批次。如果您想在每个 epoch 之间修改数据集，可以实现 on_epoch_end。

这是一个简单的示例：

class ExamplePyDataset(keras.utils.PyDataset):
    def __init__(self, x, y, batch_size, **kwargs):
        super().__init__(**kwargs)
        self.x = x
        self.y = y
        self.batch_size = batch_size

    def __len__(self):
        return int(np.ceil(len(self.x) / float(self.batch_size)))

    def __getitem__(self, idx):
        batch_x = self.x[idx * self.batch_size : (idx + 1) * self.batch_size]
        batch_y = self.y[idx * self.batch_size : (idx + 1) * self.batch_size]
        return batch_x, batch_y


train_py_dataset = ExamplePyDataset(x_train, y_train, batch_size=32)
val_py_dataset = ExamplePyDataset(x_val, y_val, batch_size=32)

要拟合模型，作为 x 参数传递数据集（不需要 y 参数，因为数据集包括目标），并将验证数据集作为 validation_data 参数传递。而且不需要传递 batch_size 参数，因为数据集已经被批处理了！

model = get_compiled_model()
model.fit(train_py_dataset, batch_size=64, validation_data=val_py_dataset, epochs=1)

 1563/1563 ━━━━━━━━━━━━━━━━━━━━ 1s 443us/step - loss: 0.5217 - sparse_categorical_accuracy: 0.8473 - val_loss: 0.1576 - val_sparse_categorical_accuracy: 0.9525

<keras.src.callbacks.history.History at 0x2f9c8d120>

评估模型同样简单：

model.evaluate(val_py_dataset)

 313/313 ━━━━━━━━━━━━━━━━━━━━ 0s 157us/step - loss: 0.1821 - sparse_categorical_accuracy: 0.9450

[0.15764616429805756, 0.9524999856948853]

重要的是，PyDataset 对象支持处理并行处理配置的三个常见构造函数参数：

workers：用于多线程或多进程的工作线程数量。通常，您会将其设置为 CPU 上的核心数量。
use_multiprocessing：是否使用 Python 多进程进行并行处理。将其设置为 True 意味着您的数据集将在多个分叉的进程中被复制。这是获得计算级别（而非 I/O 级别）并行处理好处的必要条件。但是，只有在您的数据集可以安全地进行序列化时，才能设置为 True。
max_queue_size：在多线程或多进程设置中迭代数据集时，队列中要保留的最大批次数。您可以减少此值以降低数据集的 CPU 内存消耗。默认值为 10。

默认情况下，禁用多进程（use_multiprocessing=False），并且只使用一个线程。您应确保仅在代码运行在 Python 的 if __name__ == "__main__": 块内时打开 use_multiprocessing 以避免问题。

这是一个4线程的非多进程示例：

train_py_dataset = ExamplePyDataset(x_train, y_train, batch_size=32, workers=4)
val_py_dataset = ExamplePyDataset(x_val, y_val, batch_size=32, workers=4)

model = get_compiled_model()
model.fit(train_py_dataset, batch_size=64, validation_data=val_py_dataset, epochs=1)

 1563/1563 ━━━━━━━━━━━━━━━━━━━━ 1s 561us/step - loss: 0.5146 - sparse_categorical_accuracy: 0.8516 - val_loss: 0.1623 - val_sparse_categorical_accuracy: 0.9514

<keras.src.callbacks.history.History at 0x2e7fd5ea0>

使用 PyTorch `DataLoader` 对象进行训练和评估

所有内置的训练和评估 API 也兼容 torch.utils.data.Dataset 和 torch.utils.data.DataLoader 对象——无论您是使用 PyTorch 后端，还是 JAX 或 TensorFlow 后端。让我们看一个简单的例子。

与以批处理为中心的 PyDataset 不同，PyTorch Dataset 对象以样本为中心：__len__ 方法返回样本数量，__getitem__ 方法返回特定样本。

class ExampleTorchDataset(torch.utils.data.Dataset):
    def __init__(self, x, y):
        self.x = x
        self.y = y

    def __len__(self):
        return len(self.x)

    def __getitem__(self, idx):
        return self.x[idx], self.y[idx]


train_torch_dataset = ExampleTorchDataset(x_train, y_train)
val_torch_dataset = ExampleTorchDataset(x_val, y_val)

要使用 PyTorch 数据集，您需要将其包装在一个 Dataloader 中，该 Dataloader 负责批处理和洗牌：

train_dataloader = torch.utils.data.DataLoader(
    train_torch_dataset, batch_size=32, shuffle=True
)
val_dataloader = torch.utils.data.DataLoader(
    val_torch_dataset, batch_size=32, shuffle=True
)

现在您可以像使用其他迭代器一样在 Keras API 中使用它们：

model = get_compiled_model()
model.fit(train_dataloader, batch_size=64, validation_data=val_dataloader, epochs=1)
model.evaluate(val_dataloader)

 1563/1563 ━━━━━━━━━━━━━━━━━━━━ 1s 575us/step - loss: 0.5051 - sparse_categorical_accuracy: 0.8568 - val_loss: 0.1613 - val_sparse_categorical_accuracy: 0.9528
 313/313 ━━━━━━━━━━━━━━━━━━━━ 0s 278us/step - loss: 0.1551 - sparse_categorical_accuracy: 0.9541

[0.16209803521633148, 0.9527999758720398]

使用样本加权和类别加权

在默认设置下，样本的权重由其在数据集中的频率决定。有两种方式为数据加权，与样本频率无关：

类别权重
样本权重

类别权重

这是通过将字典传递给 Model.fit() 的 class_weight 参数来设置的。该字典将类别索引映射到应使用的样本权重。

这可以在不重采样的情况下平衡类别，或训练一个对特定类别给予更多重视的模型。

例如，如果类别 "0" 的表示数量是类别 "1" 的一半，你可以使用 Model.fit(..., class_weight={0: 1., 1: 0.5})。

这里有一个 NumPy 示例，我们使用类别权重或样本权重来赋予正确分类类别 #5 的更大重要性（这是 MNIST 数据集中表示数字 "5" 的类别）。

class_weight = {
    0: 1.0,
    1: 1.0,
    2: 1.0,
    3: 1.0,
    4: 1.0,
    # 为类别 "5" 设置权重 "2"，
    # 使该类别的权重增加 2 倍
    5: 2.0,
    6: 1.0,
    7: 1.0,
    8: 1.0,
    9: 1.0,
}

print("使用类别权重进行拟合")
model = get_compiled_model()
model.fit(x_train, y_train, class_weight=class_weight, batch_size=64, epochs=1)

使用类别权重进行拟合
 782/782 ━━━━━━━━━━━━━━━━━━━━ 1s 534us/step - loss: 0.6205 - sparse_categorical_accuracy: 0.8375

<keras.src.callbacks.history.History at 0x298d44eb0>

样本权重

为了更细粒度的控制，或者如果你不是在构建分类器，你可以使用 "样本权重"。

从 NumPy 数据训练时：将 sample_weight 参数传递给 Model.fit()。
从 tf.data 或任何其他类型的迭代器进行训练时：生成 (input_batch, label_batch, sample_weight_batch) 元组。

一个 "样本权重" 数组是一个数字数组，指定每个样本在计算总损失时应具有的权重。它通常用于不平衡分类问题（理念是给稀有类别更多权重）。

当使用的权重为一和零时，该数组可以用作损失函数的掩码，完全忽略某些样本对总损失的贡献。

sample_weight = np.ones(shape=(len(y_train),))
sample_weight[y_train == 5] = 2.0

print("使用样本权重进行拟合")
model = get_compiled_model()
model.fit(x_train, y_train, sample_weight=sample_weight, batch_size=64, epochs=1)

使用样本权重进行拟合
 782/782 ━━━━━━━━━━━━━━━━━━━━ 1s 546us/step - loss: 0.6397 - sparse_categorical_accuracy: 0.8388

<keras.src.callbacks.history.History at 0x298e066e0>

这是一个匹配的 Dataset 示例：

sample_weight = np.ones(shape=(len(y_train),))
sample_weight[y_train == 5] = 2.0

# 创建一个包含样本权重的 Dataset
# （返回元组中的第 3 个元素）。
train_dataset = tf.data.Dataset.from_tensor_slices((x_train, y_train, sample_weight))

# 打乱并切片数据集。
train_dataset = train_dataset.shuffle(buffer_size=1024).batch(64)

model = get_compiled_model()
model.fit(train_dataset, epochs=1)

 782/782 ━━━━━━━━━━━━━━━━━━━━ 1s 651us/step - loss: 0.5971 - sparse_categorical_accuracy: 0.8445

<keras.src.callbacks.history.History at 0x312854100>

向多输入多输出模型传递数据

在之前的示例中，我们考虑了一个具有单个输入（形状为 (764,)）和单个输出（形状为 (10,)）的模型。那么，对于具有多个输入或输出的模型呢？

考虑以下模型，它具有形状为 (32, 32, 3) 的图像输入（即 (高度, 宽度, 通道)）和形状为 (None, 10) 的时间序列输入（即 (时间步, 特征)）。我们的模型将根据这些输入的组合计算两个输出：一个 "分数"（形状为 (1,)）和五个类别的概率分布（形状为 (5,)）。

image_input = keras.Input(shape=(32, 32, 3), name="img_input")
timeseries_input = keras.Input(shape=(None, 10), name="ts_input")

x1 = layers.Conv2D(3, 3)(image_input)
x1 = layers.GlobalMaxPooling2D()(x1)

x2 = layers.Conv1D(3, 3)(timeseries_input)
x2 = layers.GlobalMaxPooling1D()(x2)

x = layers.concatenate([x1, x2])

score_output = layers.Dense(1, name="score_output")(x)
class_output = layers.Dense(5, name="class_output")(x)

model = keras.Model(
    inputs=[image_input, timeseries_input], outputs=[score_output, class_output]
)

我们来绘制这个模型，这样你就可以清楚地看到我们在做什么（请注意，图中显示的形状是批处理形状，而不是每个样本的形状）。

keras.utils.plot_model(model, "multi_input_and_output_model.png", show_shapes=True)

png

在编译时，我们可以通过将损失函数作为列表传递来为不同的输出指定不同的损失：

model.compile(
    optimizer=keras.optimizers.RMSprop(1e-3),
    loss=[
        keras.losses.MeanSquaredError(),
        keras.losses.CategoricalCrossentropy(),
    ],
)

如果我们只向模型传递一个损失函数，则相同的损失函数将应用于每个输出（这在此情况下是不合适的）。

对于指标也是如此：

model.compile(
    optimizer=keras.optimizers.RMSprop(1e-3),
    loss=[
        keras.losses.MeanSquaredError(),
        keras.losses.CategoricalCrossentropy(),
    ],
    metrics=[
        [
            keras.metrics.MeanAbsolutePercentageError(),  # 平均绝对百分比误差
            keras.metrics.MeanAbsoluteError(),            # 平均绝对误差
        ],
        [keras.metrics.CategoricalAccuracy()],           # 类别准确率
    ],
)

由于我们已经为输出层命名，我们还可以通过字典为每个输出指定损失和度量：

model.compile(
    optimizer=keras.optimizers.RMSprop(1e-3),
    loss={
        "score_output": keras.losses.MeanSquaredError(),
        "class_output": keras.losses.CategoricalCrossentropy(),
    },
    metrics={
        "score_output": [
            keras.metrics.MeanAbsolutePercentageError(),
            keras.metrics.MeanAbsoluteError(),
        ],
        "class_output": [keras.metrics.CategoricalAccuracy()],
    },
)

如果您有超过两个输出，建议使用显式名称和字典。

可以使用loss_weights参数为不同的输出特定损失赋予不同的权重（例如，您可能希望在我们的示例中优先考虑“分数”损失，将其重要性设置为类别损失的2倍）：

model.compile(
    optimizer=keras.optimizers.RMSprop(1e-3),
    loss={
        "score_output": keras.losses.MeanSquaredError(),
        "class_output": keras.losses.CategoricalCrossentropy(),
    },
    metrics={
        "score_output": [
            keras.metrics.MeanAbsolutePercentageError(),
            keras.metrics.MeanAbsoluteError(),
        ],
        "class_output": [keras.metrics.CategoricalAccuracy()],
    },
    loss_weights={"score_output": 2.0, "class_output": 1.0},
)

如果某些输出用于预测但不用于训练，您也可以选择不计算损失：

# 列表损失版本
model.compile(
    optimizer=keras.optimizers.RMSprop(1e-3),
    loss=[None, keras.losses.CategoricalCrossentropy()],
)

# 或字典损失版本
model.compile(
    optimizer=keras.optimizers.RMSprop(1e-3),
    loss={"class_output": keras.losses.CategoricalCrossentropy()},
)

在fit()中传递数据到多输入或多输出模型的方式与在编译中指定损失函数相似：您可以传递NumPy数组的列表（与接收损失函数的输出1:1映射）或将输出名称映射到NumPy数组的字典。

model.compile(
    optimizer=keras.optimizers.RMSprop(1e-3),
    loss=[
        keras.losses.MeanSquaredError(),
        keras.losses.CategoricalCrossentropy(),
    ],
)

# 生成虚拟NumPy数据
img_data = np.random.random_sample(size=(100, 32, 32, 3))
ts_data = np.random.random_sample(size=(100, 20, 10))
score_targets = np.random.random_sample(size=(100, 1))
class_targets = np.random.random_sample(size=(100, 5))

# 在列表上拟合
model.fit([img_data, ts_data], [score_targets, class_targets], batch_size=32, epochs=1)

# 或者，在字典上拟合
model.fit(
    {"img_input": img_data, "ts_input": ts_data},
    {"score_output": score_targets, "class_output": class_targets},
    batch_size=32,
    epochs=1,
)

 4/4 ━━━━━━━━━━━━━━━━━━━━ 0s 62ms/step - loss: 18.0146
 4/4 ━━━━━━━━━━━━━━━━━━━━ 0s 56ms/step - loss: 17.6494

<keras.src.callbacks.history.History at 0x31a6c5810>

这里是Dataset的用例：与我们对NumPy数组所做的类似，Dataset应返回一个字典元组。

train_dataset = tf.data.Dataset.from_tensor_slices(
    (
        {"img_input": img_data, "ts_input": ts_data},
        {"score_output": score_targets, "class_output": class_targets},
    )
)
train_dataset = train_dataset.shuffle(buffer_size=1024).batch(64)

model.fit(train_dataset, epochs=1)

 2/2 ━━━━━━━━━━━━━━━━━━━━ 0s 197ms/step - loss: 17.8578

<keras.src.callbacks.history.History at 0x17c7e5690>

使用回调

Keras中的回调是一些对象，它们在训练的不同时间点被调用（在一个epoch开始时，在一个批次结束时，在一个epoch结束时，等等）。它们可用于实现某些行为，例如：

在训练的不同阶段进行验证（超出内置的每个epoch验证）
定期检查点模型或当其超出某个准确度阈值时
当训练似乎停滞时改变模型的学习率
当训练似乎停滞时对顶部层进行微调
当训练结束或超出某个性能阈值时发送电子邮件或即时消息通知
等等

回调可以作为列表传递给您的fit()调用：

model = get_compiled_model()

callbacks = [
    keras.callbacks.EarlyStopping(
        # 当`val_loss`不再改善时停止训练
        monitor="val_loss",
        # "不再改善" 被定义为 "没有比1e-2更好"
        min_delta=1e-2,
        # "不再改善" 进一步被定义为 "至少为2个epoch"
        patience=2,
        verbose=1,
    )
]
model.fit(
    x_train,
    y_train,
    epochs=20,
    batch_size=64,
    callbacks=callbacks,
    validation_split=0.2,
)

第 1 个周期/20
 625/625 ━━━━━━━━━━━━━━━━━━━━ 1s 622us/step - 损失: 0.6245 - 稀疏分类准确率: 0.8275 - 验证损失: 0.2231 - 验证稀疏分类准确率: 0.9330
第 2 个周期/20
 625/625 ━━━━━━━━━━━━━━━━━━━━ 0s 404us/step - 损失: 0.1809 - 稀疏分类准确率: 0.9460 - 验证损失: 0.1727 - 验证稀疏分类准确率: 0.9476
第 3 个周期/20
 625/625 ━━━━━━━━━━━━━━━━━━━━ 0s 398us/step - 损失: 0.1336 - 稀疏分类准确率: 0.9598 - 验证损失: 0.1564 - 验证稀疏分类准确率: 0.9545
第 4 个周期/20
 625/625 ━━━━━━━━━━━━━━━━━━━━ 0s 400us/step - 损失: 0.1012 - 稀疏分类准确率: 0.9699 - 验证损失: 0.1502 - 验证稀疏分类准确率: 0.9570
第 5 个周期/20
 625/625 ━━━━━━━━━━━━━━━━━━━━ 0s 403us/step - 损失: 0.0835 - 稀疏分类准确率: 0.9748 - 验证损失: 0.1436 - 验证稀疏分类准确率: 0.9589
第 6 个周期/20
 625/625 ━━━━━━━━━━━━━━━━━━━━ 0s 396us/step - 损失: 0.0699 - 稀疏分类准确率: 0.9783 - 验证损失: 0.1484 - 验证稀疏分类准确率: 0.9577
第 7 个周期/20
 625/625 ━━━━━━━━━━━━━━━━━━━━ 0s 402us/step - 损失: 0.0603 - 稀疏分类准确率: 0.9814 - 验证损失: 0.1406 - 验证稀疏分类准确率: 0.9629
第 7 轮: 早停

<keras.src.callbacks.history.History 在 0x31ae37c10>

可用的许多内置回调

Keras 中已经提供了许多内置回调，例如：

ModelCheckpoint: 定期保存模型。
EarlyStopping: 当训练不再提高验证指标时停止训练。
TensorBoard: 定期写入模型日志，可以在 TensorBoard 中可视化（更多细节在“可视化”部分）。
CSVLogger: 将损失和指标数据流式传输到 CSV 文件。
等等。

有关完整列表，请参见回调文档。

编写自己的回调

您可以通过扩展基类 keras.callbacks.Callback 来创建自定义回调。回调可以通过类属性 self.model 访问其关联的模型。

确保阅读编写自定义回调的完整指南。

这是一个简单的示例，在训练期间保存每批次的损失值列表：

class LossHistory(keras.callbacks.Callback):
    def on_train_begin(self, logs):
        self.per_batch_losses = []

    def on_batch_end(self, batch, logs):
        self.per_batch_losses.append(logs.get("loss"))

模型检查点

当您在相对较大的数据集上训练模型时，定期保存模型的检查点是至关重要的。

实现这一点的最简单方法是使用 ModelCheckpoint 回调：

model = get_compiled_model()

callbacks = [
    keras.callbacks.ModelCheckpoint(
        # 保存模型的路径
        # 下面的两个参数意味着仅当
        # `val_loss` 分数有所改善时，才会覆盖当前检查点。
        # 保存的模型名称将包含当前的 epoch。
        filepath="mymodel_{epoch}.keras",
        save_best_only=True,  # 仅当 `val_loss` 改善时才保存模型。
        monitor="val_loss",
        verbose=1,
    )
]
model.fit(
    x_train,
    y_train,
    epochs=2,
    batch_size=64,
    callbacks=callbacks,
    validation_split=0.2,
)

Epoch 1/2
 559/625 ━━━━━━━━━━━━━━━━━━━━  0s 360us/step - loss: 0.6490 - sparse_categorical_accuracy: 0.8209
Epoch 1: val_loss improved from inf to 0.22393, saving model to mymodel_1.keras
 625/625 ━━━━━━━━━━━━━━━━━━━━ 1s 577us/step - loss: 0.6194 - sparse_categorical_accuracy: 0.8289 - val_loss: 0.2239 - val_sparse_categorical_accuracy: 0.9340
Epoch 2/2
 565/625 ━━━━━━━━━━━━━━━━━━━━  0s 355us/step - loss: 0.1816 - sparse_categorical_accuracy: 0.9476
Epoch 2: val_loss improved from 0.22393 to 0.16868, saving model to mymodel_2.keras
 625/625 ━━━━━━━━━━━━━━━━━━━━ 0s 411us/step - loss: 0.1806 - sparse_categorical_accuracy: 0.9479 - val_loss: 0.1687 - val_sparse_categorical_accuracy: 0.9494

<keras.src.callbacks.history.History at 0x2e5cb7250>

ModelCheckpoint 回调可用于实现容错：在训练意外被中断的情况下，能够从模型的最后保存状态重新开始训练。以下是一个基本示例：

# 准备一个目录来存储所有检查点。
checkpoint_dir = "./ckpt"
if not os.path.exists(checkpoint_dir):
    os.makedirs(checkpoint_dir)


def make_or_restore_model():
    # 恢复最新模型，或在没有可用检查点时创建一个新模型。
    checkpoints = [checkpoint_dir + "/" + name for name in os.listdir(checkpoint_dir)]
    if checkpoints:
        latest_checkpoint = max(checkpoints, key=os.path.getctime)
        print("正在从", latest_checkpoint, "恢复")
        return keras.models.load_model(latest_checkpoint)
    print("正在创建新模型")
    return get_compiled_model()


model = make_or_restore_model()
callbacks = [
    # 该回调每100批次保存一次模型。
    # 我们在保存的模型名称中包含训练损失。
    keras.callbacks.ModelCheckpoint(
        filepath=checkpoint_dir + "/model-loss={loss:.2f}.keras", save_freq=100
    )
]
model.fit(x_train, y_train, epochs=1, callbacks=callbacks)

正在创建新模型
 1563/1563 ━━━━━━━━━━━━━━━━━━━━ 1s 390us/step - loss: 0.4910 - sparse_categorical_accuracy: 0.8623

<keras.src.callbacks.history.History at 0x2e5c454e0>

您还可以编写自己的回调来保存和恢复模型。

有关序列化和保存的完整指南，请参见保存和序列化模型指南。

使用学习率调度

在训练深度学习模型时，一个常见的模式是随着训练的进行逐渐降低学习率。这通常被称为“学习率衰减”。

学习率衰减计划可以是静态的（提前固定，作为当前 epoch 或当前批次索引的函数），也可以是动态的（对模型的当前行为做出反应，特别是验证损失）。

将调度传递给优化器

您可以通过在优化器中传递调度对象作为learning_rate参数，轻松使用静态学习率衰减调度：

initial_learning_rate = 0.1
lr_schedule = keras.optimizers.schedules.ExponentialDecay(
    initial_learning_rate, decay_steps=100000, decay_rate=0.96, staircase=True
)

optimizer = keras.optimizers.RMSprop(learning_rate=lr_schedule)

可以使用的几种内置调度包括：ExponentialDecay、PiecewiseConstantDecay、PolynomialDecay和InverseTimeDecay。

使用回调实现动态学习率调度

动态学习率调度（例如，当验证损失不再改善时降低学习率）无法通过这些调度对象实现，因为优化器没有访问验证指标的权限。

然而，回调可以访问所有指标，包括验证指标！因此，您可以通过使用一个修改优化器当前学习率的回调来实现这一模式。实际上，这已经作为ReduceLROnPlateau回调内置。

使用TensorBoard可视化训练期间的损失和指标

在训练期间观察模型的最佳方式是使用 TensorBoard – 一种浏览器基础的应用程序您可以在本地运行，为您提供：

实时绘制训练和评估的损失和指标
（可选）可视化层激活的直方图
（可选）您Embedding层学习的嵌入空间的3D可视化

如果您通过pip安装了TensorFlow，您应该可以从命令行启动TensorBoard：

tensorboard --logdir=/full_path_to_your_logs

使用TensorBoard回调

将TensorBoard与Keras模型和fit()方法结合使用的最简单方法是TensorBoard回调。

在最简单的情况下，只需指定要将日志写入的位置，您就可以开始使用了：

keras.callbacks.TensorBoard(
    log_dir="/full_path_to_your_logs",
    histogram_freq=0,  # 记录直方图可视化的频率
    embeddings_freq=0,  # 记录嵌入可视化的频率
    update_freq="epoch",
)  # 写入日志的频率（默认：每个周期一次）

<keras.src.callbacks.tensorboard.TensorBoard at 0x31b0188b0>

有关更多信息，请参见 TensorBoard回调的文档。

使用内置方法进行训练与评估

◆ 设置

◆ 介绍

◆ API 概述：第一个端到端示例

◆ compile() 方法：指定损失、指标和优化器

有许多内置的优化器、损失和指标可用

自定义损失

自定义指标

处理不符合标准签名的损失和指标

自动划分验证保留集

◆ 使用 tf.data 数据集进行训练与评估

◆ 使用 PyDataset 实例进行训练和评估

◆ 使用 PyTorch DataLoader 对象进行训练和评估

◆ 使用样本加权和类别加权

类别权重

样本权重

◆ 向多输入多输出模型传递数据

◆ 使用回调

可用的许多内置回调

编写自己的回调

◆ 模型检查点

◆ 使用学习率调度

将调度传递给优化器

使用回调实现动态学习率调度

◆ 使用TensorBoard可视化训练期间的损失和指标

使用TensorBoard回调

使用内置方法进行训练与评估

设置

介绍

API 概述：第一个端到端示例

compile() 方法：指定损失、指标和优化器

有许多内置的优化器、损失和指标可用

自定义损失

自定义指标

处理不符合标准签名的损失和指标

自动划分验证保留集

使用 tf.data 数据集进行训练与评估

使用 PyDataset 实例进行训练和评估

使用 PyTorch DataLoader 对象进行训练和评估

使用样本加权和类别加权

类别权重

样本权重

向多输入多输出模型传递数据

使用回调

可用的许多内置回调

编写自己的回调

模型检查点

使用学习率调度

将调度传递给优化器

使用回调实现动态学习率调度

使用TensorBoard可视化训练期间的损失和指标

使用TensorBoard回调

`compile()` 方法：指定损失、指标和优化器

使用 `tf.data` 数据集进行训练与评估

使用 `PyDataset` 实例进行训练和评估

使用 PyTorch `DataLoader` 对象进行训练和评估