Sklearn、TensorFlow 与 Keras 机器学习实用指南第三版（五）

原文：Hands-On Machine Learning with Scikit-Learn, Keras, and TensorFlow

译者：飞龙

协议：CC BY-NC-SA 4.0

第十二章：使用 TensorFlow 进行自定义模型和训练

到目前为止，我们只使用了 TensorFlow 的高级 API，Keras，但它已经让我们走得很远：我们构建了各种神经网络架构，包括回归和分类网络，Wide & Deep 网络，自正则化网络，使用各种技术，如批量归一化，dropout 和学习率调度。事实上，您将遇到的 95%用例不需要除了 Keras（和 tf.data）之外的任何东西（请参见第十三章）。但现在是时候深入研究 TensorFlow，看看它的低级Python API。当您需要额外控制以编写自定义损失函数，自定义指标，层，模型，初始化程序，正则化器，权重约束等时，这将非常有用。您甚至可能需要完全控制训练循环本身；例如，应用特殊的转换或约束到梯度（超出仅仅剪切它们）或为网络的不同部分使用多个优化器。我们将在本章中涵盖所有这些情况，并且还将看看如何使用 TensorFlow 的自动生成图功能来提升您的自定义模型和训练算法。但首先，让我们快速浏览一下 TensorFlow。

TensorFlow 的快速浏览

正如您所知，TensorFlow 是一个强大的用于数值计算的库，特别适用于大规模机器学习（但您也可以用它来进行需要大量计算的任何其他任务）。它由 Google Brain 团队开发，驱动了谷歌许多大规模服务，如 Google Cloud Speech，Google Photos 和 Google Search。它于 2015 年 11 月开源，现在是业界最广泛使用的深度学习库：无数项目使用 TensorFlow 进行各种机器学习任务，如图像分类，自然语言处理，推荐系统和时间序列预测。

那么 TensorFlow 提供了什么？以下是一个摘要：

• 它的核心与 NumPy 非常相似，但支持 GPU。

• 它支持分布式计算（跨多个设备和服务器）。

• 它包括一种即时（JIT）编译器，允许它优化计算以提高速度和内存使用。它通过从 Python 函数中提取计算图，优化它（例如通过修剪未使用的节点），并有效地运行它（例如通过自动并行运行独立操作）来工作。

• 计算图可以导出为可移植格式，因此您可以在一个环境中训练 TensorFlow 模型（例如在 Linux 上使用 Python），并在另一个环境中运行它（例如在 Android 设备上使用 Java）。

• 它实现了反向模式自动微分（请参见第十章和附录 B）并提供了一些优秀的优化器，如 RMSProp 和 Nadam（请参见第十一章），因此您可以轻松最小化各种损失函数。

TensorFlow 提供了许多建立在这些核心功能之上的功能：最重要的当然是 Keras，但它还有数据加载和预处理操作（tf.data，tf.io 等），图像处理操作（tf.image），信号处理操作（tf.signal）等等（请参见图 12-1 以获取 TensorFlow 的 Python API 概述）。

提示

我们将涵盖 TensorFlow API 的许多包和函数，但不可能覆盖所有内容，因此您应该花些时间浏览 API；您会发现它非常丰富且有很好的文档。

在最低级别上，每个 TensorFlow 操作（简称 op）都是使用高效的 C++代码实现的。许多操作有多个称为内核的实现：每个内核专门用于特定设备类型，如 CPU、GPU，甚至 TPU（张量处理单元）。正如您可能知道的，GPU 可以通过将计算分成许多较小的块并在许多 GPU 线程上并行运行来显着加快计算速度。TPU 速度更快：它们是专门用于深度学习操作的定制 ASIC 芯片（我们将在第十九章讨论如何使用 GPU 或 TPU 与 TensorFlow）。

图 12-1. TensorFlow 的 Python API

TensorFlow 的架构如图 12-2 所示。大部分时间，您的代码将使用高级 API（特别是 Keras 和 tf.data），但当您需要更灵活性时，您将使用较低级别的 Python API，直接处理张量。无论如何，TensorFlow 的执行引擎将有效地运行操作，即使跨多个设备和机器，如果您告诉它的话。

TensorFlow 不仅可以在 Windows、Linux 和 macOS 上运行，还可以在移动设备上运行（使用 TensorFlow Lite），包括 iOS 和 Android（请参阅第十九章）。请注意，如果您不想使用 Python API，还可以使用其他语言的 API：有 C++、Java 和 Swift 的 API。甚至还有一个名为 TensorFlow.js 的 JavaScript 实现，可以直接在浏览器中运行您的模型。

图 12-2. TensorFlow 的架构

TensorFlow 不仅仅是一个库。TensorFlow 是一个庞大生态系统中心。首先，有用于可视化的 TensorBoard（请参阅第十章）。接下来，有由 Google 构建的用于将 TensorFlow 项目投入生产的一套库，称为TensorFlow Extended (TFX)：它包括用于数据验证、预处理、模型分析和服务的工具（使用 TF Serving；请参阅第十九章）。Google 的 TensorFlow Hub 提供了一种轻松下载和重复使用预训练神经网络的方式。您还可以在 TensorFlow 的model garden中获得许多神经网络架构，其中一些是预训练的。查看TensorFlow 资源和https://github.com/jtoy/awesome-tensorflow以获取更多基于 TensorFlow 的项目。您可以在 GitHub 上找到数百个 TensorFlow 项目，因此通常很容易找到您正在尝试做的任何事情的现有代码。

提示

越来越多的机器学习论文随着它们的实现发布，有时甚至附带预训练模型。请查看https://paperswithcode.com以轻松找到它们。

最后但并非最不重要的是，TensorFlow 拥有一支充满激情和乐于助人的开发团队，以及一个庞大的社区为其改进做出贡献。要提出技术问题，您应该使用https://stackoverflow.com，并在问题中标记tensorflow和python。您可以通过GitHub提交错误和功能请求。要进行一般讨论，请加入TensorFlow 论坛。

好了，现在是开始编码的时候了！

像 NumPy 一样使用 TensorFlow

TensorFlow 的 API 围绕着张量展开，这些张量从操作流向操作，因此得名 TensorFlow。张量与 NumPy 的ndarray非常相似：通常是一个多维数组，但也可以保存标量（例如42）。当我们创建自定义成本函数、自定义指标、自定义层等时，这些张量将非常重要，让我们看看如何创建和操作它们。

张量和操作

您可以使用tf.constant()创建一个张量。例如，这里是一个表示具有两行三列浮点数的矩阵的张量：

>>> import tensorflow as tf
>>> t = tf.constant([[1., 2., 3.], [4., 5., 6.]])  # matrix
>>> t
<tf.Tensor: shape=(2, 3), dtype=float32, numpy=
array([[1., 2., 3.],
 [4., 5., 6.]], dtype=float32)>

就像ndarray一样，tf.Tensor有一个形状和一个数据类型（dtype）：

>>> t.shape
TensorShape([2, 3])
>>> t.dtype
tf.float32

索引工作方式与 NumPy 类似：

>>> t[:, 1:]
<tf.Tensor: shape=(2, 2), dtype=float32, numpy=
array([[2., 3.],
 [5., 6.]], dtype=float32)>
>>> t[..., 1, tf.newaxis]
<tf.Tensor: shape=(2, 1), dtype=float32, numpy=
array([[2.],
 [5.]], dtype=float32)>

最重要的是，各种张量操作都是可用的：

>>> t + 10
<tf.Tensor: shape=(2, 3), dtype=float32, numpy=
array([[11., 12., 13.],
 [14., 15., 16.]], dtype=float32)>
>>> tf.square(t)
<tf.Tensor: shape=(2, 3), dtype=float32, numpy=
array([[ 1.,  4.,  9.],
 [16., 25., 36.]], dtype=float32)>
>>> t @ tf.transpose(t)
<tf.Tensor: shape=(2, 2), dtype=float32, numpy=
array([[14., 32.],
 [32., 77.]], dtype=float32)>

请注意，编写t + 10等同于调用tf.add(t, 10)（实际上，Python 调用了魔术方法t.__add__(10)，它只是调用了tf.add(t, 10)）。其他运算符，如-和*，也受支持。@运算符在 Python 3.5 中添加，用于矩阵乘法：它等同于调用tf.matmul()函数。

注意

许多函数和类都有别名。例如，tf.add()和tf.math.add()是相同的函数。这使得 TensorFlow 可以为最常见的操作保留简洁的名称，同时保持良好组织的包。

张量也可以保存标量值。在这种情况下，形状为空：

>>> tf.constant(42)
<tf.Tensor: shape=(), dtype=int32, numpy=42>

注意

Keras API 有自己的低级 API，位于tf.keras.backend中。这个包通常被导入为K，以简洁为主。它曾经包括函数如K.square()、K.exp()和K.sqrt()，您可能在现有代码中遇到：这在 Keras 支持多个后端时编写可移植代码很有用，但现在 Keras 只支持 TensorFlow，您应该直接调用 TensorFlow 的低级 API（例如，使用tf.square()而不是K.square()）。从技术上讲，K.square()及其相关函数仍然存在以保持向后兼容性，但tf.keras.backend包的文档只列出了一些实用函数，例如clear_session()（在第十章中提到）。

您将找到所有您需要的基本数学运算（tf.add()、tf.multiply()、tf.square()、tf.exp()、tf.sqrt()等）以及大多数您可以在 NumPy 中找到的操作（例如tf.reshape()、tf.squeeze()、tf.tile()）。一些函数的名称与 NumPy 中的名称不同；例如，tf.reduce_mean()、tf.reduce_sum()、tf.reduce_max()和tf.math.log()相当于np.mean()、np.sum()、np.max()和np.log()。当名称不同时，通常有很好的理由。例如，在 TensorFlow 中，您必须编写tf.transpose(t)；您不能像在 NumPy 中那样只写t.T。原因是tf.transpose()函数与 NumPy 的T属性并不完全相同：在 TensorFlow 中，将创建一个具有其自己的转置数据副本的新张量，而在 NumPy 中，t.T只是相同数据的一个转置视图。同样，tf.reduce_sum()操作之所以被命名为这样，是因为其 GPU 核心（即 GPU 实现）使用的减少算法不保证元素添加的顺序：因为 32 位浮点数的精度有限，每次调用此操作时结果可能会发生微小变化。tf.reduce_mean()也是如此（当然tf.reduce_max()是确定性的）。

张量和 NumPy

张量与 NumPy 兼容：您可以从 NumPy 数组创建张量，反之亦然。您甚至可以将 TensorFlow 操作应用于 NumPy 数组，将 NumPy 操作应用于张量：

>>> import numpy as np
>>> a = np.array([2., 4., 5.])
>>> tf.constant(a)
<tf.Tensor: id=111, shape=(3,), dtype=float64, numpy=array([2., 4., 5.])>
>>> t.numpy()  # or np.array(t)
array([[1., 2., 3.],
 [4., 5., 6.]], dtype=float32)
>>> tf.square(a)
<tf.Tensor: id=116, shape=(3,), dtype=float64, numpy=array([4., 16., 25.])>
>>> np.square(t)
array([[ 1.,  4.,  9.],
 [16., 25., 36.]], dtype=float32)

警告

请注意，NumPy 默认使用 64 位精度，而 TensorFlow 使用 32 位。这是因为 32 位精度通常对神经网络来说足够了，而且运行速度更快，使用的内存更少。因此，当您从 NumPy 数组创建张量时，请确保设置dtype=tf.float32。

类型转换

类型转换可能会严重影响性能，并且当它们自动完成时很容易被忽略。为了避免这种情况，TensorFlow 不会自动执行任何类型转换：如果您尝试在具有不兼容类型的张量上执行操作，它只会引发异常。例如，您不能将浮点张量和整数张量相加，甚至不能将 32 位浮点数和 64 位浮点数相加：

>>> tf.constant(2.) + tf.constant(40)
[...] InvalidArgumentError: [...] expected to be a float tensor [...]
>>> tf.constant(2.) + tf.constant(40., dtype=tf.float64)
[...] InvalidArgumentError: [...] expected to be a float tensor [...]

这可能一开始有点烦人，但请记住这是为了一个好的目的！当然，当您真正需要转换类型时，您可以使用tf.cast()：

>>> t2 = tf.constant(40., dtype=tf.float64)
>>> tf.constant(2.0) + tf.cast(t2, tf.float32)
<tf.Tensor: id=136, shape=(), dtype=float32, numpy=42.0>

变量

到目前为止，我们看到的tf.Tensor值是不可变的：我们无法修改它们。这意味着我们不能使用常规张量来实现神经网络中的权重，因为它们需要通过反向传播进行调整。此外，其他参数可能也需要随时间变化（例如，动量优化器会跟踪过去的梯度）。我们需要的是tf.Variable：

>>> v = tf.Variable([[1., 2., 3.], [4., 5., 6.]])
>>> v
<tf.Variable 'Variable:0' shape=(2, 3) dtype=float32, numpy=
array([[1., 2., 3.],
 [4., 5., 6.]], dtype=float32)>

tf.Variable的行为很像tf.Tensor：您可以执行相同的操作，它与 NumPy 很好地配合，对类型也一样挑剔。但是它也可以使用assign()方法（或assign_add()或assign_sub()，它们会增加或减少给定值来就地修改变量）。您还可以使用单个单元格（或切片）的assign()方法或使用scatter_update()或scatter_nd_update()方法来修改单个单元格（或切片）：

v.assign(2 * v)           # v now equals [[2., 4., 6.], [8., 10., 12.]]
v[0, 1].assign(42)        # v now equals [[2., 42., 6.], [8., 10., 12.]]
v[:, 2].assign([0., 1.])  # v now equals [[2., 42., 0.], [8., 10., 1.]]
v.scatter_nd_update(      # v now equals [[100., 42., 0.], [8., 10., 200.]]
    indices=[[0, 0], [1, 2]], updates=[100., 200.])

直接赋值不起作用：

>>> v[1] = [7., 8., 9.]
[...] TypeError: 'ResourceVariable' object does not support item assignment

注意

在实践中，您很少需要手动创建变量；Keras 提供了一个add_weight()方法，它会为您处理，您将看到。此外，模型参数通常会直接由优化器更新，因此您很少需要手动更新变量。

其他数据结构

TensorFlow 支持几种其他数据结构，包括以下内容（请参阅本章笔记本中的“其他数据结构”部分或附录 C 了解更多详细信息）：

稀疏张量（tf.SparseTensor）

高效地表示大部分为零的张量。tf.sparse包含了稀疏张量的操作。

张量数组（tf.TensorArray）

是张量列表。它们默认具有固定长度，但可以选择性地扩展。它们包含的所有张量必须具有相同的形状和数据类型。

不规则张量（tf.RaggedTensor）

表示张量列表，所有张量的秩和数据类型相同，但大小不同。张量大小变化的维度称为不规则维度。tf.ragged包含了不规则张量的操作。

字符串张量

是类型为tf.string的常规张量。这些表示字节字符串，而不是 Unicode 字符串，因此如果您使用 Unicode 字符串（例如，像"café"这样的常规 Python 3 字符串）创建字符串张量，那么它将自编码为 UTF-8（例如，b"cafxc3xa9"）。或者，您可以使用类型为tf.int32的张量来表示 Unicode 字符串，其中每个项目表示一个 Unicode 代码点（例如，[99, 97, 102, 233]）。tf.strings包（带有s）包含用于字节字符串和 Unicode 字符串的操作（以及将一个转换为另一个的操作）。重要的是要注意tf.string是原子的，这意味着其长度不会出现在张量的形状中。一旦您将其转换为 Unicode 张量（即，一个包含 Unicode 代码点的tf.int32类型的张量），长度将出现在形状中。

集合

表示为常规张量（或稀疏张量）。例如，tf.constant([[1, 2], [3, 4]])表示两个集合{1, 2}和{3, 4}。更一般地，每个集合由张量的最后一个轴中的向量表示。您可以使用tf.sets包中的操作来操作集合。

队列

在多个步骤中存储张量。TensorFlow 提供各种类型的队列：基本的先进先出（FIFO）队列（FIFOQueue），以及可以优先处理某些项目的队列（PriorityQueue），对其项目进行洗牌的队列（RandomShuffleQueue），以及通过填充来批处理不同形状的项目的队列（PaddingFIFOQueue）。这些类都在tf.queue包中。

有了张量、操作、变量和各种数据结构，你现在可以定制你的模型和训练算法了！

自定义模型和训练算法

你将首先创建一个自定义损失函数，这是一个简单而常见的用例。

自定义损失函数

假设你想训练一个回归模型，但你的训练集有点嘈杂。当然，你首先尝试通过删除或修复异常值来清理数据集，但结果还不够好；数据集仍然很嘈杂。你应该使用哪种损失函数？均方误差可能会过分惩罚大误差，导致模型不够精确。平均绝对误差不会像惩罚异常值那样严重，但训练可能需要一段时间才能收敛，训练出的模型可能不够精确。这可能是使用 Huber 损失的好时机（在第十章介绍）。Huber 损失在 Keras 中是可用的（只需使用tf.keras.losses.Huber类的实例），但让我们假装它不存在。要实现它，只需创建一个函数，该函数将标签和模型预测作为参数，并使用 TensorFlow 操作来计算包含所有损失的张量（每个样本一个）：

def huber_fn(y_true, y_pred):
    error = y_true - y_pred
    is_small_error = tf.abs(error) < 1
    squared_loss = tf.square(error) / 2
    linear_loss  = tf.abs(error) - 0.5
    return tf.where(is_small_error, squared_loss, linear_loss)

警告

为了获得更好的性能，你应该使用矢量化的实现，就像这个例子一样。此外，如果你想要从 TensorFlow 的图优化功能中受益，你应该只使用 TensorFlow 操作。

也可以返回平均损失而不是单个样本损失，但这不推荐，因为这样做会使在需要时无法使用类权重或样本权重（参见第十章）。

现在你可以在编译 Keras 模型时使用这个 Huber 损失函数，然后像往常一样训练你的模型：

model.compile(loss=huber_fn, optimizer="nadam")
model.fit(X_train, y_train, [...])

就是这样！在训练期间的每个批次中，Keras 将调用huber_fn()函数来计算损失，然后使用反向模式自动微分来计算损失相对于所有模型参数的梯度，最后执行梯度下降步骤（在这个例子中使用 Nadam 优化器）。此外，它将跟踪自从 epoch 开始以来的总损失，并显示平均损失。

但是当你保存模型时，这个自定义损失会发生什么？

保存和加载包含自定义组件的模型

保存包含自定义损失函数的模型可以正常工作，但是当你加载它时，你需要提供一个将函数名称映射到实际函数的字典。更一般地，当你加载包含自定义对象的模型时，你需要将名称映射到对象：

model = tf.keras.models.load_model("my_model_with_a_custom_loss",
                                   custom_objects={"huber_fn": huber_fn})

提示

如果你用@keras.utils.​reg⁠ister_keras_serializable()装饰huber_fn()函数，它将自动可用于load_model()函数：不需要将其包含在custom_objects字典中。

使用当前的实现，任何在-1 和 1 之间的错误都被认为是“小”。但是如果你想要一个不同的阈值呢？一个解决方案是创建一个函数来创建一个配置好的损失函数：

def create_huber(threshold=1.0):
    def huber_fn(y_true, y_pred):
        error = y_true - y_pred
        is_small_error = tf.abs(error) < threshold
        squared_loss = tf.square(error) / 2
        linear_loss  = threshold * tf.abs(error) - threshold ** 2 / 2
        return tf.where(is_small_error, squared_loss, linear_loss)
    return huber_fn

model.compile(loss=create_huber(2.0), optimizer="nadam")

不幸的是，当你保存模型时，threshold不会被保存。这意味着在加载模型时你将需要指定threshold的值（注意要使用的名称是"huber_fn"，这是你给 Keras 的函数的名称，而不是创建它的函数的名称）：

model = tf.keras.models.load_model(
    "my_model_with_a_custom_loss_threshold_2",
    custom_objects={"huber_fn": create_huber(2.0)}
)

你可以通过创建tf.keras.losses.Loss类的子类，然后实现它的get_config()方法来解决这个问题：

class HuberLoss(tf.keras.losses.Loss):
    def __init__(self, threshold=1.0, **kwargs):
        self.threshold = threshold
        super().__init__(**kwargs)

    def call(self, y_true, y_pred):
        error = y_true - y_pred
        is_small_error = tf.abs(error) < self.threshold
        squared_loss = tf.square(error) / 2
        linear_loss  = self.threshold * tf.abs(error) - self.threshold**2 / 2
        return tf.where(is_small_error, squared_loss, linear_loss)

    def get_config(self):
        base_config = super().get_config()
        return {**base_config, "threshold": self.threshold}

让我们来看看这段代码：

• 构造函数接受**kwargs并将它们传递给父构造函数，父构造函数处理标准超参数：损失的name和用于聚合单个实例损失的reduction算法。默认情况下，这是"AUTO"，等同于"SUM_OVER_BATCH_SIZE"：损失将是实例损失的总和，加权后再除以批量大小（而不是加权平均）。其他可能的值是"SUM"和"NONE"。

• call()方法接受标签和预测值，计算所有实例损失，并返回它们。

• get_config()方法返回一个字典，将每个超参数名称映射到其值。它首先调用父类的get_config()方法，然后将新的超参数添加到此字典中。

然后您可以在编译模型时使用此类的任何实例：

model.compile(loss=HuberLoss(2.), optimizer="nadam")

当您保存模型时，阈值将与模型一起保存；当您加载模型时，您只需要将类名映射到类本身：

model = tf.keras.models.load_model("my_model_with_a_custom_loss_class",
                                   custom_objects={"HuberLoss": HuberLoss})

当您保存模型时，Keras 会调用损失实例的get_config()方法，并以 SavedModel 格式保存配置。当您加载模型时，它会在HuberLoss类上调用from_config()类方法：这个方法由基类（Loss）实现，并创建一个类的实例，将**config传递给构造函数。

损失就是这样了！正如您现在将看到的，自定义激活函数、初始化器、正则化器和约束并没有太大不同。

自定义激活函数、初始化器、正则化器和约束

大多数 Keras 功能，如损失、正则化器、约束、初始化器、指标、激活函数、层，甚至完整模型，都可以以类似的方式进行自定义。大多数情况下，您只需要编写一个带有适当输入和输出的简单函数。这里有一个自定义激活函数的示例（相当于tf.keras.activations.softplus()或tf.nn.softplus()）、一个自定义 Glorot 初始化器的示例（相当于tf.keras.initializers.glorot_normal()）、一个自定义ℓ[1]正则化器的示例（相当于tf.keras.regularizers.l1(0.01)）以及一个确保权重都为正的自定义约束的示例（相当于tf.keras.​con⁠straints.nonneg()或tf.nn.relu()）：

def my_softplus(z):
    return tf.math.log(1.0 + tf.exp(z))

def my_glorot_initializer(shape, dtype=tf.float32):
    stddev = tf.sqrt(2. / (shape[0] + shape[1]))
    return tf.random.normal(shape, stddev=stddev, dtype=dtype)

def my_l1_regularizer(weights):
    return tf.reduce_sum(tf.abs(0.01 * weights))

def my_positive_weights(weights):  # return value is just tf.nn.relu(weights)
    return tf.where(weights < 0., tf.zeros_like(weights), weights)

正如您所看到的，参数取决于自定义函数的类型。然后可以像这里展示的那样正常使用这些自定义函数：

layer = tf.keras.layers.Dense(1, activation=my_softplus,
                              kernel_initializer=my_glorot_initializer,
                              kernel_regularizer=my_l1_regularizer,
                              kernel_constraint=my_positive_weights)

激活函数将应用于此Dense层的输出，并将其结果传递给下一层。层的权重将使用初始化器返回的值进行初始化。在每个训练步骤中，权重将传递给正则化函数以计算正则化损失，然后将其添加到主损失中以获得用于训练的最终损失。最后，在每个训练步骤之后，将调用约束函数，并将层的权重替换为受约束的权重。

如果一个函数有需要与模型一起保存的超参数，那么您将希望子类化适当的类，比如tf.keras.regu⁠larizers.​​Reg⁠⁠ularizer、tf.keras.constraints.Constraint、tf.keras.initializers.​Ini⁠tializer或tf.keras.layers.Layer（适用于任何层，包括激活函数）。就像您为自定义损失所做的那样，这里是一个简单的ℓ[1]正则化类，它保存了其factor超参数（这次您不需要调用父构造函数或get_config()方法，因为它们不是由父类定义的）：

class MyL1Regularizer(tf.keras.regularizers.Regularizer):
    def __init__(self, factor):
        self.factor = factor

    def __call__(self, weights):
        return tf.reduce_sum(tf.abs(self.factor * weights))

    def get_config(self):
        return {"factor": self.factor}

请注意，您必须为损失、层（包括激活函数）和模型实现call()方法，或者为正则化器、初始化器和约束实现__call__()方法。对于指标，情况有些不同，您将立即看到。

自定义指标

损失和指标在概念上并不相同：损失（例如，交叉熵）被梯度下降用来训练模型，因此它们必须是可微的（至少在评估它们的点上），它们的梯度不应该在任何地方都为零。此外，如果它们不容易被人类解释也是可以的。相反，指标（例如，准确率）用于评估模型：它们必须更容易被解释，可以是不可微的或者在任何地方梯度为零。

也就是说，在大多数情况下，定义一个自定义指标函数与定义一个自定义损失函数完全相同。实际上，我们甚至可以使用我们之前创建的 Huber 损失函数作为指标；它会工作得很好（在这种情况下，持久性也会以相同的方式工作，只保存函数的名称"huber_fn"，而不是阈值）：

model.compile(loss="mse", optimizer="nadam", metrics=[create_huber(2.0)])

在训练期间的每个批次，Keras 将计算这个指标并跟踪自开始时的平均值。大多数情况下，这正是你想要的。但并非总是如此！例如，考虑一个二元分类器的精度。正如你在第三章中看到的，精度是真正例的数量除以正例的预测数量（包括真正例和假正例）。假设模型在第一个批次中做出了五个正面预测，其中四个是正确的：这是 80%的精度。然后假设模型在第二个批次中做出了三个正面预测，但它们全部是错误的：这是第二个批次的 0%精度。如果你只计算这两个精度的平均值，你会得到 40%。但等一下——这不是这两个批次的模型精度！事实上，总共有四个真正例（4 + 0）中的八个正面预测（5 + 3），所以总体精度是 50%，而不是 40%。我们需要的是一个对象，它可以跟踪真正例的数量和假正例的数量，并且可以在需要时基于这些数字计算精度。这正是tf.keras.metrics.Precision类所做的：

>>> precision = tf.keras.metrics.Precision()
>>> precision([0, 1, 1, 1, 0, 1, 0, 1], [1, 1, 0, 1, 0, 1, 0, 1])
<tf.Tensor: shape=(), dtype=float32, numpy=0.8>
>>> precision([0, 1, 0, 0, 1, 0, 1, 1], [1, 0, 1, 1, 0, 0, 0, 0])
<tf.Tensor: shape=(), dtype=float32, numpy=0.5>

在这个例子中，我们创建了一个Precision对象，然后像一个函数一样使用它，为第一个批次传递标签和预测，然后为第二个批次（如果需要，还可以传递样本权重）。我们使用了与刚才讨论的示例中相同数量的真正例和假正例。在第一个批次之后，它返回 80%的精度；然后在第二个批次之后，它返回 50%（这是到目前为止的总体精度，而不是第二个批次的精度）。这被称为流式指标（或有状态指标），因为它逐渐更新，批次之后。

在任何时候，我们可以调用result()方法来获取指标的当前值。我们还可以通过使用variables属性查看其变量（跟踪真正例和假正例的数量），并可以使用reset_states()方法重置这些变量：

>>> precision.result()
<tf.Tensor: shape=(), dtype=float32, numpy=0.5>
>>> precision.variables
[<tf.Variable 'true_positives:0' [...], numpy=array([4.], dtype=float32)>,
 <tf.Variable 'false_positives:0' [...], numpy=array([4.], dtype=float32)>]
>>> precision.reset_states()  # both variables get reset to 0.0

如果需要定义自己的自定义流式指标，创建tf.keras.metrics.Metric类的子类。这里是一个基本示例，它跟踪总 Huber 损失和迄今为止看到的实例数量。当要求结果时，它返回比率，这只是平均 Huber 损失：

class HuberMetric(tf.keras.metrics.Metric):
    def __init__(self, threshold=1.0, **kwargs):
        super().__init__(**kwargs)  # handles base args (e.g., dtype)
        self.threshold = threshold
        self.huber_fn = create_huber(threshold)
        self.total = self.add_weight("total", initializer="zeros")
        self.count = self.add_weight("count", initializer="zeros")

    def update_state(self, y_true, y_pred, sample_weight=None):
        sample_metrics = self.huber_fn(y_true, y_pred)
        self.total.assign_add(tf.reduce_sum(sample_metrics))
        self.count.assign_add(tf.cast(tf.size(y_true), tf.float32))

    def result(self):
        return self.total / self.count

    def get_config(self):
        base_config = super().get_config()
        return {**base_config, "threshold": self.threshold}

让我们走一遍这段代码：

• 构造函数使用add_weight()方法创建需要在多个批次中跟踪指标状态的变量——在这种情况下，所有 Huber 损失的总和（total）和迄今为止看到的实例数量（count）。如果愿意，你也可以手动创建变量。Keras 跟踪任何设置为属性的tf.Variable（更一般地，任何“可跟踪”的对象，如层或模型）。

• 当你将这个类的实例用作函数时（就像我们用Precision对象做的那样），update_state()方法会被调用。它根据一个批次的标签和预测更新变量（以及样本权重，但在这种情况下我们忽略它们）。

• result()方法计算并返回最终结果，在这种情况下是所有实例上的平均 Huber 指标。当你将指标用作函数时，首先调用update_state()方法，然后调用result()方法，并返回其输出。

• 我们还实现了get_config()方法，以确保threshold与模型一起保存。

• reset_states()方法的默认实现将所有变量重置为 0.0（但如果需要，你可以覆盖它）。

注意

Keras 会无缝处理变量持久性；不需要任何操作。

当你使用简单函数定义指标时，Keras 会自动为每个批次调用它，并在每个时期期间跟踪平均值，就像我们手动做的那样。因此，我们的HuberMetric类的唯一好处是threshold将被保存。但当然，有些指标，比如精度，不能简单地在批次上进行平均：在这些情况下，除了实现流式指标之外别无选择。

现在你已经构建了一个流式指标，构建一个自定义层将会变得轻而易举！

自定义层

有时候你可能想要构建一个包含一种 TensorFlow 没有提供默认实现的奇特层的架构。或者你可能只是想要构建一个非常重复的架构，在这种架构中，一个特定的层块被重复多次，将每个块视为单个层会很方便。对于这些情况，你会想要构建一个自定义层。

有一些没有权重的层，比如tf.keras.layers.Flatten或tf.keras.layers.ReLU。如果你想创建一个没有任何权重的自定义层，最简单的方法是编写一个函数并将其包装在tf.keras.layers.Lambda层中。例如，以下层将对其输入应用指数函数：

exponential_layer = tf.keras.layers.Lambda(lambda x: tf.exp(x))

然后，这个自定义层可以像任何其他层一样使用，使用序贯 API、函数式 API 或子类 API。你也可以将它用作激活函数，或者你可以使用activation=tf.exp。指数层有时用于回归模型的输出层，当要预测的值具有非常不同的规模时（例如，0.001、10.、1,000.）。事实上，指数函数是 Keras 中的标准激活函数之一，所以你可以简单地使用activation="exponential"。

你可能会猜到，要构建一个自定义的有状态层（即带有权重的层），你需要创建tf.keras.layers.Layer类的子类。例如，以下类实现了Dense层的简化版本：

class MyDense(tf.keras.layers.Layer):
    def __init__(self, units, activation=None, **kwargs):
        super().__init__(**kwargs)
        self.units = units
        self.activation = tf.keras.activations.get(activation)

    def build(self, batch_input_shape):
        self.kernel = self.add_weight(
            name="kernel", shape=[batch_input_shape[-1], self.units],
            initializer="glorot_normal")
        self.bias = self.add_weight(
            name="bias", shape=[self.units], initializer="zeros")

    def call(self, X):
        return self.activation(X @ self.kernel + self.bias)

    def get_config(self):
        base_config = super().get_config()
        return {**base_config, "units": self.units,
                "activation": tf.keras.activations.serialize(self.activation)}

让我们来看看这段代码：

• 构造函数将所有超参数作为参数（在这个例子中是units和activation），并且重要的是它还接受一个**kwargs参数。它调用父构造函数，将kwargs传递给它：这会处理标准参数，如input_shape、trainable和name。然后它将超参数保存为属性，使用tf.keras.activations.get()函数将activation参数转换为适当的激活函数（它接受函数、标准字符串如"relu"或"swish"，或者简单地None）。

• build()方法的作用是通过为每个权重调用add_weight()方法来创建层的变量。build()方法在第一次使用该层时被调用。在那时，Keras 将知道该层输入的形状，并将其传递给build()方法，这通常是创建一些权重所必需的。例如，我们需要知道前一层中的神经元数量以创建连接权重矩阵（即"kernel"）：这对应于输入的最后一个维度的大小。在build()方法的最后（仅在最后），您必须调用父类的build()方法：这告诉 Keras 该层已构建（它只是设置self.built = True）。

• call()方法执行所需的操作。在这种情况下，我们计算输入X和层的内核的矩阵乘法，添加偏置向量，并将激活函数应用于结果，这给出了层的输出。

• get_config()方法与以前的自定义类中的方法一样。请注意，通过调用tf.keras.​activa⁠tions.serialize()保存激活函数的完整配置。

现在您可以像使用任何其他层一样使用MyDense层！

注意

Keras 会自动推断输出形状，除非该层是动态的（稍后将看到）。在这种（罕见）情况下，您需要实现compute_output_shape()方法，该方法必须返回一个TensorShape对象。

要创建具有多个输入的层（例如，Concatenate），call()方法的参数应该是一个包含所有输入的元组。要创建具有多个输出的层，call()方法应该返回输出的列表。例如，以下示例玩具层接受两个输入并返回三个输出：

class MyMultiLayer(tf.keras.layers.Layer):
    def call(self, X):
        X1, X2 = X
        return X1 + X2, X1 * X2, X1 / X2

这个层现在可以像任何其他层一样使用，但当然只能使用功能 API 和子类 API，而不能使用顺序 API（顺序 API 只接受具有一个输入和一个输出的层）。

如果您的层在训练和测试期间需要具有不同的行为（例如，如果它使用Dropout或BatchNormalization层），那么您必须在call()方法中添加一个training参数，并使用此参数来决定要执行什么操作。例如，让我们创建一个在训练期间添加高斯噪声（用于正则化）但在测试期间不执行任何操作的层（Keras 有一个执行相同操作的层，tf.keras.layers.GaussianNoise）：

class MyGaussianNoise(tf.keras.layers.Layer):
    def __init__(self, stddev, **kwargs):
        super().__init__(**kwargs)
        self.stddev = stddev

    def call(self, X, training=False):
        if training:
            noise = tf.random.normal(tf.shape(X), stddev=self.stddev)
            return X + noise
        else:
            return X

有了这个，您现在可以构建任何您需要的自定义层！现在让我们看看如何创建自定义模型。

自定义模型

我们已经在第十章中讨论了使用子类 API 创建自定义模型类。这很简单：子类化tf.keras.Model类，在构造函数中创建层和变量，并实现call()方法以执行您希望模型执行的操作。例如，假设我们想要构建图 12-3 中表示的模型。

图 12-3。自定义模型示例：一个包含跳过连接的自定义ResidualBlock层的任意模型

输入首先经过一个密集层，然后通过由两个密集层和一个加法操作组成的残差块（如您将在第十四章中看到的，残差块将其输入添加到其输出中），然后通过这个相同的残差块再进行三次，然后通过第二个残差块，最终结果通过一个密集输出层。如果这个模型看起来没有太多意义，不要担心；这只是一个示例，说明您可以轻松构建任何您想要的模型，甚至包含循环和跳过连接的模型。要实现这个模型，最好首先创建一个ResidualBlock层，因为我们将创建一对相同的块（并且可能希望在另一个模型中重用它）：

class ResidualBlock(tf.keras.layers.Layer):
    def __init__(self, n_layers, n_neurons, **kwargs):
        super().__init__(**kwargs)
        self.hidden = [tf.keras.layers.Dense(n_neurons, activation="relu",
                                             kernel_initializer="he_normal")
                       for _ in range(n_layers)]

    def call(self, inputs):
        Z = inputs
        for layer in self.hidden:
            Z = layer(Z)
        return inputs + Z

这个层有点特殊，因为它包含其他层。Keras 会自动处理这一点：它会自动检测hidden属性包含可跟踪对象（在这种情况下是层），因此它们的变量会自动添加到此层的变量列表中。这个类的其余部分是不言自明的。接下来，让我们使用子类 API 来定义模型本身：

class ResidualRegressor(tf.keras.Model):
    def __init__(self, output_dim, **kwargs):
        super().__init__(**kwargs)
        self.hidden1 = tf.keras.layers.Dense(30, activation="relu",
                                             kernel_initializer="he_normal")
        self.block1 = ResidualBlock(2, 30)
        self.block2 = ResidualBlock(2, 30)
        self.out = tf.keras.layers.Dense(output_dim)

    def call(self, inputs):
        Z = self.hidden1(inputs)
        for _ in range(1 + 3):
            Z = self.block1(Z)
        Z = self.block2(Z)
        return self.out(Z)

我们在构造函数中创建层，并在call()方法中使用它们。然后可以像任何其他模型一样使用此模型（编译、拟合、评估和使用它进行预测）。如果您还希望能够使用save()方法保存模型，并使用tf.keras.models.load_model()函数加载模型，则必须在ResidualBlock类和ResidualRegressor类中实现get_config()方法（就像我们之前做的那样）。或者，您可以使用save_weights()和load_weights()方法保存和加载权重。

Model类是Layer类的子类，因此模型可以像层一样定义和使用。但是模型具有一些额外的功能，包括当然包括compile()、fit()、evaluate()和predict()方法（以及一些变体），还有get_layer()方法（可以通过名称或索引返回模型的任何层）和save()方法（以及对tf.keras.models.load_model()和tf.keras.models.clone_model()的支持）。

提示

如果模型提供的功能比层更多，为什么不将每个层都定义为模型呢？技术上您可以这样做，但通常更清晰的做法是区分模型的内部组件（即层或可重用的层块）和模型本身（即您将训练的对象）。前者应该是Layer类的子类，而后者应该是Model类的子类。

有了这些，您可以自然而简洁地构建几乎任何您在论文中找到的模型，使用顺序 API、函数 API、子类 API，甚至这些的混合。“几乎”任何模型？是的，还有一些事情我们需要看一下：首先是如何基于模型内部定义损失或指标，其次是如何构建自定义训练循环。

基于模型内部的损失和指标

我们之前定义的自定义损失和指标都是基于标签和预测（以及可选的样本权重）。有时您可能希望基于模型的其他部分（例如其隐藏层的权重或激活）定义损失。这可能对正则化目的或监视模型的某些内部方面很有用。

要基于模型内部定义自定义损失，可以根据模型的任何部分计算损失，然后将结果传递给add_loss()方法。例如，让我们构建一个由五个隐藏层堆叠加一个输出层组成的自定义回归 MLP 模型。这个自定义模型还将在最上面的隐藏层之上具有一个辅助输出。与这个辅助输出相关联的损失将被称为重建损失（参见第十七章）：它是重建和输入之间的均方差差异。通过将这个重建损失添加到主要损失中，我们将鼓励模型通过隐藏层尽可能保留更多信息，即使这些信息对于回归任务本身并不直接有用。在实践中，这种损失有时会改善泛化能力（它是一种正则化损失）。还可以使用模型的add_metric()方法添加自定义指标。以下是具有自定义重建损失和相应指标的自定义模型的代码：

class ReconstructingRegressor(tf.keras.Model):
    def __init__(self, output_dim, **kwargs):
        super().__init__(**kwargs)
        self.hidden = [tf.keras.layers.Dense(30, activation="relu",
                                             kernel_initializer="he_normal")
                       for _ in range(5)]
        self.out = tf.keras.layers.Dense(output_dim)
        self.reconstruction_mean = tf.keras.metrics.Mean(
            name="reconstruction_error")

    def build(self, batch_input_shape):
        n_inputs = batch_input_shape[-1]
        self.reconstruct = tf.keras.layers.Dense(n_inputs)

    def call(self, inputs, training=False):
        Z = inputs
        for layer in self.hidden:
            Z = layer(Z)
        reconstruction = self.reconstruct(Z)
        recon_loss = tf.reduce_mean(tf.square(reconstruction - inputs))
        self.add_loss(0.05 * recon_loss)
        if training:
            result = self.reconstruction_mean(recon_loss)
            self.add_metric(result)
        return self.out(Z)

让我们来看一下这段代码：

• 构造函数创建了一个具有五个密集隐藏层和一个密集输出层的 DNN。我们还创建了一个Mean流式指标，用于在训练过程中跟踪重建误差。

• build()方法创建一个额外的密集层，用于重构模型的输入。它必须在这里创建，因为其单元数必须等于输入的数量，在调用build()方法之前这个数量是未知的。

• call()方法通过所有五个隐藏层处理输入，然后将结果传递给重构层，该层生成重构。

• 然后call()方法计算重构损失（重构和输入之间的均方差），并使用add_loss()方法将其添加到模型的损失列表中。请注意，我们通过将重构损失乘以 0.05 来缩小重构损失（这是一个可以调整的超参数）。这确保了重构损失不会主导主要损失。

• 接下来，在训练过程中，call()方法更新重构度量并将其添加到模型中以便显示。这段代码示例实际上可以通过调用self.add_metric(recon_loss)来简化：Keras 将自动为您跟踪均值。

• 最后，call()方法将隐藏层的输出传递给输出层，并返回其输出。

在训练过程中，总损失和重构损失都会下降：

Epoch 1/5
363/363 [========] - 1s 820us/step - loss: 0.7640 - reconstruction_error: 1.2728
Epoch 2/5
363/363 [========] - 0s 809us/step - loss: 0.4584 - reconstruction_error: 0.6340
[...]

在大多数情况下，到目前为止我们讨论的一切将足以实现您想构建的任何模型，即使是具有复杂架构、损失和指标。然而，对于一些架构，如 GANs（参见第十七章），您将不得不自定义训练循环本身。在我们到达那里之前，我们必须看看如何在 TensorFlow 中自动计算梯度。

使用自动微分计算梯度

要了解如何使用自动微分（参见第十章和附录 B）自动计算梯度，让我们考虑一个简单的玩具函数：

def f(w1, w2):
    return 3 * w1 ** 2 + 2 * w1 * w2

如果你懂微积分，你可以分析地找到这个函数相对于w1的偏导数是6 * w1 + 2 * w2。你也可以找到它相对于w2的偏导数是2 * w1。例如，在点(w1, w2) = (5, 3)，这些偏导数分别等于 36 和 10，因此在这一点的梯度向量是（36，10）。但如果这是一个神经网络，这个函数会复杂得多，通常有数万个参数，通过手工分析找到偏导数将是一个几乎不可能的任务。一个解决方案是通过测量当你微调相应参数一点点时函数的输出如何变化来计算每个偏导数的近似值：

>>> w1, w2 = 5, 3
>>> eps = 1e-6
>>> (f(w1 + eps, w2) - f(w1, w2)) / eps
36.000003007075065
>>> (f(w1, w2 + eps) - f(w1, w2)) / eps
10.000000003174137

看起来不错！这个方法运行得相当好，而且易于实现，但它只是一个近似值，重要的是你需要至少针对每个参数调用一次f()（不是两次，因为我们可以只计算一次f(w1, w2)）。每个参数至少调用一次f()使得这种方法在大型神经网络中变得难以处理。因此，我们应该使用反向模式自动微分。TensorFlow 使这变得非常简单：

w1, w2 = tf.Variable(5.), tf.Variable(3.)
with tf.GradientTape() as tape:
    z = f(w1, w2)

gradients = tape.gradient(z, [w1, w2])

首先我们定义两个变量w1和w2，然后我们创建一个tf.GradientTape上下文，它将自动记录涉及变量的每个操作，最后我们要求这个磁带计算结果z相对于两个变量[w1, w2]的梯度。让我们看看 TensorFlow 计算的梯度：

>>> gradients
[<tf.Tensor: shape=(), dtype=float32, numpy=36.0>,
 <tf.Tensor: shape=(), dtype=float32, numpy=10.0>]

太棒了！结果不仅准确（精度仅受浮点误差限制），而且gradient()方法只需通过记录的计算一次（按相反顺序），无论有多少变量，因此非常高效。就像魔术一样！

提示

为了节省内存，在tf.GradientTape()块中只放入严格的最小值。或者，通过在tf.GradientTape()块内创建一个with tape.stop_recording()块来暂停记录。

在调用其gradient()方法后，磁带会立即被擦除，因此如果尝试两次调用gradient()，将会收到异常：

with tf.GradientTape() as tape:
    z = f(w1, w2)

dz_dw1 = tape.gradient(z, w1)  # returns tensor 36.0
dz_dw2 = tape.gradient(z, w2)  # raises a RuntimeError!

如果您需要多次调用gradient()，您必须使磁带持久化，并在每次完成后删除它以释放资源：

with tf.GradientTape(persistent=True) as tape:
    z = f(w1, w2)

dz_dw1 = tape.gradient(z, w1)  # returns tensor 36.0
dz_dw2 = tape.gradient(z, w2)  # returns tensor 10.0, works fine now!
del tape

默认情况下，磁带只会跟踪涉及变量的操作，因此，如果您尝试计算z相对于除变量以外的任何东西的梯度，结果将是None：

c1, c2 = tf.constant(5.), tf.constant(3.)
with tf.GradientTape() as tape:
    z = f(c1, c2)

gradients = tape.gradient(z, [c1, c2])  # returns [None, None]

但是，您可以强制磁带监视任何您喜欢的张量，记录涉及它们的每个操作。然后，您可以计算相对于这些张量的梯度，就像它们是变量一样：

with tf.GradientTape() as tape:
    tape.watch(c1)
    tape.watch(c2)
    z = f(c1, c2)

gradients = tape.gradient(z, [c1, c2])  # returns [tensor 36., tensor 10.]

在某些情况下，这可能很有用，比如如果您想要实现一个正则化损失，惩罚激活在输入变化很小时变化很大的情况：损失将基于激活相对于输入的梯度。由于输入不是变量，您需要告诉磁带监视它们。

大多数情况下，梯度磁带用于计算单个值（通常是损失）相对于一组值（通常是模型参数）的梯度。这就是反向模式自动微分的优势所在，因为它只需要进行一次前向传递和一次反向传递就可以一次性获得所有梯度。如果尝试计算向量的梯度，例如包含多个损失的向量，那么 TensorFlow 将计算向量总和的梯度。因此，如果您需要获取各个梯度（例如，每个损失相对于模型参数的梯度），您必须调用磁带的jacobian()方法：它将为向量中的每个损失执行一次反向模式自动微分（默认情况下全部并行）。甚至可以计算二阶偏导数（Hessians，即偏导数的偏导数），但在实践中很少需要（请参阅本章笔记本的“使用自动微分计算梯度”部分以获取示例）。

在某些情况下，您可能希望阻止梯度通过神经网络的某些部分进行反向传播。为此，您必须使用tf.stop_gradient()函数。该函数在前向传递期间返回其输入（类似于tf.identity()），但在反向传播期间不允许梯度通过（它的作用类似于常数）：

def f(w1, w2):
    return 3 * w1 ** 2 + tf.stop_gradient(2 * w1 * w2)

with tf.GradientTape() as tape:
    z = f(w1, w2)  # the forward pass is not affected by stop_gradient()

gradients = tape.gradient(z, [w1, w2])  # returns [tensor 30., None]

最后，当计算梯度时，您可能偶尔会遇到一些数值问题。例如，如果在x=10^（-50）处计算平方根函数的梯度，结果将是无穷大。实际上，该点的斜率并不是无穷大，但它超过了 32 位浮点数的处理能力：

>>> x = tf.Variable(1e-50)
>>> with tf.GradientTape() as tape:
...     z = tf.sqrt(x)
...
>>> tape.gradient(z, [x])
[<tf.Tensor: shape=(), dtype=float32, numpy=inf>]

为了解决这个问题，在计算平方根时，通常建议向x（例如 10^（-6））添加一个微小值。

指数函数也经常引起头痛，因为它增长非常快。例如，之前定义的my_softplus()的方式在数值上不稳定。如果计算my_softplus(100.0)，您将得到无穷大而不是正确的结果（约为 100）。但是可以重写该函数以使其在数值上稳定：softplus 函数被定义为 log(1 + exp(z))，这也等于 log(1 + exp(–|z|)) + max(z, 0)（请参阅数学证明的笔记本），第二种形式的优势在于指数项不会爆炸。因此，这是my_softplus()函数的更好实现：

def my_softplus(z):
    return tf.math.log(1 + tf.exp(-tf.abs(z))) + tf.maximum(0., z)

在一些罕见的情况下，一个数值稳定的函数可能仍然具有数值不稳定的梯度。在这种情况下，你将不得不告诉 TensorFlow 使用哪个方程来计算梯度，而不是让它使用自动微分。为此，你必须在定义函数时使用@tf.​cus⁠tom_gradient装饰器，并返回函数的通常结果以及计算梯度的函数。例如，让我们更新my_softplus()函数，使其也返回一个数值稳定的梯度函数：

@tf.custom_gradient
def my_softplus(z):
    def my_softplus_gradients(grads):  # grads = backprop'ed from upper layers
        return grads * (1 - 1 / (1 + tf.exp(z)))  # stable grads of softplus

    result = tf.math.log(1 + tf.exp(-tf.abs(z))) + tf.maximum(0., z)
    return result, my_softplus_gradients

如果你懂微积分（参见关于这个主题的教程笔记本），你会发现 log(1 + exp(z))的导数是 exp(z) / (1 + exp(z))。但这种形式是不稳定的：对于较大的z值，它最终会计算出无穷大除以无穷大，返回 NaN。然而，通过一点代数操作，你可以证明它也等于 1 – 1 / (1 + exp(z))，这是稳定的。my_softplus_gradients()函数使用这个方程来计算梯度。请注意，这个函数将接收到目前为止反向传播的梯度，一直到my_softplus()函数，并根据链式法则，我们必须将它们与这个函数的梯度相乘。

现在当我们计算my_softplus()函数的梯度时，即使对于较大的输入值，我们也会得到正确的结果。

恭喜！现在你可以计算任何函数的梯度（只要在计算时它是可微的），甚至在需要时阻止反向传播，并编写自己的梯度函数！这可能比你需要的灵活性更多，即使你构建自己的自定义训练循环。接下来你将看到如何做到这一点。

自定义训练循环

在某些情况下，fit()方法可能不够灵活以满足你的需求。例如，我们在第十章中讨论的Wide & Deep 论文使用了两种不同的优化器：一种用于宽路径，另一种用于深路径。由于fit()方法只使用一个优化器（在编译模型时指定的那个），实现这篇论文需要编写自己的自定义循环。

你可能也喜欢编写自定义训练循环，只是为了更有信心地确保它们确实按照你的意图执行（也许你对fit()方法的一些细节不确定）。有时候，让一切都显式化可能会感觉更安全。然而，请记住，编写自定义训练循环会使你的代码变得更长、更容易出错，并且更难维护。

提示

除非你在学习或确实需要额外的灵活性，否则应该优先使用fit()方法而不是实现自己的训练循环，特别是如果你在团队中工作。

首先，让我们构建一个简单的模型。不需要编译它，因为我们将手动处理训练循环：

l2_reg = tf.keras.regularizers.l2(0.05)
model = tf.keras.models.Sequential([
    tf.keras.layers.Dense(30, activation="relu", kernel_initializer="he_normal",
                          kernel_regularizer=l2_reg),
    tf.keras.layers.Dense(1, kernel_regularizer=l2_reg)
])

接下来，让我们创建一个小函数，从训练集中随机抽取一个批次的实例（在第十三章中，我们将讨论 tf.data API，它提供了一个更好的替代方案）：

def random_batch(X, y, batch_size=32):
    idx = np.random.randint(len(X), size=batch_size)
    return X[idx], y[idx]

让我们还定义一个函数，用于显示训练状态，包括步数、总步数、自开始时的平均损失（我们将使用Mean指标来计算），以及其他指标：

def print_status_bar(step, total, loss, metrics=None):
    metrics = " - ".join([f"{m.name}: {m.result():.4f}"
                          for m in [loss] + (metrics or [])])
    end = "" if step < total else "n"
    print(f"r{step}/{total} - " + metrics, end=end)

这段代码很容易理解，除非你不熟悉 Python 的字符串格式化：{m.result():.4f}将指标的结果格式化为小数点后四位的浮点数，使用r（回车）和end=""确保状态栏始终打印在同一行上。

有了这个，让我们开始吧！首先，我们需要定义一些超参数，并选择优化器、损失函数和指标（在这个例子中只有 MAE）：

n_epochs = 5
batch_size = 32
n_steps = len(X_train) // batch_size
optimizer = tf.keras.optimizers.SGD(learning_rate=0.01)
loss_fn = tf.keras.losses.mean_squared_error
mean_loss = tf.keras.metrics.Mean(name="mean_loss")
metrics = [tf.keras.metrics.MeanAbsoluteError()]

现在我们准备构建自定义循环了！

for epoch in range(1, n_epochs + 1):
    print("Epoch {}/{}".format(epoch, n_epochs))
    for step in range(1, n_steps + 1):
        X_batch, y_batch = random_batch(X_train_scaled, y_train)
        with tf.GradientTape() as tape:
            y_pred = model(X_batch, training=True)
            main_loss = tf.reduce_mean(loss_fn(y_batch, y_pred))
            loss = tf.add_n([main_loss] + model.losses)

        gradients = tape.gradient(loss, model.trainable_variables)
        optimizer.apply_gradients(zip(gradients, model.trainable_variables))
        mean_loss(loss)
        for metric in metrics:
            metric(y_batch, y_pred)

        print_status_bar(step, n_steps, mean_loss, metrics)

    for metric in [mean_loss] + metrics:
        metric.reset_states()

这段代码中有很多内容，让我们来逐步解释一下：

• 我们创建两个嵌套循环：一个用于时期，另一个用于时期内的批次。

• 然后我们从训练集中抽取一个随机批次。

• 在 tf.GradientTape() 块内，我们对一个批次进行预测，使用模型作为一个函数，并计算损失：它等于主要损失加上其他损失（在这个模型中，每层有一个正则化损失）。由于 mean_squared_error() 函数返回每个实例的一个损失，我们使用 tf.reduce_mean() 计算批次的平均值（如果您想对每个实例应用不同的权重，这就是您应该做的地方）。正则化损失已经被减少为每个单一标量，所以我们只需要对它们求和（使用 tf.add_n()，它对相同形状和数据类型的多个张量求和）。

• 接下来，我们要求磁带计算损失相对于每个可训练变量的梯度——不是所有变量！——并将它们应用于优化器以执行梯度下降步骤。

• 然后我们更新平均损失和指标（在当前时期内），并显示状态栏。

• 在每个时期结束时，我们重置平均损失和指标的状态。

如果您想应用梯度裁剪（参见第十一章），请设置优化器的 clipnorm 或 clipvalue 超参数。如果您想对梯度应用任何其他转换，只需在调用 apply_gradients() 方法之前这样做。如果您想向模型添加权重约束（例如，在创建层时设置 kernel_constraint 或 bias_constraint），您应该更新训练循环以在 apply_gradients() 之后应用这些约束，就像这样：

for variable in model.variables:
    if variable.constraint is not None:
        variable.assign(variable.constraint(variable))

警告

在训练循环中调用模型时不要忘记设置 training=True，特别是如果您的模型在训练和测试期间表现不同（例如，如果它使用 BatchNormalization 或 Dropout）。如果是自定义模型，请确保将 training 参数传播到您的模型调用的层。

正如您所看到的，有很多事情需要做对，很容易出错。但好的一面是，您可以完全控制。

现在您知道如何自定义模型的任何部分⁠¹⁵和训练算法，让我们看看如何使用 TensorFlow 的自动生成图形功能：它可以显著加快您的自定义代码，并且还将其移植到 TensorFlow 支持的任何平台（参见第十九章）。

TensorFlow 函数和图形

回到 TensorFlow 1，图形是不可避免的（伴随着复杂性），因为它们是 TensorFlow API 的核心部分。自从 TensorFlow 2（2019 年发布）以来，图形仍然存在，但不再是核心部分，而且使用起来简单得多（多得多！）。为了展示它们有多简单，让我们从一个计算其输入的立方的微不足道的函数开始：

def cube(x):
    return x ** 3

我们显然可以使用 Python 值（如整数或浮点数）调用此函数，或者我们可以使用张量调用它：

>>> cube(2)
8
>>> cube(tf.constant(2.0))
<tf.Tensor: shape=(), dtype=float32, numpy=8.0>

现在，让我们使用 tf.function() 将这个 Python 函数转换为 TensorFlow 函数：

>>> tf_cube = tf.function(cube)
>>> tf_cube
<tensorflow.python.eager.def_function.Function at 0x7fbfe0c54d50>

然后，这个 TF 函数可以像原始的 Python 函数一样使用，并且将返回相同的结果（但始终作为张量）：

>>> tf_cube(2)
<tf.Tensor: shape=(), dtype=int32, numpy=8>
>>> tf_cube(tf.constant(2.0))
<tf.Tensor: shape=(), dtype=float32, numpy=8.0>

在幕后，tf.function() 分析了 cube() 函数执行的计算，并生成了一个等效的计算图！正如您所看到的，这是相当轻松的（我们很快会看到这是如何工作的）。或者，我们也可以将 tf.function 用作装饰器；这实际上更常见：

@tf.function
def tf_cube(x):
    return x ** 3

原始的 Python 函数仍然可以通过 TF 函数的 python_function 属性访问，以防您需要它：

>>> tf_cube.python_function(2)
8

TensorFlow 优化计算图，修剪未使用的节点，简化表达式（例如，1 + 2 将被替换为 3）等。一旦优化的图准备就绪，TF 函数将有效地执行图中的操作，按适当的顺序（并在可能时并行执行）。因此，TF 函数通常比原始 Python 函数运行得快得多，特别是如果它执行复杂计算。大多数情况下，您实际上不需要知道更多：当您想要提升 Python 函数时，只需将其转换为 TF 函数。就这样！

此外，如果在调用tf.function()时设置jit_compile=True，那么 TensorFlow 将使用加速线性代数（XLA）为您的图编译专用内核，通常融合多个操作。例如，如果您的 TF 函数调用tf.reduce_sum(a * b + c)，那么没有 XLA，函数首先需要计算a * b并将结果存储在临时变量中，然后将c添加到该变量中，最后调用tf.reduce_sum()处理结果。使用 XLA，整个计算将编译为单个内核，该内核将一次性计算tf.reduce_sum(a * b + c)，而不使用任何大型临时变量。这不仅速度更快，而且使用的 RAM 大大减少。

当您编写自定义损失函数、自定义指标、自定义层或任何其他自定义函数，并在 Keras 模型中使用它（就像我们在本章中一直做的那样），Keras 会自动将您的函数转换为 TF 函数——无需使用tf.function()。因此，大多数情况下，这种魔术是 100%透明的。如果您希望 Keras 使用 XLA，只需在调用compile()方法时设置jit_compile=True。简单！

提示

您可以通过在创建自定义层或自定义模型时设置dynamic=True来告诉 Keras不将您的 Python 函数转换为 TF 函数。或者，您可以在调用模型的compile()方法时设置run_eagerly=True。

默认情况下，TF 函数为每个唯一的输入形状和数据类型生成一个新图，并将其缓存以供后续调用。例如，如果您调用tf_cube(tf.constant(10))，将为形状为[]的 int32 张量生成一个图。然后，如果您调用tf_cube(tf.constant(20))，将重用相同的图。但是，如果您随后调用tf_cube(tf.constant([10, 20]))，将为形状为[2]的 int32 张量生成一个新图。这就是 TF 函数处理多态性（即不同的参数类型和形状）的方式。但是，这仅适用于张量参数：如果将数值 Python 值传递给 TF 函数，则将为每个不同的值生成一个新图：例如，调用tf_cube(10)和tf_cube(20)将生成两个图。

警告

如果您多次使用不同的数值 Python 值调用 TF 函数，则将生成许多图，减慢程序速度并使用大量 RAM（您必须删除 TF 函数才能释放它）。Python 值应保留用于将具有少量唯一值的参数，例如每层神经元的数量之类的超参数。这样可以使 TensorFlow 更好地优化模型的每个变体。

AutoGraph 和跟踪

那么 TensorFlow 如何生成图呢？它首先通过分析 Python 函数的源代码来捕获所有控制流语句，比如for循环、while循环和if语句，以及break、continue和return语句。这第一步被称为AutoGraph。TensorFlow 必须分析源代码的原因是 Python 没有提供其他捕获控制流语句的方法：它提供了像__add__()和__mul__()这样的魔术方法来捕获+和*等运算符，但没有__while__()或__if__()这样的魔术方法。在分析函数代码之后，AutoGraph 会输出一个升级版本的函数，其中所有控制流语句都被适当的 TensorFlow 操作替换，比如tf.while_loop()用于循环，tf.cond()用于if语句。例如，在图 12-4 中，AutoGraph 分析了sum_squares() Python 函数的源代码，并生成了tf__sum_squares()函数。在这个函数中，for循环被替换为loop_body()函数的定义（包含原始for循环的主体），然后调用for_stmt()函数。这个调用将在计算图中构建适当的tf.while_loop()操作。

图 12-4. TensorFlow 如何使用 AutoGraph 和跟踪生成图

接下来，TensorFlow 调用这个“升级”函数，但不是传递参数，而是传递一个符号张量—一个没有实际值的张量，只有一个名称、一个数据类型和一个形状。例如，如果您调用sum_squares(tf.constant(10))，那么tf__sum_squares()函数将被调用，传递一个类型为 int32、形状为[]的符号张量。该函数将在图模式下运行，这意味着每个 TensorFlow 操作都会在图中添加一个节点来表示自己和其输出张量（与常规模式相反，称为急切执行或急切模式）。在图模式下，TF 操作不执行任何计算。图模式是 TensorFlow 1 中的默认模式。在图 12-4 中，您可以看到tf__sum_squares()函数被调用，其参数是一个符号张量（在这种情况下，一个形状为[]的 int32 张量），以及在跟踪期间生成的最终图。节点表示操作，箭头表示张量（生成的函数和图都被简化了）。

提示

为了查看生成的函数源代码，您可以调用tf.autograph.to_code(sum_squares.python_function)。代码并不一定要漂亮，但有时可以帮助调试。

TF 函数规则

大多数情况下，将执行 TensorFlow 操作的 Python 函数转换为 TF 函数是微不足道的：用@tf.function装饰它，或者让 Keras 为您处理。但是，有一些规则需要遵守：

• 如果调用任何外部库，包括 NumPy 甚至标准库，这个调用只会在跟踪期间运行；它不会成为图的一部分。实际上，TensorFlow 图只能包括 TensorFlow 构造（张量、操作、变量、数据集等）。因此，请确保使用tf.reduce_sum()而不是np.sum()，tf.sort()而不是内置的sorted()函数，等等（除非您真的希望代码只在跟踪期间运行）。这还有一些额外的影响：

  • 如果您定义了一个 TF 函数f(*x*)，它只返回np.random.rand()，那么只有在跟踪函数时才会生成一个随机数，因此f(tf.constant(2.))和f(tf.constant(3.))将返回相同的随机数，但f(tf.constant([2., 3.]))将返回一个不同的随机数。如果将np.random.rand()替换为tf.random.uniform([])，那么每次调用都会生成一个新的随机数，因为该操作将成为图的一部分。

  • 如果您的非 TensorFlow 代码具有副作用（例如记录某些内容或更新 Python 计数器），那么您不应该期望每次调用 TF 函数时都会发生这些副作用，因为它们只会在函数被跟踪时发生。

  • 您可以在 tf.py_function() 操作中包装任意的 Python 代码，但这样做会影响性能，因为 TensorFlow 将无法对此代码进行任何图优化。这也会降低可移植性，因为图仅在安装了正确库的平台上运行 Python 可用（和 Python 可用的平台）。

• 您可以调用其他 Python 函数或 TF 函数，但它们应该遵循相同的规则，因为 TensorFlow 将捕获它们的操作在计算图中。请注意，这些其他函数不需要用 @tf.function 装饰。

• 如果函数创建了 TensorFlow 变量（或任何其他有状态的 TensorFlow 对象，例如数据集或队列），它必须在第一次调用时才能这样做，否则您将收到异常。通常最好在 TF 函数之外创建变量（例如，在自定义层的 build() 方法中）。如果要为变量分配新值，请确保调用其 assign() 方法，而不是使用 = 运算符。

• 您的 Python 函数的源代码应该对 TensorFlow 可用。如果源代码不可用（例如，如果您在 Python shell 中定义函数，无法访问源代码，或者如果您仅将编译后的 *.pyc Python 文件部署到生产环境），则图生成过程将失败或功能有限。

• TensorFlow 仅会捕获对张量或 tf.data.Dataset 进行迭代的 for 循环（请参见第十三章）。因此，请确保使用 for i in tf.range(*x*) 而不是 for i in range(*x*)，否则循环将不会在图中被捕获。相反，它将在跟踪期间运行。（如果 for 循环旨在构建图，例如在神经网络中创建每个层，那么这可能是您想要的。）

• 一如既往，出于性能原因，您应该尽可能使用矢量化实现，而不是使用循环。

是时候总结了！在本章中，我们从 TensorFlow 的简要概述开始，然后看了 TensorFlow 的低级 API，包括张量、操作、变量和特殊数据结构。然后我们使用这些工具来自定义 Keras API 中的几乎每个组件。最后，我们看了 TF 函数如何提升性能，如何使用 AutoGraph 和跟踪生成图形，以及编写 TF 函数时应遵循的规则（如果您想进一步打开黑匣子并探索生成的图形，您将在附录 D 中找到技术细节）。

在下一章中，我们将学习如何使用 TensorFlow 高效加载和预处理数据。

练习

1. 您如何用简短的句子描述 TensorFlow？它的主要特点是什么？您能否列出其他流行的深度学习库？

2. TensorFlow 是否可以替代 NumPy？它们之间的主要区别是什么？

3. tf.range(10) 和 tf.constant(np.​ara⁠nge(10)) 会得到相同的结果吗？

4. 您能否列出 TensorFlow 中除了常规张量之外的其他六种数据结构？

5. 您可以通过编写函数或子类化 tf.keras.losses.Loss 类来定义自定义损失函数。您会在什么时候使用每个选项？

6. 同样，您可以在函数中定义自定义指标，也可以作为 tf.keras.metrics.Metric 的子类。您会在什么时候使用每个选项？

7. 何时应该创建自定义层而不是自定义模型？

8. 有哪些需要编写自定义训练循环的用例？

9. 自定义 Keras 组件可以包含任意的 Python 代码吗，还是必须可转换为 TF 函数？

10. 如果您希望函数可转换为 TF 函数，主要需要遵守哪些规则？

11. 何时需要创建一个动态的 Keras 模型？如何做到这一点？为什么不将所有模型都设置为动态的呢？

12. 实现一个执行层归一化的自定义层（我们将在第十五章中使用这种类型的层）：

    1. build()方法应该定义两个可训练的权重α和β，形状都是input_shape[-1:]，数据类型为tf.float32。α应该初始化为 1，β初始化为 0。

    2. call()方法应该计算每个实例特征的平均值μ和标准差σ。为此，您可以使用tf.nn.moments(inputs, axes=-1, keepdims=True)，它返回所有实例的平均值μ和方差σ²（计算方差的平方根以获得标准差）。然后函数应该计算并返回α ⊗ (X – μ)/(σ + ε) + β，其中 ⊗ 表示逐元素乘法（*），ε是一个平滑项（一个小常数，避免除以零，例如 0.001）。

    3. 确保您的自定义层产生与tf.keras.layers.LayerNormalization层相同（或非常接近）的输出。

13. 使用自定义训练循环训练一个模型，以处理 Fashion MNIST 数据集（参见第十章）：

    1. 显示每个时代、迭代、平均训练损失和每个时代的平均准确率（在每次迭代更新），以及每个时代结束时的验证损失和准确率。

    2. 尝试使用不同的优化器以及不同的学习率来处理上层和下层。

这些练习的解决方案可以在本章笔记本的末尾找到，网址为https://homl.info/colab3。

然而，Facebook 的 PyTorch 库目前在学术界更受欢迎：比起 TensorFlow 或 Keras，更多的论文引用 PyTorch。此外，Google 的 JAX 库正在获得动力，尤其是在学术界。

TensorFlow 包括另一个名为estimators API的深度学习 API，但现在已经不推荐使用。

如果您有需要（但您可能不会），您可以使用 C++ API 编写自己的操作。

要了解更多关于 TPU 以及它们如何工作的信息，请查看https://homl.info/tpus。

tf.math.log()是一个值得注意的例外，它通常被使用，但没有tf.log()的别名，因为这可能会与日志记录混淆。

使用加权平均值不是一个好主意：如果这样做，那么具有相同权重但在不同批次中的两个实例将对训练产生不同的影响，这取决于每个批次的总权重。

{**x, [...]}语法是在 Python 3.5 中添加的，用于将字典x中的所有键/值对合并到另一个字典中。自 Python 3.9 起，您可以使用更好的x | y语法（其中x和y是两个字典）。

然而，Huber 损失很少用作度量标准——通常更喜欢使用 MAE 或 MSE。

这个类仅用于说明目的。一个更简单和更好的实现方法是只需子类化tf.keras.metrics.Mean类；请参阅本章笔记本的“流式指标”部分以获取示例。

Keras API 将此参数称为input_shape，但由于它还包括批量维度，我更喜欢将其称为batch_input_shape。

在 Keras 中，“子类 API”通常只指通过子类化创建自定义模型，尽管在本章中您已经看到，许多其他东西也可以通过子类化创建。

由于 TensorFlow 问题＃46858，这种情况下调用super().build()可能会失败，除非在您阅读此内容时已修复该问题。如果没有，请将此行替换为self.built = True。

您还可以在模型内的任何层上调用add_loss()，因为模型会递归地从所有层中收集损失。

如果磁带超出范围，例如当使用它的函数返回时，Python 的垃圾收集器会为您删除它。

除了优化器之外，很少有人会自定义这些；请参阅笔记本中的“自定义优化器”部分以获取示例。

然而，在这个简单的例子中，计算图非常小，几乎没有任何优化的空间，所以tf_cube()实际上比cube()运行得慢得多。

第十三章：使用 TensorFlow 加载和预处理数据

在第二章中，您看到加载和预处理数据是任何机器学习项目的重要部分。您使用 Pandas 加载和探索（修改后的）加利福尼亚房屋数据集——该数据集存储在 CSV 文件中——并应用 Scikit-Learn 的转换器进行预处理。这些工具非常方便，您可能会经常使用它们，特别是在探索和实验数据时。

然而，在大型数据集上训练 TensorFlow 模型时，您可能更喜欢使用 TensorFlow 自己的数据加载和预处理 API，称为tf.data。它能够非常高效地加载和预处理数据，使用多线程和排队从多个文件中并行读取数据，对样本进行洗牌和分批处理等。此外，它可以实时执行所有这些操作——在 GPU 或 TPU 正在训练当前批次数据时，它会在多个 CPU 核心上加载和预处理下一批数据。

tf.data API 允许您处理无法放入内存的数据集，并充分利用硬件资源，从而加快训练速度。tf.data API 可以直接从文本文件（如 CSV 文件）、具有固定大小记录的二进制文件以及使用 TensorFlow 的 TFRecord 格式的二进制文件中读取数据。

TFRecord 是一种灵活高效的二进制格式，通常包含协议缓冲区（一种开源二进制格式）。tf.data API 还支持从 SQL 数据库中读取数据。此外，许多开源扩展可用于从各种数据源中读取数据，例如 Google 的 BigQuery 服务（请参阅https://tensorflow.org/io）。

Keras 还提供了强大而易于使用的预处理层，可以嵌入到您的模型中：这样，当您将模型部署到生产环境时，它将能够直接摄取原始数据，而无需您添加任何额外的预处理代码。这消除了训练期间使用的预处理代码与生产中使用的预处理代码之间不匹配的风险，这可能会导致训练/服务偏差。如果您将模型部署在使用不同编程语言编写的多个应用程序中，您不必多次重新实现相同的预处理代码，这也减少了不匹配的风险。

正如您将看到的，这两个 API 可以联合使用——例如，从 tf.data 提供的高效数据加载和 Keras 预处理层的便利性中受益。

在本章中，我们将首先介绍 tf.data API 和 TFRecord 格式。然后我们将探索 Keras 预处理层以及如何将它们与 tf.data API 一起使用。最后，我们将快速查看一些相关的库，您可能会发现它们在加载和预处理数据时很有用，例如 TensorFlow Datasets 和 TensorFlow Hub。所以，让我们开始吧！

tf.data API

整个 tf.data API 围绕着 tf.data.Dataset 的概念展开：这代表了一系列数据项。通常，您会使用逐渐从磁盘读取数据的数据集，但为了简单起见，让我们使用 tf.data.Dataset.from_tensor_slices() 从一个简单的数据张量创建数据集：

>>> import tensorflow as tf
>>> X = tf.range(10)  # any data tensor
>>> dataset = tf.data.Dataset.from_tensor_slices(X)
>>> dataset
<TensorSliceDataset shapes: (), types: tf.int32>

from_tensor_slices() 函数接受一个张量，并创建一个 tf.data.Dataset，其中的元素是沿着第一维度的所有 X 的切片，因此这个数据集包含 10 个项目：张量 0、1、2、…​、9。在这种情况下，如果我们使用 tf.data.Dataset.range(10)，我们将获得相同的数据集（除了元素将是 64 位整数而不是 32 位整数）。

您可以简单地迭代数据集的项目，如下所示：

>>> for item in dataset:
...     print(item)
...
tf.Tensor(0, shape=(), dtype=int32)
tf.Tensor(1, shape=(), dtype=int32)
[...]
tf.Tensor(9, shape=(), dtype=int32)

注意

tf.data API 是一个流式 API：您可以非常高效地迭代数据集的项目，但该 API 不适用于索引或切片。

数据集还可以包含张量的元组，或名称/张量对的字典，甚至是张量的嵌套元组和字典。在对元组、字典或嵌套结构进行切片时，数据集将仅切片它包含的张量，同时保留元组/字典结构。例如：

>>> X_nested = {"a": ([1, 2, 3], [4, 5, 6]), "b": [7, 8, 9]}
>>> dataset = tf.data.Dataset.from_tensor_slices(X_nested)
>>> for item in dataset:
...     print(item)
...
{'a': (<tf.Tensor: [...]=1>, <tf.Tensor: [...]=4>), 'b': <tf.Tensor: [...]=7>}
{'a': (<tf.Tensor: [...]=2>, <tf.Tensor: [...]=5>), 'b': <tf.Tensor: [...]=8>}
{'a': (<tf.Tensor: [...]=3>, <tf.Tensor: [...]=6>), 'b': <tf.Tensor: [...]=9>}

链接转换

一旦您有了数据集，您可以通过调用其转换方法对其应用各种转换。每个方法都会返回一个新的数据集，因此您可以像这样链接转换（此链在图 13-1 中有示例）：

>>> dataset = tf.data.Dataset.from_tensor_slices(tf.range(10))
>>> dataset = dataset.repeat(3).batch(7)
>>> for item in dataset:
...     print(item)
...
tf.Tensor([0 1 2 3 4 5 6], shape=(7,), dtype=int32)
tf.Tensor([7 8 9 0 1 2 3], shape=(7,), dtype=int32)
tf.Tensor([4 5 6 7 8 9 0], shape=(7,), dtype=int32)
tf.Tensor([1 2 3 4 5 6 7], shape=(7,), dtype=int32)
tf.Tensor([8 9], shape=(2,), dtype=int32)

在这个例子中，我们首先在原始数据集上调用repeat()方法，它返回一个将原始数据集的项目重复三次的新数据集。当然，这不会将所有数据在内存中复制三次！如果您调用此方法而没有参数，新数据集将永远重复源数据集，因此迭代数据集的代码将不得不决定何时停止。

然后我们在这个新数据集上调用batch()方法，再次创建一个新数据集。这个新数据集将把前一个数据集的项目分组成七个项目一组的批次。

图 13-1. 链接数据集转换

最后，我们迭代这个最终数据集的项目。batch()方法必须输出一个大小为两而不是七的最终批次，但是如果您希望删除这个最终批次，使所有批次具有完全相同的大小，可以调用batch()并使用drop_remainder=True。

警告

数据集方法不会修改数据集，它们会创建新的数据集。因此，请确保保留对这些新数据集的引用（例如，使用dataset = ...），否则什么也不会发生。

您还可以通过调用map()方法来转换项目。例如，这将创建一个所有批次乘以二的新数据集：

>>> dataset = dataset.map(lambda x: x * 2)  # x is a batch
>>> for item in dataset:
...     print(item)
...
tf.Tensor([ 0  2  4  6  8 10 12], shape=(7,), dtype=int32)
tf.Tensor([14 16 18  0  2  4  6], shape=(7,), dtype=int32)
[...]

这个map()方法是您将调用的方法，用于对数据进行任何预处理。有时这将包括一些可能相当密集的计算，比如重塑或旋转图像，因此您通常会希望启动多个线程以加快速度。这可以通过将num_parallel_calls参数设置为要运行的线程数，或者设置为tf.data.AUTOTUNE来完成。请注意，您传递给map()方法的函数必须可以转换为 TF 函数（请参阅第十二章）。

还可以使用filter()方法简单地过滤数据集。例如，此代码创建一个仅包含总和大于 50 的批次的数据集：

>>> dataset = dataset.filter(lambda x: tf.reduce_sum(x) > 50)
>>> for item in dataset:
...     print(item)
...
tf.Tensor([14 16 18  0  2  4  6], shape=(7,), dtype=int32)
tf.Tensor([ 8 10 12 14 16 18  0], shape=(7,), dtype=int32)
tf.Tensor([ 2  4  6  8 10 12 14], shape=(7,), dtype=int32)

您经常会想查看数据集中的一些项目。您可以使用take()方法来实现：

>>> for item in dataset.take(2):
...     print(item)
...
tf.Tensor([14 16 18  0  2  4  6], shape=(7,), dtype=int32)
tf.Tensor([ 8 10 12 14 16 18  0], shape=(7,), dtype=int32)

数据洗牌

正如我们在第四章中讨论的，梯度下降在训练集中的实例是独立且同分布（IID）时效果最好。确保这一点的一个简单方法是对实例进行洗牌，使用shuffle()方法。它将创建一个新数据集，首先用源数据集的前几个项目填充缓冲区。然后，每当需要一个项目时，它将从缓冲区随机取出一个项目，并用源数据集中的新项目替换它，直到完全迭代源数据集。在这一点上，它将继续从缓冲区随机取出项目，直到缓冲区为空。您必须指定缓冲区大小，并且很重要的是要足够大，否则洗牌效果不会很好。⁠¹ 只是不要超出您拥有的 RAM 量，尽管即使您有很多 RAM，也没有必要超出数据集的大小。如果您希望每次运行程序时都获得相同的随机顺序，可以提供一个随机种子。例如，以下代码创建并显示一个包含 0 到 9 的整数，重复两次，使用大小为 4 的缓冲区和随机种子 42 进行洗牌，并使用批次大小为 7 进行批处理的数据集：

>>> dataset = tf.data.Dataset.range(10).repeat(2)
>>> dataset = dataset.shuffle(buffer_size=4, seed=42).batch(7)
>>> for item in dataset:
...     print(item)
...
tf.Tensor([3 0 1 6 2 5 7], shape=(7,), dtype=int64)
tf.Tensor([8 4 1 9 4 2 3], shape=(7,), dtype=int64)
tf.Tensor([7 5 0 8 9 6], shape=(6,), dtype=int64)

提示

如果在打乱的数据集上调用repeat()，默认情况下它将在每次迭代时生成一个新的顺序。这通常是个好主意，但是如果您希望在每次迭代中重复使用相同的顺序（例如，用于测试或调试），可以在调用shuffle()时设置reshuffle_each_​itera⁠tion=False。

对于一个无法放入内存的大型数据集，这种简单的打乱缓冲区方法可能不够，因为缓冲区相对于数据集来说很小。一个解决方案是对源数据本身进行打乱（例如，在 Linux 上可以使用shuf命令对文本文件进行打乱）。这将显著改善打乱效果！即使源数据已经被打乱，通常也会希望再次打乱，否则每个时期将重复相同的顺序，模型可能会出现偏差（例如，由于源数据顺序中偶然存在的一些虚假模式）。为了进一步打乱实例，一个常见的方法是将源数据拆分为多个文件，然后在训练过程中以随机顺序读取它们。然而，位于同一文件中的实例仍然会相互靠近。为了避免这种情况，您可以随机选择多个文件并同时读取它们，交错它们的记录。然后在此基础上使用shuffle()方法添加一个打乱缓冲区。如果这听起来很费力，不用担心：tf.data API 可以在几行代码中实现所有这些。让我们看看您可以如何做到这一点。

从多个文件中交错行

首先，假设您已经加载了加利福尼亚房屋数据集，对其进行了打乱（除非已经打乱），并将其分为训练集、验证集和测试集。然后将每个集合分成许多 CSV 文件，每个文件看起来像这样（每行包含八个输入特征加上目标中位房价）：

MedInc,HouseAge,AveRooms,AveBedrms,Popul…,AveOccup,Lat…,Long…,MedianHouseValue
3.5214,15.0,3.050,1.107,1447.0,1.606,37.63,-122.43,1.442
5.3275,5.0,6.490,0.991,3464.0,3.443,33.69,-117.39,1.687
3.1,29.0,7.542,1.592,1328.0,2.251,38.44,-122.98,1.621
[...]

假设train_filepaths包含训练文件路径列表（您还有valid_filepaths和test_filepaths）：

>>> train_filepaths
['datasets/housing/my_train_00.csv', 'datasets/housing/my_train_01.csv', ...]

或者，您可以使用文件模式；例如，train_filepaths = "datasets/housing/my_train_*.csv"。现在让我们创建一个仅包含这些文件路径的数据集：

filepath_dataset = tf.data.Dataset.list_files(train_filepaths, seed=42)

默认情况下，list_files()函数返回一个打乱文件路径的数据集。一般来说这是件好事，但是如果出于某种原因不想要这样，可以设置shuffle=False。

接下来，您可以调用interleave()方法一次从五个文件中读取并交错它们的行。您还可以使用skip()方法跳过每个文件的第一行（即标题行）：

n_readers = 5
dataset = filepath_dataset.interleave(
    lambda filepath: tf.data.TextLineDataset(filepath).skip(1),
    cycle_length=n_readers)

interleave()方法将创建一个数据集，从filepath_dataset中提取五个文件路径，对于每个文件路径，它将调用您提供的函数（在本例中是 lambda 函数）来创建一个新的数据集（在本例中是TextLineDataset）。清楚地说，在这个阶段总共会有七个数据集：文件路径数据集、交错数据集以及交错数据集内部创建的五个TextLineDataset。当您迭代交错数据集时，它将循环遍历这五个TextLineDataset，从每个数据集中逐行读取，直到所有数据集都用完。然后它将从filepath_dataset中获取下一个五个文件路径，并以相同的方式交错它们，依此类推，直到文件路径用完。为了使交错效果最佳，最好拥有相同长度的文件；否则最长文件的末尾将不会被交错。

默认情况下，interleave()不使用并行处理；它只是顺序地从每个文件中一次读取一行。如果您希望实际并行读取文件，可以将interleave()方法的num_parallel_calls参数设置为您想要的线程数（请记住，map()方法也有这个参数）。甚至可以将其设置为tf.data.AUTOTUNE，让 TensorFlow 根据可用的 CPU 动态选择正确的线程数。现在让我们看看数据集现在包含什么：

>>> for line in dataset.take(5):
...     print(line)
...
tf.Tensor(b'4.5909,16.0,[...],33.63,-117.71,2.418', shape=(), dtype=string)
tf.Tensor(b'2.4792,24.0,[...],34.18,-118.38,2.0', shape=(), dtype=string)
tf.Tensor(b'4.2708,45.0,[...],37.48,-122.19,2.67', shape=(), dtype=string)
tf.Tensor(b'2.1856,41.0,[...],32.76,-117.12,1.205', shape=(), dtype=string)
tf.Tensor(b'4.1812,52.0,[...],33.73,-118.31,3.215', shape=(), dtype=string)

这些是随机选择的五个 CSV 文件的第一行（忽略标题行）。看起来不错！

注意

可以将文件路径列表传递给 TextLineDataset 构造函数：它将按顺序遍历每个文件的每一行。如果还将 num_parallel_reads 参数设置为大于一的数字，那么数据集将并行读取该数量的文件，并交错它们的行（无需调用 interleave() 方法）。但是，它不会对文件进行洗牌，也不会跳过标题行。

数据预处理

现在我们有一个返回每个实例的住房数据集，其中包含一个字节字符串的张量，我们需要进行一些预处理，包括解析字符串和缩放数据。让我们实现一些自定义函数来执行这些预处理：

X_mean, X_std = [...]  # mean and scale of each feature in the training set
n_inputs = 8

def parse_csv_line(line):
    defs = [0.] * n_inputs + [tf.constant([], dtype=tf.float32)]
    fields = tf.io.decode_csv(line, record_defaults=defs)
    return tf.stack(fields[:-1]), tf.stack(fields[-1:])

def preprocess(line):
    x, y = parse_csv_line(line)
    return (x - X_mean) / X_std, y

让我们逐步解释这段代码：

• 首先，代码假设我们已经预先计算了训练集中每个特征的均值和标准差。X_mean 和 X_std 只是包含八个浮点数的 1D 张量（或 NumPy 数组），每个输入特征一个。可以使用 Scikit-Learn 的 StandardScaler 在数据集的足够大的随机样本上完成这个操作。在本章的后面，我们将使用 Keras 预处理层来代替。

• parse_csv_line() 函数接受一个 CSV 行并对其进行解析。为了帮助实现这一点，它使用 tf.io.decode_csv() 函数，该函数接受两个参数：第一个是要解析的行，第二个是包含 CSV 文件中每列的默认值的数组。这个数组（defs）告诉 TensorFlow 不仅每列的默认值是什么，还告诉它列的数量和类型。在这个例子中，我们告诉它所有特征列都是浮点数，缺失值应默认为零，但我们为最后一列（目标）提供了一个空的 tf.float32 类型的默认值数组：该数组告诉 TensorFlow 这一列包含浮点数，但没有默认值，因此如果遇到缺失值，它将引发异常。

• tf.io.decode_csv() 函数返回一个标量张量列表（每列一个），但我们需要返回一个 1D 张量数组。因此，我们对除最后一个（目标）之外的所有张量调用 tf.stack()：这将这些张量堆叠成一个 1D 数组。然后我们对目标值做同样的操作：这将使其成为一个包含单个值的 1D 张量数组，而不是标量张量。tf.io.decode_csv() 函数完成后，它将返回输入特征和目标。

• 最后，自定义的 preprocess() 函数只调用 parse_csv_line() 函数，通过减去特征均值然后除以特征标准差来缩放输入特征，并返回一个包含缩放特征和目标的元组。

让我们测试这个预处理函数：

>>> preprocess(b'4.2083,44.0,5.3232,0.9171,846.0,2.3370,37.47,-122.2,2.782')
(<tf.Tensor: shape=(8,), dtype=float32, numpy=
 array([ 0.16579159,  1.216324  , -0.05204564, -0.39215982, -0.5277444 ,
 -0.2633488 ,  0.8543046 , -1.3072058 ], dtype=float32)>,
 <tf.Tensor: shape=(1,), dtype=float32, numpy=array([2.782], dtype=float32)>)

看起来不错！preprocess() 函数可以将一个实例从字节字符串转换为一个漂亮的缩放张量，带有相应的标签。我们现在可以使用数据集的 map() 方法将 preprocess() 函数应用于数据集中的每个样本。

将所有内容放在一起

为了使代码更具重用性，让我们将迄今为止讨论的所有内容放在另一个辅助函数中；它将创建并返回一个数据集，该数据集将高效地从多个 CSV 文件中加载加利福尼亚房屋数据，对其进行预处理、洗牌和分批处理（参见图 13-2）：

def csv_reader_dataset(filepaths, n_readers=5, n_read_threads=None,
                       n_parse_threads=5, shuffle_buffer_size=10_000, seed=42,
                       batch_size=32):
    dataset = tf.data.Dataset.list_files(filepaths, seed=seed)
    dataset = dataset.interleave(
        lambda filepath: tf.data.TextLineDataset(filepath).skip(1),
        cycle_length=n_readers, num_parallel_calls=n_read_threads)
    dataset = dataset.map(preprocess, num_parallel_calls=n_parse_threads)
    dataset = dataset.shuffle(shuffle_buffer_size, seed=seed)
    return dataset.batch(batch_size).prefetch(1)

请注意，我们在最后一行使用了 prefetch() 方法。这对性能很重要，你现在会看到。

图 13-2. 从多个 CSV 文件加载和预处理数据

预取

通过在自定义csv_reader_dataset()函数末尾调用prefetch(1)，我们正在创建一个数据集，该数据集将尽力始终领先一个批次。换句话说，当我们的训练算法在处理一个批次时，数据集将已经在并行工作，准备好获取下一个批次（例如，从磁盘读取数据并对其进行预处理）。这可以显著提高性能，如图 13-3 所示。

如果我们还确保加载和预处理是多线程的（通过在调用interleave()和map()时设置num_parallel_calls），我们可以利用多个 CPU 核心，希望准备一个数据批次的时间比在 GPU 上运行训练步骤要短：这样 GPU 将几乎 100%利用（除了从 CPU 到 GPU 的数据传输时间）[3]，训练将运行得更快。

图 13-3。通过预取，CPU 和 GPU 并行工作：当 GPU 处理一个批次时，CPU 处理下一个批次

提示

如果您计划购买 GPU 卡，其处理能力和内存大小当然非常重要（特别是对于大型计算机视觉或自然语言处理模型，大量的 RAM 至关重要）。对于良好性能同样重要的是 GPU 的内存带宽；这是它每秒可以将多少千兆字节的数据进出其 RAM。

如果数据集足够小，可以放入内存，您可以通过使用数据集的cache()方法将其内容缓存到 RAM 来显着加快训练速度。通常应在加载和预处理数据之后，但在洗牌、重复、批处理和预取之前执行此操作。这样，每个实例只会被读取和预处理一次（而不是每个时期一次），但数据仍然会在每个时期以不同的方式洗牌，下一批数据仍然会提前准备好。

您现在已经学会了如何构建高效的输入管道，从多个文本文件加载和预处理数据。我们已经讨论了最常见的数据集方法，但还有一些您可能想看看的方法，例如concatenate()、zip()、window()、reduce()、shard()、flat_map()、apply()、unbatch()和padded_batch()。还有一些更多的类方法，例如from_generator()和from_tensors()，它们分别从 Python 生成器或张量列表创建新数据集。请查看 API 文档以获取更多详细信息。还请注意，tf.data.experimental中提供了一些实验性功能，其中许多功能可能会在未来的版本中成为核心 API 的一部分（例如，请查看CsvDataset类，以及make_csv_dataset()方法，该方法负责推断每列的类型）。

使用数据集与 Keras

现在，我们可以使用我们之前编写的自定义csv_reader_dataset()函数为训练集、验证集和测试集创建数据集。训练集将在每个时期进行洗牌（请注意，验证集和测试集也将进行洗牌，尽管我们实际上并不需要）：

train_set = csv_reader_dataset(train_filepaths)
valid_set = csv_reader_dataset(valid_filepaths)
test_set = csv_reader_dataset(test_filepaths)

现在，您可以简单地使用这些数据集构建和训练 Keras 模型。当您调用模型的fit()方法时，您传递train_set而不是X_train, y_train，并传递validation_data=valid_set而不是validation_data=(X_valid, y_valid)。fit()方法将负责每个时期重复训练数据集，每个时期使用不同的随机顺序：

model = tf.keras.Sequential([...])
model.compile(loss="mse", optimizer="sgd")
model.fit(train_set, validation_data=valid_set, epochs=5)

同样，您可以将数据集传递给evaluate()和predict()方法：

test_mse = model.evaluate(test_set)
new_set = test_set.take(3)  # pretend we have 3 new samples
y_pred = model.predict(new_set)  # or you could just pass a NumPy array

与其他数据集不同，new_set通常不包含标签。如果包含标签，就像这里一样，Keras 会忽略它们。请注意，在所有这些情况下，您仍然可以使用 NumPy 数组而不是数据集（但当然它们需要先加载和预处理）。

如果您想构建自己的自定义训练循环（如第十二章中讨论的），您可以很自然地遍历训练集：

n_epochs = 5
for epoch in range(n_epochs):
    for X_batch, y_batch in train_set:
        [...]  # perform one gradient descent step

实际上，甚至可以创建一个 TF 函数（参见第十二章），用于整个时期训练模型。这可以真正加快训练速度：

@tf.function
def train_one_epoch(model, optimizer, loss_fn, train_set):
    for X_batch, y_batch in train_set:
        with tf.GradientTape() as tape:
            y_pred = model(X_batch)
            main_loss = tf.reduce_mean(loss_fn(y_batch, y_pred))
            loss = tf.add_n([main_loss] + model.losses)
        gradients = tape.gradient(loss, model.trainable_variables)
        optimizer.apply_gradients(zip(gradients, model.trainable_variables))

optimizer = tf.keras.optimizers.SGD(learning_rate=0.01)
loss_fn = tf.keras.losses.mean_squared_error
for epoch in range(n_epochs):
    print("rEpoch {}/{}".format(epoch + 1, n_epochs), end="")
    train_one_epoch(model, optimizer, loss_fn, train_set)

在 Keras 中，compile()方法的steps_per_execution参数允许您定义fit()方法在每次调用用于训练的tf.function时将处理的批次数。默认值只是 1，因此如果将其设置为 50，您通常会看到显着的性能改进。但是，Keras 回调的on_batch_*()方法只会在每 50 批次时调用一次。

恭喜，您现在知道如何使用 tf.data API 构建强大的输入管道！然而，到目前为止，我们一直在使用常见、简单和方便但不是真正高效的 CSV 文件，并且不太支持大型或复杂的数据结构（如图像或音频）。因此，让我们看看如何改用 TFRecords。

提示

如果您对 CSV 文件（或者您正在使用的其他格式）感到满意，您不一定必须使用 TFRecords。俗话说，如果它没有坏，就不要修理！当训练过程中的瓶颈是加载和解析数据时，TFRecords 非常有用。

TFRecord 格式

TFRecord 格式是 TensorFlow 存储大量数据并高效读取的首选格式。它是一个非常简单的二进制格式，只包含一系列大小不同的二进制记录（每个记录由长度、用于检查长度是否损坏的 CRC 校验和、实际数据，最后是数据的 CRC 校验和组成）。您可以使用tf.io.TFRecordWriter类轻松创建 TFRecord 文件：

with tf.io.TFRecordWriter("my_data.tfrecord") as f:
    f.write(b"This is the first record")
    f.write(b"And this is the second record")

然后，您可以使用tf.data.TFRecordDataset来读取一个或多个 TFRecord 文件：

filepaths = ["my_data.tfrecord"]
dataset = tf.data.TFRecordDataset(filepaths)
for item in dataset:
    print(item)

这将输出：

tf.Tensor(b'This is the first record', shape=(), dtype=string)
tf.Tensor(b'And this is the second record', shape=(), dtype=string)

提示

默认情况下，TFRecordDataset将逐个读取文件，但您可以使其并行读取多个文件，并通过传递文件路径列表给构造函数并将num_parallel_reads设置为大于 1 的数字来交错它们的记录。或者，您可以通过使用list_files()和interleave()来获得与我们之前读取多个 CSV 文件相同的结果。

压缩的 TFRecord 文件

有时将 TFRecord 文件压缩可能很有用，特别是如果它们需要通过网络连接加载。您可以通过设置options参数创建一个压缩的 TFRecord 文件：

options = tf.io.TFRecordOptions(compression_type="GZIP")
with tf.io.TFRecordWriter("my_compressed.tfrecord", options) as f:
    f.write(b"Compress, compress, compress!")

在读取压缩的 TFRecord 文件时，您需要指定压缩类型：

dataset = tf.data.TFRecordDataset(["my_compressed.tfrecord"],
                                  compression_type="GZIP")

协议缓冲区简介

尽管每个记录可以使用您想要的任何二进制格式，但 TFRecord 文件通常包含序列化的协议缓冲区（也称为protobufs）。这是一个在 2001 年由谷歌开发的便携式、可扩展和高效的二进制格式，并于 2008 年开源；protobufs 现在被广泛使用，特别是在grpc中，谷歌的远程过程调用系统。它们使用一个看起来像这样的简单语言进行定义：

syntax = "proto3";
message Person {
    string name = 1;
    int32 id = 2;
    repeated string email = 3;
}

这个 protobuf 定义表示我们正在使用 protobuf 格式的第 3 版，并且指定每个Person对象（可选）可能具有一个字符串类型的name、一个 int32 类型的id，以及零个或多个字符串类型的email字段。数字1、2和3是字段标识符：它们将在每个记录的二进制表示中使用。一旦你在*.proto文件中有了一个定义，你就可以编译它。这需要使用 protobuf 编译器protoc在 Python（或其他语言）中生成访问类。请注意，你通常在 TensorFlow 中使用的 protobuf 定义已经为你编译好了，并且它们的 Python 类是 TensorFlow 库的一部分，因此你不需要使用protoc。你只需要知道如何在 Python 中使用*protobuf 访问类。为了说明基础知识，让我们看一个简单的示例，使用为Personprotobuf 生成的访问类（代码在注释中有解释）：

>>> from person_pb2 import Person  # import the generated access class
>>> person = Person(name="Al", id=123, email=["a@b.com"])  # create a Person
>>> print(person)  # display the Person
name: "Al"
id: 123
email: "a@b.com"
>>> person.name  # read a field
'Al'
>>> person.name = "Alice"  # modify a field
>>> person.email[0]  # repeated fields can be accessed like arrays
'a@b.com'
>>> person.email.append("c@d.com")  # add an email address
>>> serialized = person.SerializeToString()  # serialize person to a byte string
>>> serialized
b'nx05Alicex10{x1ax07a@b.comx1ax07c@d.com'
>>> person2 = Person()  # create a new Person
>>> person2.ParseFromString(serialized)  # parse the byte string (27 bytes long)
27
>>> person == person2  # now they are equal
True

简而言之，我们导入由protoc生成的Person类，创建一个实例并对其进行操作，可视化它并读取和写入一些字段，然后使用SerializeToString()方法对其进行序列化。这是准备保存或通过网络传输的二进制数据。当读取或接收这些二进制数据时，我们可以使用ParseFromString()方法进行解析，并获得被序列化的对象的副本。

你可以将序列化的Person对象保存到 TFRecord 文件中，然后加载和解析它：一切都会正常工作。然而，ParseFromString()不是一个 TensorFlow 操作，所以你不能在 tf.data 管道中的预处理函数中使用它（除非将其包装在tf.py_function()操作中，这会使代码变慢且不太可移植，正如你在第十二章中看到的）。然而，你可以使用tf.io.decode_proto()函数，它可以解析任何你想要的 protobuf，只要你提供 protobuf 定义（请参考笔记本中的示例）。也就是说，在实践中，你通常会希望使用 TensorFlow 提供的专用解析操作的预定义 protobuf。现在让我们来看看这些预定义的 protobuf。

TensorFlow Protobufs

TFRecord 文件中通常使用的主要 protobuf 是Exampleprotobuf，它表示数据集中的一个实例。它包含一个命名特征列表，其中每个特征可以是一个字节字符串列表、一个浮点数列表或一个整数列表。以下是 protobuf 定义（来自 TensorFlow 源代码）：

syntax = "proto3";
message BytesList { repeated bytes value = 1; }
message FloatList { repeated float value = 1 [packed = true]; }
message Int64List { repeated int64 value = 1 [packed = true]; }
message Feature {
    oneof kind {
        BytesList bytes_list = 1;
        FloatList float_list = 2;
        Int64List int64_list = 3;
    }
};
message Features { map<string, Feature> feature = 1; };
message Example { Features features = 1; };

BytesList、FloatList和Int64List的定义足够简单明了。请注意，对于重复的数值字段，使用[packed = true]进行更有效的编码。Feature包含一个BytesList、一个FloatList或一个Int64List。一个Features（带有s）包含一个将特征名称映射到相应特征值的字典。最后，一个Example只包含一个Features对象。

注意

为什么会定义Example，因为它只包含一个Features对象？嗯，TensorFlow 的开发人员可能有一天决定向其中添加更多字段。只要新的Example定义仍然包含相同 ID 的features字段，它就是向后兼容的。这种可扩展性是 protobuf 的一个伟大特性。

这是你如何创建一个代表同一个人的tf.train.Example：

from tensorflow.train import BytesList, FloatList, Int64List
from tensorflow.train import Feature, Features, Example

person_example = Example(
    features=Features(
        feature={
            "name": Feature(bytes_list=BytesList(value=[b"Alice"])),
            "id": Feature(int64_list=Int64List(value=[123])),
            "emails": Feature(bytes_list=BytesList(value=[b"a@b.com",
                                                          b"c@d.com"]))
        }))

这段代码有点冗长和重复，但你可以很容易地将其包装在一个小的辅助函数中。现在我们有了一个Example protobuf，我们可以通过调用其SerializeToString()方法将其序列化，然后将生成的数据写入 TFRecord 文件。让我们假装写入五次，以假装我们有几个联系人：

with tf.io.TFRecordWriter("my_contacts.tfrecord") as f:
    for _ in range(5):
        f.write(person_example.SerializeToString())

通常，您会写比五个Example更多的内容！通常情况下，您会创建一个转换脚本，从当前格式（比如 CSV 文件）读取数据，为每个实例创建一个Example protobuf，将它们序列化，并保存到几个 TFRecord 文件中，最好在此过程中对它们进行洗牌。这需要一些工作，所以再次确保这确实是必要的（也许您的流水线使用 CSV 文件运行良好）。

现在我们有一个包含多个序列化Example的漂亮 TFRecord 文件，让我们尝试加载它。

加载和解析示例

为了加载序列化的Example protobufs，我们将再次使用tf.data.TFRecordDataset，并使用tf.io.parse_single_example()解析每个Example。它至少需要两个参数：包含序列化数据的字符串标量张量，以及每个特征的描述。描述是一个字典，将每个特征名称映射到tf.io.FixedLenFeature描述符，指示特征的形状、类型和默认值，或者tf.io.VarLenFeature描述符，仅指示特征列表的长度可能变化的类型（例如"emails"特征）。

以下代码定义了一个描述字典，然后创建了一个TFRecordDataset，并对其应用了一个自定义预处理函数，以解析该数据集包含的每个序列化Example protobuf：

feature_description = {
    "name": tf.io.FixedLenFeature([], tf.string, default_value=""),
    "id": tf.io.FixedLenFeature([], tf.int64, default_value=0),
    "emails": tf.io.VarLenFeature(tf.string),
}

def parse(serialized_example):
    return tf.io.parse_single_example(serialized_example, feature_description)

dataset = tf.data.TFRecordDataset(["my_contacts.tfrecord"]).map(parse)
for parsed_example in dataset:
    print(parsed_example)

固定长度的特征被解析为常规张量，但变长特征被解析为稀疏张量。您可以使用tf.sparse.to_dense()将稀疏张量转换为密集张量，但在这种情况下，更简单的方法是直接访问其值：

>>> tf.sparse.to_dense(parsed_example["emails"], default_value=b"")
<tf.Tensor: [...] dtype=string, numpy=array([b'a@b.com', b'c@d.com'], [...])>
>>> parsed_example["emails"].values
<tf.Tensor: [...] dtype=string, numpy=array([b'a@b.com', b'c@d.com'], [...])>

您可以使用tf.io.parse_example()批量解析示例，而不是使用tf.io.parse_single_example()逐个解析它们：

def parse(serialized_examples):
    return tf.io.parse_example(serialized_examples, feature_description)

dataset = tf.data.TFRecordDataset(["my_contacts.tfrecord"]).batch(2).map(parse)
for parsed_examples in dataset:
    print(parsed_examples)  # two examples at a time

最后，BytesList可以包含您想要的任何二进制数据，包括任何序列化对象。例如，您可以使用tf.io.encode_jpeg()使用 JPEG 格式对图像进行编码，并将这些二进制数据放入BytesList中。稍后，当您的代码读取 TFRecord 时，它将从解析Example开始，然后需要调用tf.io.decode_jpeg()来解析数据并获取原始图像（或者您可以使用tf.io.decode_image()，它可以解码任何 BMP、GIF、JPEG 或 PNG 图像）。您还可以通过使用tf.io.serialize_tensor()对张量进行序列化，然后将生成的字节字符串放入BytesList特征中，将任何您想要的张量存储在BytesList中。稍后，当您解析 TFRecord 时，您可以使用tf.io.parse_tensor()解析这些数据。请参阅本章的笔记本https://homl.info/colab3 ，了解在 TFRecord 文件中存储图像和张量的示例。

正如您所看到的，Example protobuf 非常灵活，因此对于大多数用例来说可能已经足够了。但是，当您处理列表列表时，可能会有些繁琐。例如，假设您想对文本文档进行分类。每个文档可以表示为一个句子列表，其中每个句子表示为一个单词列表。也许每个文档还有一个评论列表，其中每个评论表示为一个单词列表。还可能有一些上下文数据，比如文档的作者、标题和发布日期。TensorFlow 的SequenceExample protobuf 就是为这种用例而设计的。

使用 SequenceExample Protobuf 处理列表列表

这是SequenceExample protobuf 的定义：

message FeatureList { repeated Feature feature = 1; };
message FeatureLists { map<string, FeatureList> feature_list = 1; };
message SequenceExample {
    Features context = 1;
    FeatureLists feature_lists = 2;
};

SequenceExample包含一个Features对象用于上下文数据和一个包含一个或多个命名FeatureList对象（例如，一个名为"content"的FeatureList和另一个名为"comments"的FeatureList）的FeatureLists对象。每个FeatureList包含一个Feature对象列表，每个Feature对象可能是字节字符串列表、64 位整数列表或浮点数列表（在此示例中，每个Feature可能代表一个句子或评论，可能以单词标识符列表的形式）。构建SequenceExample、序列化它并解析它类似于构建、序列化和解析Example，但您必须使用tf.io.parse_single_sequence_example()来解析单个SequenceExample或tf.io.parse_sequence_example()来解析批处理。这两个函数返回一个包含上下文特征（作为字典）和特征列表（也作为字典）的元组。如果特征列表包含不同大小的序列（如前面的示例），您可能希望使用tf.RaggedTensor.from_sparse()将它们转换为不规则张量（请参阅完整代码的笔记本）：

parsed_context, parsed_feature_lists = tf.io.parse_single_sequence_example(
    serialized_sequence_example, context_feature_descriptions,
    sequence_feature_descriptions)
parsed_content = tf.RaggedTensor.from_sparse(parsed_feature_lists["content"])

现在您已经知道如何使用 tf.data API、TFRecords 和 protobufs 高效存储、加载、解析和预处理数据，是时候将注意力转向 Keras 预处理层了。

Keras 预处理层

为神经网络准备数据通常需要对数值特征进行归一化、对分类特征和文本进行编码、裁剪和调整图像等。有几种选项：

• 预处理可以提前在准备训练数据文件时完成，使用您喜欢的任何工具，如 NumPy、Pandas 或 Scikit-Learn。您需要在生产中应用完全相同的预处理步骤，以确保您的生产模型接收到与训练时相似的预处理输入。

• 或者，您可以在加载数据时使用 tf.data 进行即时预处理，通过使用该数据集的map()方法对数据集的每个元素应用预处理函数，就像本章前面所做的那样。同样，您需要在生产中应用相同的预处理步骤。

• 最后一种方法是直接在模型内部包含预处理层，这样它可以在训练期间即时预处理所有输入数据，然后在生产中使用相同的预处理层。本章的其余部分将讨论这种最后一种方法。

Keras 提供了许多预处理层，您可以将其包含在模型中：它们可以应用于数值特征、分类特征、图像和文本。我们将在接下来的部分中讨论数值和分类特征，以及基本文本预处理，我们将在第十四章中涵盖图像预处理，以及在第十六章中涵盖更高级的文本预处理。

归一化层

正如我们在第十章中看到的，Keras 提供了一个Normalization层，我们可以用来标准化输入特征。我们可以在创建层时指定每个特征的均值和方差，或者更简单地在拟合模型之前将训练集传递给该层的adapt()方法，以便该层可以在训练之前自行测量特征的均值和方差：

norm_layer = tf.keras.layers.Normalization()
model = tf.keras.models.Sequential([
    norm_layer,
    tf.keras.layers.Dense(1)
])
model.compile(loss="mse", optimizer=tf.keras.optimizers.SGD(learning_rate=2e-3))
norm_layer.adapt(X_train)  # computes the mean and variance of every feature
model.fit(X_train, y_train, validation_data=(X_valid, y_valid), epochs=5)

提示

传递给adapt()方法的数据样本必须足够大，以代表您的数据集，但不必是完整的训练集：对于Normalization层，从训练集中随机抽取的几百个实例通常足以获得特征均值和方差的良好估计。

由于我们在模型中包含了Normalization层，现在我们可以将这个模型部署到生产环境中，而不必再担心归一化的问题：模型会自动处理（参见图 13-4）。太棒了！这种方法完全消除了预处理不匹配的风险，当人们尝试为训练和生产维护不同的预处理代码，但更新其中一个并忘记更新另一个时，就会发生这种情况。生产模型最终会接收到以其不期望的方式预处理的数据。如果他们幸运的话，会得到一个明显的错误。如果不幸的话，模型的准确性会悄悄下降。

图 13-4。在模型中包含预处理层

直接在模型中包含预处理层很简单明了，但会减慢训练速度（在Normalization层的情况下只会稍微减慢）：实际上，由于预处理是在训练过程中实时进行的，每个时期只会发生一次。我们可以通过在训练之前仅对整个训练集进行一次归一化来做得更好。为此，我们可以像使用 Scikit-Learn 的StandardScaler一样单独使用Normalization层：

norm_layer = tf.keras.layers.Normalization()
norm_layer.adapt(X_train)
X_train_scaled = norm_layer(X_train)
X_valid_scaled = norm_layer(X_valid)

现在我们可以在经过缩放的数据上训练模型，这次不需要Normalization层：

model = tf.keras.models.Sequential([tf.keras.layers.Dense(1)])
model.compile(loss="mse", optimizer=tf.keras.optimizers.SGD(learning_rate=2e-3))
model.fit(X_train_scaled, y_train, epochs=5,
          validation_data=(X_valid_scaled, y_valid))

很好！这应该会加快训练速度。但是现在当我们将模型部署到生产环境时，模型不会对其输入进行预处理。为了解决这个问题，我们只需要创建一个新模型，将适应的Normalization层和刚刚训练的模型包装在一起。然后我们可以将这个最终模型部署到生产环境中，它将负责对其输入进行预处理和进行预测（参见图 13-5）：

final_model = tf.keras.Sequential([norm_layer, model])
X_new = X_test[:3]  # pretend we have a few new instances (unscaled)
y_pred = final_model(X_new)  # preprocesses the data and makes predictions

图 13-5。在训练之前仅对数据进行一次预处理，然后将这些层部署到最终模型中

现在我们拥有了最佳的两种方式：训练很快，因为我们只在训练开始前对数据进行一次预处理，而最终模型可以在运行时对其输入进行预处理，而不会有任何预处理不匹配的风险。

此外，Keras 预处理层与 tf.data API 很好地配合。例如，可以将tf.data.Dataset传递给预处理层的adapt()方法。还可以使用数据集的map()方法将 Keras 预处理层应用于tf.data.Dataset。例如，以下是如何将适应的Normalization层应用于数据集中每个批次的输入特征的方法：

dataset = dataset.map(lambda X, y: (norm_layer(X), y))

最后，如果您需要比 Keras 预处理层提供的更多特性，您可以随时编写自己的 Keras 层，就像我们在第十二章中讨论的那样。例如，如果Normalization层不存在，您可以使用以下自定义层获得类似的结果：

import numpy as np

class MyNormalization(tf.keras.layers.Layer):
    def adapt(self, X):
        self.mean_ = np.mean(X, axis=0, keepdims=True)
        self.std_ = np.std(X, axis=0, keepdims=True)

    def call(self, inputs):
        eps = tf.keras.backend.epsilon()  # a small smoothing term
        return (inputs - self.mean_) / (self.std_ + eps)

接下来，让我们看看另一个用于数值特征的 Keras 预处理层：Discretization层。

Discretization 层

Discretization层的目标是通过将值范围（称为箱）映射到类别，将数值特征转换为分类特征。这对于具有多峰分布的特征或与目标具有高度非线性关系的特征有时是有用的。例如，以下代码将数值age特征映射到三个类别，小于 18 岁，18 到 50 岁（不包括），50 岁或以上：

>>> age = tf.constant([[10.], [93.], [57.], [18.], [37.], [5.]])
>>> discretize_layer = tf.keras.layers.Discretization(bin_boundaries=[18., 50.])
>>> age_categories = discretize_layer(age)
>>> age_categories
<tf.Tensor: shape=(6, 1), dtype=int64, numpy=array([[0],[2],[2],[1],[1],[0]])>

在这个例子中，我们提供了期望的分箱边界。如果你愿意，你可以提供你想要的箱数，然后调用层的adapt()方法，让它根据值的百分位数找到合适的箱边界。例如，如果我们设置num_bins=3，那么箱边界将位于第 33 和第 66 百分位数之下的值（在这个例子中，值为 10 和 37）：

>>> discretize_layer = tf.keras.layers.Discretization(num_bins=3)
>>> discretize_layer.adapt(age)
>>> age_categories = discretize_layer(age)
>>> age_categories
<tf.Tensor: shape=(6, 1), dtype=int64, numpy=array([[1],[2],[2],[1],[2],[0]])>

通常不应将诸如此类的类别标识符直接传递给神经网络，因为它们的值无法有意义地进行比较。相反，它们应该被编码，例如使用独热编码。现在让我们看看如何做到这一点。

CategoryEncoding 层

当只有少量类别（例如，少于十几个或二十个）时，独热编码通常是一个不错的选择（如第二章中讨论的）。为此，Keras 提供了CategoryEncoding层。例如，让我们对刚刚创建的age_categories特征进行独热编码：

>>> onehot_layer = tf.keras.layers.CategoryEncoding(num_tokens=3)
>>> onehot_layer(age_categories)
<tf.Tensor: shape=(6, 3), dtype=float32, numpy=
array([[0., 1., 0.],
 [0., 0., 1.],
 [0., 0., 1.],
 [0., 1., 0.],
 [0., 0., 1.],
 [1., 0., 0.]], dtype=float32)>

如果尝试一次对多个分类特征进行编码（只有当它们都使用相同的类别时才有意义），CategoryEncoding类将默认执行多热编码：输出张量将包含每个输入特征中存在的每个类别的 1。例如：

>>> two_age_categories = np.array([[1, 0], [2, 2], [2, 0]])
>>> onehot_layer(two_age_categories)
<tf.Tensor: shape=(3, 3), dtype=float32, numpy=
array([[1., 1., 0.],
 [0., 0., 1.],
 [1., 0., 1.]], dtype=float32)>

如果您认为知道每个类别出现的次数是有用的，可以在创建CategoryEncoding层时设置output_mode="count"，在这种情况下，输出张量将包含每个类别的出现次数。在前面的示例中，输出将与之前相同，只是第二行将变为[0., 0., 2.]。

请注意，多热编码和计数编码都会丢失信息，因为无法知道每个活动类别来自哪个特征。例如，[0, 1]和[1, 0]都被编码为[1., 1., 0.]。如果要避免这种情况，那么您需要分别对每个特征进行独热编码，然后连接输出。这样，[0, 1]将被编码为[1., 0., 0., 0., 1., 0.]，[1, 0]将被编码为[0., 1., 0., 1., 0., 0.]。您可以通过调整类别标识符来获得相同的结果，以便它们不重叠。例如：

>>> onehot_layer = tf.keras.layers.CategoryEncoding(num_tokens=3 + 3)
>>> onehot_layer(two_age_categories + [0, 3])  # adds 3 to the second feature
<tf.Tensor: shape=(3, 6), dtype=float32, numpy=
array([[0., 1., 0., 1., 0., 0.],
 [0., 0., 1., 0., 0., 1.],
 [0., 0., 1., 1., 0., 0.]], dtype=float32)>

在此输出中，前三列对应于第一个特征，最后三列对应于第二个特征。这使模型能够区分这两个特征。但是，这也增加了馈送到模型的特征数量，因此需要更多的模型参数。很难事先知道单个多热编码还是每个特征的独热编码哪个效果最好：这取决于任务，您可能需要测试两种选项。

现在您可以使用独热编码或多热编码对分类整数特征进行编码。但是对于分类文本特征呢？为此，您可以使用StringLookup层。

StringLookup 层

让我们使用 Keras 的StringLookup层对cities特征进行独热编码：

>>> cities = ["Auckland", "Paris", "Paris", "San Francisco"]
>>> str_lookup_layer = tf.keras.layers.StringLookup()
>>> str_lookup_layer.adapt(cities)
>>> str_lookup_layer([["Paris"], ["Auckland"], ["Auckland"], ["Montreal"]])
<tf.Tensor: shape=(4, 1), dtype=int64, numpy=array([[1], [3], [3], [0]])>

我们首先创建一个StringLookup层，然后将其适应到数据：它发现有三个不同的类别。然后我们使用该层对一些城市进行编码。默认情况下，它们被编码为整数。未知类别被映射为 0，就像在这个例子中的“Montreal”一样。已知类别从最常见的类别开始编号，从最常见到最不常见。

方便的是，当创建StringLookup层时设置output_mode="one_hot"，它将为每个类别输出一个独热向量，而不是一个整数：

>>> str_lookup_layer = tf.keras.layers.StringLookup(output_mode="one_hot")
>>> str_lookup_layer.adapt(cities)
>>> str_lookup_layer([["Paris"], ["Auckland"], ["Auckland"], ["Montreal"]])
<tf.Tensor: shape=(4, 4), dtype=float32, numpy=
array([[0., 1., 0., 0.],
 [0., 0., 0., 1.],
 [0., 0., 0., 1.],
 [1., 0., 0., 0.]], dtype=float32)>

提示

Keras 还包括一个IntegerLookup层，其功能类似于StringLookup层，但输入为整数，而不是字符串。

如果训练集非常大，可能会方便地将层适应于训练集的随机子集。在这种情况下，层的adapt()方法可能会错过一些较少见的类别。默认情况下，它会将它们全部映射到类别 0，使它们在模型中无法区分。为了减少这种风险（同时仅在训练集的子集上调整层），您可以将num_oov_indices设置为大于 1 的整数。这是要使用的未知词汇（OOV）桶的数量：每个未知类别将使用哈希函数对 OOV 桶的数量取模，伪随机地映射到其中一个 OOV 桶。这将使模型能够区分至少一些罕见的类别。例如：

>>> str_lookup_layer = tf.keras.layers.StringLookup(num_oov_indices=5)
>>> str_lookup_layer.adapt(cities)
>>> str_lookup_layer([["Paris"], ["Auckland"], ["Foo"], ["Bar"], ["Baz"]])
<tf.Tensor: shape=(4, 1), dtype=int64, numpy=array([[5], [7], [4], [3], [4]])>

由于有五个 OOV 桶，第一个已知类别的 ID 现在是 5（“巴黎”）。但是，"Foo"、"Bar"和"Baz"是未知的，因此它们各自被映射到 OOV 桶中的一个。 "Bar"有自己的专用桶（ID 为 3），但不幸的是，"Foo"和"Baz"被映射到相同的桶中（ID 为 4），因此它们在模型中保持不可区分。这被称为哈希碰撞。减少碰撞风险的唯一方法是增加 OOV 桶的数量。但是，这也会增加总类别数，这将需要更多的 RAM 和额外的模型参数，一旦类别被独热编码。因此，不要将该数字增加得太多。

将类别伪随机映射到桶中的这种想法称为哈希技巧。Keras 提供了一个专用的层，就是Hashing层。

哈希层

对于每个类别，Keras 的Hashing层计算一个哈希值，取模于桶（或“bin”）的数量。映射完全是伪随机的，但在运行和平台之间是稳定的（即，只要桶的数量不变，相同的类别将始终被映射到相同的整数）。例如，让我们使用Hashing层来编码一些城市：

>>> hashing_layer = tf.keras.layers.Hashing(num_bins=10)
>>> hashing_layer([["Paris"], ["Tokyo"], ["Auckland"], ["Montreal"]])
<tf.Tensor: shape=(4, 1), dtype=int64, numpy=array([[0], [1], [9], [1]])>

这个层的好处是它根本不需要适应，这有时可能很有用，特别是在核外设置中（当数据集太大而无法放入内存时）。然而，我们再次遇到了哈希碰撞：“东京”和“蒙特利尔”被映射到相同的 ID，使它们在模型中无法区分。因此，通常最好坚持使用StringLookup层。

现在让我们看另一种编码类别的方法：可训练的嵌入。

使用嵌入编码分类特征

嵌入是一种高维数据（例如类别或词汇中的单词）的密集表示。如果有 50,000 个可能的类别，那么独热编码将产生一个 50,000 维的稀疏向量（即，大部分为零）。相比之下，嵌入将是一个相对较小的密集向量；例如，只有 100 个维度。

在深度学习中，嵌入通常是随机初始化的，然后通过梯度下降与其他模型参数一起训练。例如，在加利福尼亚住房数据集中，"NEAR BAY"类别最初可以由一个随机向量表示，例如[0.131, 0.890]，而"NEAR OCEAN"类别可能由另一个随机向量表示，例如[0.631, 0.791]。在这个例子中，我们使用了 2D 嵌入，但维度的数量是一个可以调整的超参数。

由于这些嵌入是可训练的，它们在训练过程中会逐渐改进；由于它们在这种情况下代表的是相当相似的类别，梯度下降肯定会使它们彼此更接近，同时也会使它们远离"INLAND"类别的嵌入（参见图 13-6）。实际上，表示得越好，神经网络就越容易做出准确的预测，因此训练倾向于使嵌入成为类别的有用表示。这被称为表示学习（您将在第十七章中看到其他类型的表示学习）。

图 13-6。嵌入将在训练过程中逐渐改进

Keras 提供了一个Embedding层，它包装了一个嵌入矩阵：这个矩阵每行对应一个类别，每列对应一个嵌入维度。默认情况下，它是随机初始化的。要将类别 ID 转换为嵌入，Embedding层只需查找并返回对应于该类别的行。就是这样！例如，让我们用五行和 2D 嵌入初始化一个Embedding层，并用它来编码一些类别：

>>> tf.random.set_seed(42)
>>> embedding_layer = tf.keras.layers.Embedding(input_dim=5, output_dim=2)
>>> embedding_layer(np.array([2, 4, 2]))
<tf.Tensor: shape=(3, 2), dtype=float32, numpy=
array([[-0.04663396,  0.01846724],
 [-0.02736737, -0.02768031],
 [-0.04663396,  0.01846724]], dtype=float32)>

正如您所看到的，类别 2 被编码（两次）为 2D 向量[-0.04663396, 0.01846724]，而类别 4 被编码为[-0.02736737, -0.02768031]。由于该层尚未训练，这些编码只是随机的。

警告

Embedding层是随机初始化的，因此除非使用预训练权重初始化，否则在模型之外作为独立的预处理层使用它是没有意义的。

如果要嵌入一个分类文本属性，您可以简单地将StringLookup层和Embedding层连接起来，就像这样：

>>> tf.random.set_seed(42)
>>> ocean_prox = ["<1H OCEAN", "INLAND", "NEAR OCEAN", "NEAR BAY", "ISLAND"]
>>> str_lookup_layer = tf.keras.layers.StringLookup()
>>> str_lookup_layer.adapt(ocean_prox)
>>> lookup_and_embed = tf.keras.Sequential([
...     str_lookup_layer,
...     tf.keras.layers.Embedding(input_dim=str_lookup_layer.vocabulary_size(),
...                               output_dim=2)
... ])
...
>>> lookup_and_embed(np.array([["<1H OCEAN"], ["ISLAND"], ["<1H OCEAN"]]))
<tf.Tensor: shape=(3, 2), dtype=float32, numpy=
array([[-0.01896119,  0.02223358],
 [ 0.02401174,  0.03724445],
 [-0.01896119,  0.02223358]], dtype=float32)>

请注意，嵌入矩阵中的行数需要等于词汇量的大小：这是总类别数，包括已知类别和 OOV 桶（默认只有一个）。StringLookup类的vocabulary_size()方法方便地返回这个数字。

提示

在这个例子中，我们使用了 2D 嵌入，但一般来说，嵌入通常有 10 到 300 个维度，取决于任务、词汇量和训练集的大小。您将需要调整这个超参数。

将所有内容放在一起，现在我们可以创建一个 Keras 模型，可以处理分类文本特征以及常规数值特征，并为每个类别（以及每个 OOV 桶）学习一个嵌入：

X_train_num, X_train_cat, y_train = [...]  # load the training set
X_valid_num, X_valid_cat, y_valid = [...]  # and the validation set

num_input = tf.keras.layers.Input(shape=[8], name="num")
cat_input = tf.keras.layers.Input(shape=[], dtype=tf.string, name="cat")
cat_embeddings = lookup_and_embed(cat_input)
encoded_inputs = tf.keras.layers.concatenate([num_input, cat_embeddings])
outputs = tf.keras.layers.Dense(1)(encoded_inputs)
model = tf.keras.models.Model(inputs=[num_input, cat_input], outputs=[outputs])
model.compile(loss="mse", optimizer="sgd")
history = model.fit((X_train_num, X_train_cat), y_train, epochs=5,
                    validation_data=((X_valid_num, X_valid_cat), y_valid))

这个模型有两个输入：num_input，每个实例包含八个数值特征，以及cat_input，每个实例包含一个分类文本输入。该模型使用我们之前创建的lookup_and_embed模型来将每个海洋接近类别编码为相应的可训练嵌入。接下来，它使用concatenate()函数将数值输入和嵌入连接起来，生成完整的编码输入，准备输入神经网络。在这一点上，我们可以添加任何类型的神经网络，但为了简单起见，我们只添加一个单一的密集输出层，然后我们创建 KerasModel，使用我们刚刚定义的输入和输出。接下来，我们编译模型并训练它，传递数值和分类输入。

正如您在第十章中看到的，由于Input层的名称是"num"和"cat"，我们也可以将训练数据传递给fit()方法，使用字典而不是元组：{"num": X_train_num, "cat": X_train_cat}。或者，我们可以传递一个包含批次的tf.data.Dataset，每个批次表示为((X_batch_num, X_batch_cat), y_batch)或者({"num": X_batch_num, "cat": X_batch_cat}, y_batch)。当然，验证数据也是一样的。

注意

先进行独热编码，然后通过一个没有激活函数和偏置的Dense层等同于一个Embedding层。然而，Embedding层使用的计算量要少得多，因为它避免了许多零乘法——当嵌入矩阵的大小增长时，性能差异变得明显。Dense层的权重矩阵起到了嵌入矩阵的作用。例如，使用大小为 20 的独热向量和一个具有 10 个单元的Dense层等同于使用一个input_dim=20和output_dim=10的Embedding层。因此，在Embedding层后面的层中使用的嵌入维度不应该超过单元数。

好了，现在您已经学会了如何对分类特征进行编码，是时候将注意力转向文本预处理了。

文本预处理

Keras 为基本文本预处理提供了一个TextVectorization层。与StringLookup层类似，您必须在创建时传递一个词汇表，或者使用adapt()方法从一些训练数据中学习词汇表。让我们看一个例子：

>>> train_data = ["To be", "!(to be)", "That's the question", "Be, be, be."]
>>> text_vec_layer = tf.keras.layers.TextVectorization()
>>> text_vec_layer.adapt(train_data)
>>> text_vec_layer(["Be good!", "Question: be or be?"])
<tf.Tensor: shape=(2, 4), dtype=int64, numpy=
array([[2, 1, 0, 0],
 [6, 2, 1, 2]])>

两个句子“Be good!”和“Question: be or be?”分别被编码为[2, 1, 0, 0]和[6, 2, 1, 2]。词汇表是从训练数据中的四个句子中学习的：“be” = 2，“to” = 3，等等。为构建词汇表，adapt()方法首先将训练句子转换为小写并去除标点，这就是为什么“Be”、“be”和“be?”都被编码为“be” = 2。接下来，句子被按空格拆分，生成的单词按降序频率排序，产生最终的词汇表。在编码句子时，未知单词被编码为 1。最后，由于第一个句子比第二个句子短，因此用 0 进行了填充。

提示

TextVectorization层有许多选项。例如，您可以通过设置standardize=None来保留大小写和标点，或者您可以将任何标准化函数作为standardize参数传递。您可以通过设置split=None来防止拆分，或者您可以传递自己的拆分函数。您可以设置output_sequence_length参数以确保输出序列都被裁剪或填充到所需的长度，或者您可以设置ragged=True以获得一个不规则张量而不是常规张量。请查看文档以获取更多选项。

单词 ID 必须进行编码，通常使用Embedding层：我们将在第十六章中进行这样做。或者，您可以将TextVectorization层的output_mode参数设置为"multi_hot"或"count"以获得相应的编码。然而，简单地计算单词通常不是理想的：像“to”和“the”这样的单词非常频繁，几乎没有影响，而“basketball”等更稀有的单词则更具信息量。因此，通常最好将output_mode设置为"tf_idf"，它代表词频 × 逆文档频率（TF-IDF）。这类似于计数编码，但在训练数据中频繁出现的单词被降权，反之，稀有单词被升权。例如：

>>> text_vec_layer = tf.keras.layers.TextVectorization(output_mode="tf_idf")
>>> text_vec_layer.adapt(train_data)
>>> text_vec_layer(["Be good!", "Question: be or be?"])
<tf.Tensor: shape=(2, 6), dtype=float32, numpy=
array([[0.96725637, 0.6931472 , 0. , 0. , 0. , 0.        ],
 [0.96725637, 1.3862944 , 0. , 0. , 0. , 1.0986123 ]], dtype=float32)>

TF-IDF 的变体有很多种，但TextVectorization层实现的方式是将每个单词的计数乘以一个权重，该权重等于 log(1 + d / (f + 1))，其中d是训练数据中的句子总数（也称为文档），f表示这些训练句子中包含给定单词的数量。例如，在这种情况下，训练数据中有d = 4 个句子，单词“be”出现在f = 3 个句子中。由于单词“be”在句子“Question: be or be?”中出现了两次，它被编码为 2 × log(1 + 4 / (1 + 3)) ≈ 1.3862944。单词“question”只出现一次，但由于它是一个不太常见的单词，它的编码几乎一样高：1 × log(1 + 4 / (1 + 1)) ≈ 1.0986123。请注意，对于未知单词，使用平均权重。

这种文本编码方法易于使用，并且对于基本的自然语言处理任务可以得到相当不错的结果，但它有几个重要的局限性：它只适用于用空格分隔单词的语言，它不区分同音异义词（例如“to bear”与“teddy bear”），它不提示您的模型单词“evolution”和“evolutionary”之间的关系等。如果使用多热编码、计数或 TF-IDF 编码，则单词的顺序会丢失。那么还有哪些其他选项呢？

一种选择是使用TensorFlow Text 库，它提供比TextVectorization层更高级的文本预处理功能。例如，它包括几种子词标记器，能够将文本分割成比单词更小的标记，这使得模型更容易检测到“evolution”和“evolutionary”之间有一些共同之处（有关子词标记化的更多信息，请参阅第十六章）。

另一个选择是使用预训练的语言模型组件。现在让我们来看看这个。

使用预训练语言模型组件

TensorFlow Hub 库使得在您自己的模型中重用预训练模型组件变得容易，用于文本、图像、音频等。这些模型组件称为模块。只需浏览TF Hub 存储库，找到您需要的模块，将代码示例复制到您的项目中，模块将自动下载并捆绑到一个 Keras 层中，您可以直接包含在您的模型中。模块通常包含预处理代码和预训练权重，并且通常不需要额外的训练（但当然，您的模型的其余部分肯定需要训练）。

例如，一些强大的预训练语言模型是可用的。最强大的模型非常庞大（几个千兆字节），因此为了快速示例，让我们使用nnlm-en-dim50模块，版本 2，这是一个相当基本的模块，它将原始文本作为输入并输出 50 维句子嵌入。我们将导入 TensorFlow Hub 并使用它来加载模块，然后使用该模块将两个句子编码为向量：

>>> import tensorflow_hub as hub
>>> hub_layer = hub.KerasLayer("https://tfhub.dev/google/nnlm-en-dim50/2")
>>> sentence_embeddings = hub_layer(tf.constant(["To be", "Not to be"]))
>>> sentence_embeddings.numpy().round(2)
array([[-0.25,  0.28,  0.01,  0.1 ,  [...] ,  0.05,  0.31],
 [-0.2 ,  0.2 , -0.08,  0.02,  [...] , -0.04,  0.15]], dtype=float32)

hub.KerasLayer层从给定的 URL 下载模块。这个特定的模块是一个句子编码器：它将字符串作为输入，并将每个字符串编码为单个向量（在本例中是一个 50 维向量）。在内部，它解析字符串（在空格上拆分单词）并使用在一个巨大的语料库上预训练的嵌入矩阵嵌入每个单词：Google News 7B 语料库（七十亿字长！）。然后计算所有单词嵌入的平均值，结果就是句子嵌入。

您只需要在您的模型中包含这个hub_layer，然后就可以开始了。请注意，这个特定的语言模型是在英语上训练的，但许多其他语言也可用，以及多语言模型。

最后，由 Hugging Face 提供的优秀开源Transformers 库也使得在您自己的模型中包含强大的语言模型组件变得容易。您可以浏览Hugging Face Hub，选择您想要的模型，并使用提供的代码示例开始。它以前只包含语言模型，但现在已扩展到包括图像模型等。

我们将在第十六章中更深入地讨论自然语言处理。现在让我们看一下 Keras 的图像预处理层。

图像预处理层

Keras 预处理 API 包括三个图像预处理层：

• tf.keras.layers.Resizing将输入图像调整为所需大小。例如，Resizing(height=100, width=200)将每个图像调整为 100×200，可能会扭曲图像。如果设置crop_to_aspect_ratio=True，则图像将被裁剪到目标图像比例，以避免扭曲。

• tf.keras.layers.Rescaling重新缩放像素值。例如，Rescaling(scale=2/255, offset=-1)将值从 0 → 255 缩放到-1 → 1。

• tf.keras.layers.CenterCrop裁剪图像，保留所需高度和宽度的中心区域。

例如，让我们加载一些示例图像并对它们进行中心裁剪。为此，我们将使用 Scikit-Learn 的load_sample_images()函数；这将加载两个彩色图像，一个是中国寺庙的图像，另一个是花朵的图像（这需要 Pillow 库，如果您正在使用 Colab 或者按照安装说明进行操作，应该已经安装）：

from sklearn.datasets import load_sample_images

images = load_sample_images()["images"]
crop_image_layer = tf.keras.layers.CenterCrop(height=100, width=100)
cropped_images = crop_image_layer(images)

Keras 还包括几个用于数据增强的层，如RandomCrop、RandomFlip、RandomTranslation、RandomRotation、RandomZoom、RandomHeight、RandomWidth和RandomContrast。这些层仅在训练期间激活，并随机对输入图像应用一些转换（它们的名称是不言自明的）。数据增强将人为增加训练集的大小，通常会导致性能提升，只要转换后的图像看起来像真实的（非增强的）图像。我们将在下一章更详细地介绍图像处理。

注意

在幕后，Keras 预处理层基于 TensorFlow 的低级 API。例如，Normalization层使用tf.nn.moments()来计算均值和方差，Discretization层使用tf.raw_ops.Bucketize()，CategoricalEncoding使用tf.math.bincount()，IntegerLookup和StringLookup使用tf.lookup包，Hashing和TextVectorization使用tf.strings包中的几个操作，Embedding使用tf.nn.embedding_lookup()，图像预处理层使用tf.image包中的操作。如果 Keras 预处理 API 不满足您的需求，您可能偶尔需要直接使用 TensorFlow 的低级 API。

现在让我们看看在 TensorFlow 中另一种轻松高效地加载数据的方法。

TensorFlow 数据集项目

TensorFlow 数据集（TFDS）项目使加载常见数据集变得非常容易，从小型数据集如 MNIST 或 Fashion MNIST 到像 ImageNet 这样的大型数据集（您将需要相当大的磁盘空间！）。列表包括图像数据集、文本数据集（包括翻译数据集）、音频和视频数据集、时间序列等等。您可以访问https://homl.info/tfds查看完整列表，以及每个数据集的描述。您还可以查看了解您的数据，这是一个用于探索和理解 TFDS 提供的许多数据集的工具。

TFDS 并未与 TensorFlow 捆绑在一起，但如果您在 Colab 上运行或者按照https://homl.info/install的安装说明进行安装，那么它已经安装好了。然后您可以导入tensorflow_datasets，通常为tfds，然后调用tfds.load()函数，它将下载您想要的数据（除非之前已经下载过），并将数据作为数据集字典返回（通常一个用于训练，一个用于测试，但这取决于您选择的数据集）。例如，让我们下载 MNIST：

import tensorflow_datasets as tfds

datasets = tfds.load(name="mnist")
mnist_train, mnist_test = datasets["train"], datasets["test"]

然后您可以应用任何您想要的转换（通常是洗牌、批处理和预取），然后准备训练您的模型。这里是一个简单的示例：

for batch in mnist_train.shuffle(10_000, seed=42).batch(32).prefetch(1):
    images = batch["image"]
    labels = batch["label"]
    # [...] do something with the images and labels

提示

load()函数可以对其下载的文件进行洗牌：只需设置shuffle_files=True。但是这可能不够，最好对训练数据进行更多的洗牌。

请注意，数据集中的每个项目都是一个包含特征和标签的字典。但是 Keras 期望每个项目是一个包含两个元素的元组（再次，特征和标签）。您可以使用map()方法转换数据集，就像这样：

mnist_train = mnist_train.shuffle(buffer_size=10_000, seed=42).batch(32)
mnist_train = mnist_train.map(lambda items: (items["image"], items["label"]))
mnist_train = mnist_train.prefetch(1)

但是通过设置as_supervised=True，让load()函数为您执行此操作会更简单（显然，这仅适用于带标签的数据集）。

最后，TFDS 提供了一种方便的方法来使用split参数拆分数据。例如，如果您想要使用训练集的前 90%进行训练，剩余的 10%进行验证，整个测试集进行测试，那么您可以设置split=["train[:90%]", "train[90%:]", "test"]。load()函数将返回所有三个集合。这里是一个完整的示例，使用 TFDS 加载和拆分 MNIST 数据集，然后使用这些集合来训练和评估一个简单的 Keras 模型：

train_set, valid_set, test_set = tfds.load(
    name="mnist",
    split=["train[:90%]", "train[90%:]", "test"],
    as_supervised=True
)
train_set = train_set.shuffle(buffer_size=10_000, seed=42).batch(32).prefetch(1)
valid_set = valid_set.batch(32).cache()
test_set = test_set.batch(32).cache()
tf.random.set_seed(42)
model = tf.keras.Sequential([
    tf.keras.layers.Flatten(input_shape=(28, 28)),
    tf.keras.layers.Dense(10, activation="softmax")
])
model.compile(loss="sparse_categorical_crossentropy", optimizer="nadam",
              metrics=["accuracy"])
history = model.fit(train_set, validation_data=valid_set, epochs=5)
test_loss, test_accuracy = model.evaluate(test_set)

恭喜，您已经到达了这个相当技术性的章节的结尾！您可能会觉得它与神经网络的抽象美有些远，但事实是深度学习通常涉及大量数据，知道如何高效加载、解析和预处理数据是一项至关重要的技能。在下一章中，我们将看一下卷积神经网络，这是图像处理和许多其他应用中最成功的神经网络架构之一。

练习

1. 为什么要使用 tf.data API？

2. 将大型数据集拆分为多个文件的好处是什么？

3. 在训练过程中，如何判断您的输入管道是瓶颈？您可以做些什么来解决它？

4. 您可以将任何二进制数据保存到 TFRecord 文件中吗，还是只能序列化协议缓冲区？

5. 为什么要费心将所有数据转换为Example协议缓冲区格式？为什么不使用自己的协议缓冲区定义？

6. 在使用 TFRecords 时，何时应该激活压缩？为什么不系统地这样做？

7. 数据可以在编写数据文件时直接进行预处理，或者在 tf.data 管道中进行，或者在模型内的预处理层中进行。您能列出每个选项的一些优缺点吗？

8. 列举一些常见的编码分类整数特征的方法。文本呢？

9. 加载时尚 MNIST 数据集（在第十章中介绍）；将其分为训练集、验证集和测试集；对训练集进行洗牌；并将每个数据集保存到多个 TFRecord 文件中。每个记录应该是一个序列化的Example协议缓冲区，具有两个特征：序列化图像（使用tf.io.serialize_tensor()来序列化每个图像），和标签。然后使用 tf.data 为每个集创建一个高效的数据集。最后，使用 Keras 模型来训练这些数据集，包括一个预处理层来标准化每个输入特征。尝试使输入管道尽可能高效，使用 TensorBoard 来可视化分析数据。

10. 在这个练习中，您将下载一个数据集，将其拆分，创建一个tf.data.Dataset来高效加载和预处理数据，然后构建和训练一个包含Embedding层的二元分类模型：

    1. 下载大型电影评论数据集，其中包含来自互联网电影数据库（IMDb）的 50,000 条电影评论。数据组织在两个目录中，train和test，每个目录包含一个pos子目录，其中包含 12,500 条正面评论，以及一个neg子目录，其中包含 12,500 条负面评论。每个评论存储在单独的文本文件中。还有其他文件和文件夹（包括预处理的词袋版本），但在这个练习中我们将忽略它们。

    2. 将测试集分为验证集（15,000）和测试集（10,000）。

    3. 使用 tf.data 为每个集创建一个高效的数据集。

    4. 创建一个二元分类模型，使用TextVectorization层来预处理每个评论。

    5. 添加一个Embedding层，并计算每个评论的平均嵌入，乘以单词数量的平方根（参见第十六章）。然后将这个重新缩放的平均嵌入传递给您模型的其余部分。

    6. 训练模型并查看您获得的准确性。尝试优化您的管道，使训练尽可能快。

    7. 使用 TFDS 更轻松地加载相同的数据集：tfds.load("imdb_reviews")。

这些练习的解决方案可以在本章笔记本的末尾找到，网址为https://homl.info/colab3。

¹ 想象一副排好序的扑克牌在您的左边：假设您只拿出前三张牌并洗牌，然后随机选取一张放在右边，将另外两张留在手中。再从左边拿一张牌，在手中的三张牌中洗牌，随机选取一张放在右边。当您像这样处理完所有的牌后，您的右边将有一副扑克牌：您认为它会被完美洗牌吗？

² 一般来说，只预取一个批次就可以了，但在某些情况下，您可能需要预取更多。或者，您可以通过将tf.data.AUTOTUNE传递给prefetch()，让 TensorFlow 自动决定。

³ 但是请查看实验性的tf.data.experimental.prefetch_to_device()函数，它可以直接将数据预取到 GPU。任何带有experimental的 TensorFlow 函数或类的名称可能会在未来版本中发生更改而没有警告。如果实验性函数失败，请尝试删除experimental一词：它可能已经移至核心 API。如果没有，请查看笔记本，我会确保其中包含最新的代码。

⁴ 由于 protobuf 对象旨在被序列化和传输，它们被称为消息。

⁵ 本章包含了您使用 TFRecords 所需了解的最基本知识。要了解更多关于 protobufs 的信息，请访问https://homl.info/protobuf。

⁶ Tomáš Mikolov 等人，“单词和短语的分布式表示及其组合性”，第 26 届国际神经信息处理系统会议论文集 2（2013）：3111–3119。

⁷ Malvina Nissim 等人，“公平比耸人听闻更好：男人对医生，女人对医生”，arXiv 预印本 arXiv:1905.09866（2019）。

⁸ TensorFlow Hub 没有与 TensorFlow 捆绑在一起，但如果您在 Colab 上运行或者按照https://homl.info/install的安装说明进行安装，那么它已经安装好了。

⁹ 要精确，句子嵌入等于句子中单词嵌入的平均值乘以句子中单词数的平方根。这是为了弥补随着n增长，n个随机向量的平均值会变短的事实。

¹⁰ 对于大图像，您可以使用tf.io.encode_jpeg()。这将节省大量空间，但会损失一些图像质量。