AI 工程基础体系 · 第 68/100 篇。内容覆盖机器学习、深度学习与生成式 AI；模型、数据、评测、权限和成本会作为同一生产系统处理。

TensorFlow 工程基础：Tensor、GradientTape、tf.data、训练与导出

TensorFlow 中一次完整的模型执行，可以抽象为：

$$\text{原始数据} \rightarrow \text{Tensor} \rightarrow \text{模型前向计算} \rightarrow \text{损失} \rightarrow \text{梯度} \rightarrow \text{参数更新} \rightarrow \text{评测与导出}$$

其中，Tensor 表示数据，GradientTape 记录可微计算并计算梯度，tf.data 负责把数据组织成训练可消费的批次，训练循环负责更新模型状态，导出则把训练时的 Python 对象转换为可部署的计算接口。

这几个组件不是相互独立的 API。它们共同决定了数据形状、数据类型、自动微分、并发读取、训练状态和部署边界。

---

一、运行环境与最小验证

本文示例以 TensorFlow 2.x 的 eager execution、tf.keras 和 tf.data 为基础。安装 CPU 版本可以使用：

python -m pip install "tensorflow>=2.15"

如果使用 GPU，TensorFlow 对 CUDA、cuDNN、驱动和操作系统有版本匹配要求，不能仅通过安装 Python 包保证 GPU 可用。先验证 TensorFlow 是否能运行：

import tensorflow as tf

print(tf.__version__)
print(tf.config.list_physical_devices("GPU"))

x = tf.constant([[1.0, 2.0], [3.0, 4.0]])
print(x)
print(x.dtype)
print(x.shape)

典型输出类似：

2.x.x
[]
tf.Tensor(
[[1. 2.]
 [3. 4.]], shape=(2, 2), dtype=float32)
<dtype: 'float32'>
(2, 2)

[] 表示当前进程没有发现 GPU，不一定表示机器没有 GPU，也可能是驱动、安装包或容器权限配置不正确。CPU 下仍然可以运行本文示例。

---

二、Tensor：带有形状和类型的数据对象

2.1 Tensor 的基本属性

TensorFlow 的 Tensor 是一个多维、同质类型的数据对象。每个 Tensor 至少包含：

• dtype：元素类型，例如 tf.float32、tf.int32；
• shape：每个维度的大小；
• rank：维度个数；
• 所在设备：CPU、GPU 或其他 TensorFlow 设备；
• 实际数据值。

例如：

import tensorflow as tf

x = tf.constant(
    [[1.0, 2.0, 3.0],
     [4.0, 5.0, 6.0]],
    dtype=tf.float32,
)

print(x.shape)  # (2, 3)
print(tf.rank(x).numpy())  # 2
print(x.dtype)  # <dtype: 'float32'>

这里 x 是一个形状为 (2, 3) 的二维 Tensor。第一个维度通常可以解释为样本数，第二个维度可以解释为特征数，但 TensorFlow 本身并不知道这些语义。

“形状为 (batch_size, feature_dim)”是模型和数据约定的结果，而不是 Tensor 的内建含义。

2.2 Tensor 与 Variable 的区别

普通 Tensor 通常被当作不可变值使用：

x = tf.constant([1.0, 2.0])
# x[0] = 3.0  # 不支持这种原地赋值

模型参数需要被更新，因此使用 tf.Variable：

w = tf.Variable([1.0, 2.0], dtype=tf.float32)
w.assign([3.0, 4.0])
print(w.numpy())  # [3. 4.]

tf.keras.layers.Dense 内部创建的权重就是 tf.Variable。优化器不会修改普通输入 Tensor，而是根据梯度更新这些可训练变量。

一个训练步骤通常具有如下状态：

$$\theta_{t+1} = \operatorname{OptimizerUpdate} (\theta_t,\nabla_\theta L,\text{optimizer state})$$

其中：

• $\theta_t$ 是当前模型参数；
• $L$ 是损失函数；
• $\nabla_\theta L$ 是损失对参数的梯度；
• optimizer state 可能包括 Adam 的一阶、二阶动量等变量。

因此，恢复训练不仅要恢复模型权重，还通常要恢复优化器状态和当前训练进度。

2.3 Shape、广播和数据类型

Tensor 运算遵循操作本身规定的形状规则，并在很多场景下使用 NumPy 风格广播：

x = tf.constant([[1.0, 2.0], [3.0, 4.0]])
bias = tf.constant([10.0, 20.0])

print(x + bias)

结果是：

[[11. 22.]
 [13. 24.]]

bias 的形状为 (2,)，被广播到每一行。这个规则对于全连接层、卷积层和批量计算很重要，但也容易掩盖错误。例如标签本来应为 (batch_size,)，却变成 (batch_size, 1) 或 (batch_size, batch_size) 时，某些损失计算可能发生非预期广播，而不是立即报错。

训练前应明确检查：

tf.debugging.assert_rank(x, 2)
tf.debugging.assert_type(x, tf.float32)
tf.debugging.assert_equal(tf.shape(x)[1], 2)

整数输入也需要特别注意。神经网络中的权重通常是浮点类型，数据应显式转换：

x = tf.cast(x, tf.float32)
y = tf.cast(y, tf.float32)

如果输入是 float64 而模型权重是 float32，TensorFlow 可能触发类型转换、性能下降，或者在某些操作中直接报类型不匹配错误。

---

三、从 Tensor 到模型：前向计算的数学含义

以二分类模型为例，设输入为 $x$，参数为权重 $w$ 和偏置 $b$：

$$z = xw+b$$

再通过 Sigmoid 得到正类概率：

$$p = \sigma(z)=\frac{1}{1+e^{-z}}$$

标签 $y$ 取 0 或 1，二元交叉熵为：

$$L(y,p) = -\left[ y\log(p)+(1-y)\log(1-p) \right]$$

对于一批样本，通常对样本损失取平均：

$$\bar L = \frac{1}{B}\sum_{i=1}^{B}L(y_i,p_i)$$

其中 $B$ 是 batch size。这个“平均”非常重要：如果改成求和，梯度大小会随着 batch size 增大，学习率的实际效果也会改变。

对应的 TensorFlow 模型可以写成：

import tensorflow as tf

model = tf.keras.Sequential([
    tf.keras.layers.Input(shape=(2,)),
    tf.keras.layers.Dense(16, activation="relu"),
    tf.keras.layers.Dense(1, activation="sigmoid"),
])

Input(shape=(2,)) 表示每个样本有 2 个特征，不包括批次维度。因此模型实际接收的输入形状是：

(batch_size, 2)

模型输出形状是：

(batch_size, 1)

如果标签使用形状 (batch_size,)，就需要在损失计算前统一形状：

y = tf.reshape(y, (-1, 1))

或者从数据源开始就让标签保持 (batch_size, 1)。

---

四、GradientTape：记录计算并求导

4.1 自动微分的工作方式

tf.GradientTape 会在上下文中记录 TensorFlow 运算。离开上下文后，可以根据输出计算其对输入或变量的梯度：

x = tf.Variable(3.0)

with tf.GradientTape() as tape:
    y = x * x + 2.0 * x

dy_dx = tape.gradient(y, x)
print(dy_dx.numpy())  # 8.0

因为：

$$y=x^2+2x$$

所以：

$$\frac{dy}{dx}=2x+2$$

在 $x=3$ 时结果为 8。

默认情况下，GradientTape 会自动跟踪参与计算的、trainable=True 的 tf.Variable。对于普通 Tensor，如果希望计算相对于它的梯度，需要显式调用：

x = tf.constant(3.0)

with tf.GradientTape() as tape:
    tape.watch(x)
    y = x * x

print(tape.gradient(y, x).numpy())  # 6.0

4.2 完整推导一个训练更新

考虑单个样本：

$$x=1,\quad y=1,\quad w=0,\quad b=0$$

前向计算为：

$$z=wx+b=0$$

$$p=\sigma(0)=0.5$$

二元交叉熵为：

$$L=-\log(0.5)\approx 0.6931$$

对 Sigmoid 加交叉熵组合，损失对 logits 的导数为：

$$\frac{\partial L}{\partial z}=p-y$$

因此：

$$\frac{\partial L}{\partial z}=0.5-1=-0.5$$

又因为：

$$z=wx+b$$

所以：

$$\frac{\partial z}{\partial w}=x=1,\qquad \frac{\partial z}{\partial b}=1$$

链式法则给出：

$$\frac{\partial L}{\partial w} = \frac{\partial L}{\partial z} \frac{\partial z}{\partial w} = -0.5$$

$$\frac{\partial L}{\partial b}=-0.5$$

使用学习率 $\eta=0.1$ 的梯度下降：

$$w_{\text{new}}=w-\eta\frac{\partial L}{\partial w} =0-0.1(-0.5)=0.05$$

$$b_{\text{new}}=0.05$$

参数沿着降低损失的方向增加，使输入 $x=1$ 更倾向于被预测为正类。

TensorFlow 会自动完成上述链式求导：

w = tf.Variable(0.0)
b = tf.Variable(0.0)
x = tf.constant([[1.0]])
y = tf.constant([[1.0]])

with tf.GradientTape() as tape:
    logits = x * w + b
    loss = tf.reduce_mean(
        tf.nn.sigmoid_cross_entropy_with_logits(
            labels=y,
            logits=logits,
        )
    )

dw, db = tape.gradient(loss, [w, b])

print(loss.numpy())  # 约 0.6931472
print(dw.numpy())    # 约 -0.5
print(db.numpy())    # 约 -0.5

这里使用的是 sigmoid_cross_entropy_with_logits，因此模型输出的是未经过 Sigmoid 的 logits。这种写法通常比先调用 Sigmoid、再手动计算交叉熵具有更好的数值稳定性。

对应关系必须保持一致：

# 输出 logits 时
loss_fn = tf.keras.losses.BinaryCrossentropy(from_logits=True)

# 输出 sigmoid 概率时
loss_fn = tf.keras.losses.BinaryCrossentropy(from_logits=False)

如果模型最后已经有 activation="sigmoid"，却又设置 from_logits=True，损失函数会把概率误当成 logits，训练可能不收敛或收敛异常。

4.3 GradientTape 的生命周期

普通 GradientTape 通常只能调用一次 gradient：

x = tf.Variable(2.0)

with tf.GradientTape() as tape:
    y = x ** 3

grad = tape.gradient(y, x)

如果需要对同一段记录同时求多个梯度，应使用 persistent=True，并在使用后释放：

x = tf.Variable(2.0)

with tf.GradientTape(persistent=True) as tape:
    y = x ** 3
    z = x ** 2

dy_dx = tape.gradient(y, x)
dz_dx = tape.gradient(z, x)

del tape

persistent=True 会保留更多计算记录，长时间或大模型中会增加内存压力，不应默认启用。

---

五、tf.data：把数据变成可消费的训练流

5.1 Dataset 的抽象

tf.data.Dataset 是一个元素序列抽象。元素可以是 Tensor、元组、字典或嵌套结构：

dataset = tf.data.Dataset.from_tensor_slices(
    (
        [[1.0, 2.0], [3.0, 4.0]],
        [0.0, 1.0],
    )
)

for features, labels in dataset:
    print(features.shape, labels.shape)

未批处理时，每次迭代产生一个样本：

(2,) ()
(2,) ()

调用 batch(2) 后：

dataset = dataset.batch(2)

for features, labels in dataset:
    print(features.shape, labels.shape)

输出为：

(2, 2) (2,)

Dataset 的操作通常是惰性的。调用 shuffle、map、batch、prefetch 只是构造数据管道，真正迭代时才执行。

5.2 shuffle、batch、repeat 和 prefetch

一个常见训练管道如下：

train_ds = (
    tf.data.Dataset.from_tensor_slices((x_train, y_train))
    .shuffle(buffer_size=len(x_train), seed=42, reshuffle_each_iteration=True)
    .batch(32)
    .prefetch(tf.data.AUTOTUNE)
)

每一步的因果关系是：

1. from_tensor_slices 把第一个维度切成独立样本；
2. shuffle 随机化样本顺序，降低原始排序对梯度的影响；
3. batch(32) 把样本组成小批次；
4. prefetch 在模型计算当前 batch 时准备后续 batch，减少输入等待。

buffer_size 不是“数据集必须有多大”，而是洗牌缓冲区大小。缓冲区小于数据集时，通常是近似洗牌；缓冲区等于数据集时，内存占用更高，但更接近全量洗牌。

repeat 会重复数据集：

dataset = dataset.repeat(3)

如果使用 repeat() 而不指定次数，数据集变成无限序列。此时训练必须显式指定 steps_per_epoch，否则迭代不会自然结束：

train_ds = train_ds.repeat()

model.fit(
    train_ds,
    steps_per_epoch=100,
    epochs=10,
)

验证集通常不需要 shuffle，也不应无意中使用无限 repeat，否则评估可能无法结束。

5.3 map 的性能和正确性边界

def normalize(x, y):
    x = tf.cast(x, tf.float32) / 255.0
    return x, y

train_ds = train_ds.map(
    normalize,
    num_parallel_calls=tf.data.AUTOTUNE,
)

num_parallel_calls 允许并行执行映射函数。映射函数应尽量使用 TensorFlow 运算，而不是每个样本都进入 Python：

# 可以被 TensorFlow 执行
x = tf.cast(x, tf.float32)

# 在 tf.function 或并行 map 中容易造成性能问题
x = x.numpy()

对于大规模数据，from_tensor_slices 会把 Python/NumPy 数据交给数据管道；如果全部数据本身无法放入内存，应使用 TFRecord、文件列表、tf.data.Dataset.list_files 或自定义生成器等流式来源。

---

六、一个可运行的端到端训练示例

下面示例构造一个二维二分类数据集，使用：

• tf.data 组织训练和验证数据；
• GradientTape 实现自定义训练步骤；
• tf.function 将训练步骤转换为图执行；
• tf.keras.metrics 记录指标；
• tf.saved_model.save 导出带签名的 SavedModel。

import os
import tempfile

import numpy as np
import tensorflow as tf


# 1. 固定随机种子，便于复现实验。
# 这不能保证所有硬件和并行算子都完全确定性。
tf.keras.utils.set_random_seed(42)

# 2. 构造简单的二维二分类数据
n_samples = 2000
x = np.random.randn(n_samples, 2).astype("float32")
y = ((x[:, 0] + x[:, 1]) > 0).astype("float32").reshape(-1, 1)

# 3. 划分训练集和验证集
x_train, y_train = x[:1600], y[:1600]
x_valid, y_valid = x[1600:], y[1600:]

# 4. 构造 tf.data 管道
batch_size = 32

train_ds = (
    tf.data.Dataset.from_tensor_slices((x_train, y_train))
    .shuffle(
        buffer_size=len(x_train),
        seed=42,
        reshuffle_each_iteration=True,
    )
    .batch(batch_size)
    .prefetch(tf.data.AUTOTUNE)
)

valid_ds = (
    tf.data.Dataset.from_tensor_slices((x_valid, y_valid))
    .batch(batch_size)
    .prefetch(tf.data.AUTOTUNE)
)

# 5. 构造模型。
# 最后一层输出 logits，不在模型中调用 sigmoid。
model = tf.keras.Sequential([
    tf.keras.layers.Input(shape=(2,), dtype=tf.float32),
    tf.keras.layers.Dense(16, activation="relu"),
    tf.keras.layers.Dense(1),
])

loss_fn = tf.keras.losses.BinaryCrossentropy(from_logits=True)
optimizer = tf.keras.optimizers.Adam(learning_rate=1e-2)


@tf.function
def train_step(features, labels):
    labels = tf.cast(labels, tf.float32)

    with tf.GradientTape() as tape:
        logits = model(features, training=True)
        loss = loss_fn(labels, logits)

    gradients = tape.gradient(loss, model.trainable_variables)

    # 某些变量可能没有参与当前损失计算。
    # 过滤 None，避免 optimizer.apply_gradients 报错。
    gradient_variables = [
        (grad, var)
        for grad, var in zip(gradients, model.trainable_variables)
        if grad is not None
    ]
    optimizer.apply_gradients(gradient_variables)

    predictions = tf.sigmoid(logits)
    accuracy = tf.reduce_mean(
        tf.cast(tf.equal(tf.cast(predictions >= 0.5, tf.float32), labels), tf.float32)
    )
    return loss, accuracy


@tf.function
def valid_step(features, labels):
    labels = tf.cast(labels, tf.float32)
    logits = model(features, training=False)
    loss = loss_fn(labels, logits)

    predictions = tf.sigmoid(logits)
    accuracy = tf.reduce_mean(
        tf.cast(tf.equal(tf.cast(predictions >= 0.5, tf.float32), labels), tf.float32)
    )
    return loss, accuracy


# 6. 训练若干轮
for epoch in range(5):
    train_loss = tf.keras.metrics.Mean()
    train_accuracy = tf.keras.metrics.Mean()

    for features, labels in train_ds:
        loss, accuracy = train_step(features, labels)
        train_loss.update_state(loss)
        train_accuracy.update_state(accuracy)

    valid_loss = tf.keras.metrics.Mean()
    valid_accuracy = tf.keras.metrics.Mean()

    for features, labels in valid_ds:
        loss, accuracy = valid_step(features, labels)
        valid_loss.update_state(loss)
        valid_accuracy.update_state(accuracy)

    print(
        f"epoch={epoch + 1}, "
        f"train_loss={train_loss.result().numpy():.4f}, "
        f"train_accuracy={train_accuracy.result().numpy():.4f}, "
        f"valid_loss={valid_loss.result().numpy():.4f}, "
        f"valid_accuracy={valid_accuracy.result().numpy():.4f}"
    )

# 7. 导出 SavedModel
export_dir = os.path.join(tempfile.gettempdir(), "tf_binary_classifier_saved_model")

# 示例代码为了可重复运行，先删除旧目录。
# 生产环境中不要无条件删除共享目录，应使用版本化路径和权限控制。
if tf.io.gfile.exists(export_dir):
    tf.io.gfile.rmtree(export_dir)


@tf.function(input_signature=[
    tf.TensorSpec(shape=[None, 2], dtype=tf.float32, name="features")
])
def serving_fn(features):
    logits = model(features, training=False)
    probabilities = tf.sigmoid(logits)
    return {
        "probabilities": probabilities,
    }


tf.saved_model.save(
    model,
    export_dir,
    signatures={"serving_default": serving_fn},
)

print("saved to:", export_dir)

# 8. 从 SavedModel 加载签名并推理
loaded = tf.saved_model.load(export_dir)
serving = loaded.signatures["serving_default"]

result = serving(
    features=tf.constant([[2.0, 1.0], [-2.0, -1.0]], dtype=tf.float32)
)

print(result["probabilities"].numpy())

输出中的概率应表现为：第一行更接近 1，第二行更接近 0。由于训练是随机初始化和有限轮次，具体数值不应被当作固定结果。

6.1 训练步骤的状态变化

每次执行 train_step 时，状态转换如下：

flowchart LR
    A[tf.data 产生 batch] --> B[model training=True]
    B --> C[logits]
    C --> D[loss]
    D --> E[GradientTape 求梯度]
    E --> F[optimizer 更新 Variable]
    F --> G[指标累计]
    G --> A

关键点有三个：

1. model(features, training=True) 会让 Dropout、Batch Normalization 等训练相关层使用训练行为；
2. 梯度只在 GradientTape 上下文中记录的计算图上产生；
3. optimizer.apply_gradients 才真正修改模型变量，单独计算梯度不会更新参数。

验证时使用 training=False，否则 Dropout 仍可能随机丢弃单元，Batch Normalization 也可能使用批次统计量，导致评估结果不稳定或与部署行为不一致。

---

七、tf.function：把 Python 函数转换为 TensorFlow 图

tf.function 会根据输入 Tensor 的结构、形状和类型追踪函数，并生成可执行的 TensorFlow 图：

@tf.function
def add_one(x):
    return x + 1

print(add_one(tf.constant([1, 2])))

图执行的优点包括：

• 减少 Python 解释器参与；
• 便于设备调度和图优化；
• 可以被 SavedModel 的签名引用；
• 更适合服务端稳定执行。

但 tf.function 不是普通缓存装饰器。输入形状或 Python 参数变化可能触发重新追踪，即 retracing：

@tf.function
def f(x):
    return x * 2

f(tf.ones([4, 2]))
f(tf.ones([8, 2]))

如果希望允许不同 batch size 而不因第一维变化反复追踪，可以指定输入签名：

@tf.function(input_signature=[
    tf.TensorSpec([None, 2], tf.float32)
])
def f(x):
    return x * 2

None 表示该维度可变；特征维度 2 仍然固定。

常见错误是把 Python 副作用放进图函数：

values = []

@tf.function
def bad_fn(x):
    values.append(x)
    return x + 1

列表追加发生在追踪阶段和图执行阶段的行为并不等价，不能把它当作每次调用都会执行的普通 Python 代码。图内状态应使用 TensorFlow 的变量、指标或返回值表达。

另一个常见错误是在 tf.function 中调用 .numpy()：

@tf.function
def bad_fn(x):
    return x.numpy()  # 通常会失败

图执行时 Tensor 不一定有可直接访问的 NumPy 值。调试时可以暂时移除 @tf.function，或使用 tf.print：

@tf.function
def debug_fn(x):
    tf.print("shape:", tf.shape(x))
    return x + 1

---

八、训练接口：自定义循环与 model.fit 的边界

TensorFlow 提供两种常见训练方式。

8.1 model.fit

如果模型使用标准损失、优化器和指标，compile 与 fit 可以减少训练样板代码：

model = tf.keras.Sequential([
    tf.keras.layers.Input(shape=(2,)),
    tf.keras.layers.Dense(16, activation="relu"),
    tf.keras.layers.Dense(1, activation="sigmoid"),
])

model.compile(
    optimizer=tf.keras.optimizers.Adam(1e-2),
    loss=tf.keras.losses.BinaryCrossentropy(),
    metrics=[tf.keras.metrics.BinaryAccuracy()],
)

model.fit(
    train_ds,
    validation_data=valid_ds,
    epochs=5,
)

这里因为最后一层输出概率，所以使用默认的 from_logits=False。

fit 适合标准监督学习流程，但它并不意味着训练过程没有状态。模型权重、优化器变量、指标状态、回调状态和数据迭代都会影响结果。

8.2 自定义训练循环

需要以下能力时，自定义循环更合适：

• 多个损失，例如生成器和判别器损失；
• 梯度裁剪、梯度累积或特殊更新顺序；
• 强化学习或非标准数据流；
• 自定义混合精度、正则项或约束；
• 需要精确控制每一步的日志和故障处理。

梯度裁剪示例：

gradients = tape.gradient(loss, model.trainable_variables)
gradients, _ = tf.clip_by_global_norm(gradients, 1.0)
optimizer.apply_gradients(zip(gradients, model.trainable_variables))

如果梯度出现 NaN 或 Inf，裁剪只能缓解部分爆炸问题，不能修复错误的标签、错误的 logits 配置、数据溢出或学习率过大。应先检查：

for grad, var in zip(gradients, model.trainable_variables):
    if grad is not None:
        tf.debugging.check_numerics(grad, var.name + "_gradient")

---

九、评测：不要把训练损失当作完整质量

训练损失回答的是“当前优化目标是否下降”，不等于生产质量。

二分类至少应区分：

• loss：用于优化；
• accuracy：阈值为 0.5 时的分类正确率；
• precision：预测为正的样本中有多少真正为正；
• recall：真正为正的样本中有多少被找出；
• AUC：不同阈值下的排序能力。

例如类别极不平衡时，99% 的负类可能让 accuracy 很高，但正类 recall 接近 0。生成式 AI 中还需要根据任务选择人工评测、事实性、拒答率、延迟、token 成本和安全指标，不能只看训练 loss。

验证集必须与训练集在数据生成路径上隔离。若先对全量数据做归一化、词表构建或特征选择，再划分训练集和验证集，统计信息可能泄漏，造成验证指标虚高。

---

十、保存、检查点与导出

10.1 Checkpoint 与部署导出的区别

Checkpoint 主要服务于“继续训练”：

checkpoint = tf.train.Checkpoint(
    model=model,
    optimizer=optimizer,
)
manager = tf.train.CheckpointManager(
    checkpoint,
    directory="./checkpoints",
    max_to_keep=3,
)

manager.save()

它通常保存模型变量和优化器变量，适合训练恢复。

SavedModel 主要服务于“加载并推理”。它可以保存计算图、变量和具体输入输出签名。部署方不应依赖训练脚本中的 Python 类、全局变量或数据加载器。

10.2 输入签名是部署契约

本文导出时使用：

tf.TensorSpec(
    shape=[None, 2],
    dtype=tf.float32,
    name="features",
)

这定义了部署接口：

• 输入名为 features；
• 类型为 float32；
• 第一维是动态 batch；
• 每个样本必须有 2 个特征。

如果服务传入 int64、形状 [2]、特征数 3，加载后的签名调用应报错或无法得到正确结果。签名不是文档装饰，而是模型与服务之间的类型契约。

可以查看导出的签名：

saved_model_cli show \
  --dir /tmp/tf_binary_classifier_saved_model \
  --all

输出会列出 serving_default 以及输入、输出的名称、dtype 和 shape。生产发布前应使用一组固定样例执行加载后的模型，并比较导出前后的输出。

10.3 Keras 3 的导出差异

在较新的 Keras 3 环境中，model.export(path) 是面向 SavedModel 推理导出的明确接口：

model.export("./exported_model")

但不同 TensorFlow/Keras 版本在模型加载方式、原生 Keras 格式和 SavedModel 之间存在差异：

• .keras 文件主要用于保存 Keras 模型配置和权重；
• SavedModel 主要用于 TensorFlow 生态中的图和签名部署；
• tf.saved_model.load 返回的是带签名的对象，不一定是可直接调用 fit 的 Keras 模型；
• 如果需要继续用 Keras 训练，应保存或恢复 Keras 模型及其编译状态，而不是只保存服务签名。

因此，导出格式应由下游运行时决定，而不是仅由训练代码决定。

---

十一、常见失败表现与诊断路径

11.1 No gradients provided

常见原因包括：

1. 在 GradientTape 外计算了 loss；
2. 用 .numpy() 或 NumPy 运算切断了 TensorFlow 计算图；
3. 模型变量没有参与 loss；
4. 参数被错误地设置为 trainable=False；
5. 损失返回 Python 标量，而不是 TensorFlow Tensor。

基本诊断方式：

print(len(model.trainable_variables))

with tf.GradientTape() as tape:
    logits = model(features, training=True)
    loss = loss_fn(labels, logits)

gradients = tape.gradient(loss, model.trainable_variables)

for variable, gradient in zip(model.trainable_variables, gradients):
    print(variable.name, variable.shape, gradient is None)

11.2 loss 为 NaN

应按以下顺序检查：

tf.debugging.check_numerics(features, "features")
tf.debugging.check_numerics(labels, "labels")
tf.debugging.check_numerics(logits, "logits")
tf.debugging.check_numerics(loss, "loss")

重点排查：

• 输入是否包含 NaN 或无穷大；
• 是否对 0 取对数；
• 是否把 logits 和概率混用；
• 学习率是否过大；
• 混合精度下是否发生数值下溢或溢出；
• 标签是否超出损失函数要求的范围。

11.3 训练指标异常稳定或完全不变

可能不是优化器问题，而是数据管道问题：

• repeat 和 steps_per_epoch 配置错误；
• 标签始终为同一个类别；
• shuffle 没有真正作用于样本和标签的对应元组；
• 训练步骤中误用了 training=False；
• 没有执行 apply_gradients；
• 输出层激活函数和损失函数配置冲突。

可以先取一个 batch 做最小闭环验证：

features, labels = next(iter(train_ds))
print(features.shape, features.dtype)
print(labels.shape, labels.dtype)
print(tf.reduce_min(features), tf.reduce_max(features))
print(tf.reduce_min(labels), tf.reduce_max(labels))

然后确认同一个 batch 上连续执行两次训练步骤后，某个权重确实发生变化。

---

十二、生产系统中的数据、权限与成本边界

TensorFlow 代码只是生产系统的一部分。训练数据可能包含个人信息、业务机密或受限内容，数据读取身份应只拥有必要的对象存储、数据库或文件系统权限。导出的 SavedModel、Checkpoint 和日志同样可能包含模型能力、敏感样本痕迹或输入输出信息，不能因为它们是“模型文件”就放入公开目录。

数据管道还会直接影响成本：

• cache() 全量缓存可能消耗大量内存；
• 过大的 prefetch 会增加内存占用；
• CPU 解码或增强跟不上 GPU 时，GPU 会空转；
• shuffle 缓冲区越大，通常需要更多内存；
• Checkpoint 保存过于频繁会增加存储和 I/O；
• 训练失败后若没有保存随机种子、数据版本、代码版本和超参数，恢复成本会显著增加。

生产训练应至少记录：

• 数据集版本和切分方式；
• 模型结构、TensorFlow/Keras 版本；
• 优化器、学习率、batch size 和训练步数；
• 训练与验证指标；
• Checkpoint 路径和校验结果；
• 导出签名及样例输入输出；
• 使用的计算资源和训练时长。

部署前验证不是“文件存在”就结束，而应包含：

1. 从新进程加载导出目录；
2. 用合法输入调用 serving_default；
3. 用非法 dtype、错误 rank 和错误特征维度确认接口能拒绝错误请求；
4. 比较导出前后同一批样例的输出；
5. 在目标服务运行时中验证权限、内存和超时配置。

---

十三、核心边界

Tensor 解决的是数据表示，不负责业务语义；GradientTape 解决的是可微计算，不会自动保证损失定义正确；tf.data 解决的是数据迭代，不会自动防止数据泄漏；训练循环解决参数更新，不会自动提供可靠评测；SavedModel 解决模型计算和接口导出，也不会自动解决访问控制、数据脱敏、容量规划或服务监控。

一个可维护的 TensorFlow 工程，需要同时保持以下契约：

$$\text{数据 dtype/shape} \Rightarrow \text{模型输入契约} \Rightarrow \text{损失定义} \Rightarrow \text{梯度路径} \Rightarrow \text{参数更新} \Rightarrow \text{评测指标} \Rightarrow \text{导出签名}$$

其中任意一环不一致，都可能出现“代码可以运行，但结果不可信”的情况。真正稳定的基础，不是记住某个训练 API，而是能从 Tensor 的形状和类型开始，沿着数据流追踪到梯度、状态、评测和部署接口。

---

系列导航与关联阅读

• 系列入口：AI 工程完整学习路线：从机器学习与 Transformer 到 RAG、Agent 和生产治理
• 上一篇：模型检查点与训练恢复：状态、随机数、分片、容错和一致性
• 下一篇：JAX 工程基础：JIT、Grad、Vmap、Pmap、纯函数和设备执行

官方资料

• PyTorch Documentation
• PyTorch Tutorials

本文依据研究论文、标准组织与主流框架官方文档重新梳理；正文、示例与工程清单由 WR BLOG 编写。