tensorflow-deep-learning

TensorFlow 与深度学习

本技能涵盖使用 TensorFlow 和 Keras 构建神经网络的项目结构、模型开发、训练、评估和部署实践。

训练并交付模型的工作流

1. 按职责拆分模块 — 将数据加载、模型定义、训练、评估和服务代码放在不同文件中；仅将笔记本用于探索。
2. 先构建输入流水线 — 使用 tf.data.Dataset 实现可扩展的预取式输入处理；编写任何模型代码之前，先验证形状、数据类型、标签范围和类别平衡。
3. 拟合前先拆分数据 — 在拟合归一化统计量或增强参数之前，先创建训练集、验证集和测试集拆分，避免数据泄漏。
4. 从小型基线开始 — 构建一个小模型并进行小规模过拟合测试（用少量样本拟合，直到损失降至约为 0），以确认训练循环正常运行，再扩大规模。
5. 使用回调进行训练 — 配置 ModelCheckpoint（根据验证指标保存最佳模型，而非最后一个轮次的模型）、EarlyStopping、学习率计划和 TensorBoard 日志记录。
6. 使用适合任务的指标进行评估 — 仅在最终报告时使用留出测试集，绝不用于调参。
7. 使用显式签名导出 — 使用 model.export() 或 tf.saved_model.save() 保存模型，并显式定义服务输入签名。
8. 对导出结果进行冒烟测试 — 部署前加载导出的模型，并使用样本输入运行推理。

项目结构

• 将数据加载、模型定义、训练、评估和服务代码分别放入不同模块（例如，data.py、model.py、train.py、evaluate.py、serve.py）。
• 将模型超参数保存在带类型的配置对象（数据类）或配置文件中，不要将字面值散布在训练代码中。
• 将检查点、日志和导出的模型存放在源代码目录之外（例如，放在 artifacts/ 或配置的输出路径下），以免意外纳入版本控制。
• 将笔记本用于探索；验证某种方法后，将可重复使用的训练代码移到可作为脚本运行并由测试覆盖的模块中。

project/
  data.py          # tf.data pipeline construction
  model.py         # Keras model definition
  config.py        # typed hyperparameter config
  train.py         # training loop / Keras fit orchestration
  evaluate.py       # metrics computation on held-out data
  serve.py         # export + inference smoke test
  artifacts/       # checkpoints, logs, saved models (gitignored)

输入流水线（tf.data）

import tensorflow as tf

def build_dataset(file_pattern: str, batch_size: int, shuffle: bool) -> tf.data.Dataset:
    files = tf.data.Dataset.list_files(file_pattern, shuffle=shuffle)
    ds = files.interleave(
        tf.data.TFRecordDataset,
        cycle_length=tf.data.AUTOTUNE,
        num_parallel_calls=tf.data.AUTOTUNE,
    )
    ds = ds.map(_parse_example, num_parallel_calls=tf.data.AUTOTUNE)
    if shuffle:
        ds = ds.shuffle(buffer_size=10_000)
    ds = ds.batch(batch_size, drop_remainder=False)
    return ds.prefetch(tf.data.AUTOTUNE)

def _parse_example(record: tf.Tensor) -> tuple[tf.Tensor, tf.Tensor]:
    feature_spec = {
        "features": tf.io.FixedLenFeature([32], tf.float32),
        "label": tf.io.FixedLenFeature([], tf.int64),
    }
    parsed = tf.io.parse_single_example(record, feature_spec)
    return parsed["features"], parsed["label"]

• 在 .map()/.interleave() 上使用 num_parallel_calls=tf.data.AUTOTUNE，并始终以 .prefetch(tf.data.AUTOTUNE) 结束流水线。
• 在批处理前打乱数据，并使用足够大的缓冲区，以便根据数据集大小实现近似全量打乱。
• 开始长时间训练前，先在一个样本批次上验证数据形状、数据类型、标签范围和类别平衡。

模型开发

• 除非有特定需求必须使用较底层的 tf.GradientTape 控制，否则使用 Keras 层和模型（tf.keras.Model、tf.keras.Sequential）。
• 优先使用显式输入形状和命名输入/输出，这会让服务签名和调试更加容易。
• 从小型基线模型开始，并确认它能够对一小部分数据过拟合（损失接近零），然后再增加深度、宽度或数据集规模，以便尽早发现流水线和损失函数中的问题。
• 在需要可复现性时固定随机种子（tf.random.set_seed、numpy.random.seed），但要说明 GPU 执行中的某些操作仍可能存在非确定性（例如，某些 cuDNN 归约操作）。
• 只有在全精度下验证数值稳定性后，才启用混合精度（tf.keras.mixed_precision.set_global_policy("mixed_float16")）；某些损失（例如涉及 log 或 exp 的损失）需要进行损失缩放，以避免下溢。

import tensorflow as tf

def build_model(input_dim: int, num_classes: int) -> tf.keras.Model:
    inputs = tf.keras.Input(shape=(input_dim,), name="features")
    x = tf.keras.layers.Dense(128, activation="relu")(inputs)
    x = tf.keras.layers.Dropout(0.3)(x)
    x = tf.keras.layers.Dense(64, activation="relu")(x)
    outputs = tf.keras.layers.Dense(num_classes, activation="softmax", name="predictions")(x)
    return tf.keras.Model(inputs=inputs, outputs=outputs, name="baseline_classifier")

训练

import tensorflow as tf

model = build_model(input_dim=32, num_classes=10)
model.compile(
    optimizer=tf.keras.optimizers.Adam(learning_rate=1e-3),
    loss="sparse_categorical_crossentropy",
    metrics=["accuracy"],
)

callbacks = [
    tf.keras.callbacks.ModelCheckpoint(
        filepath="artifacts/checkpoints/best.keras",
        monitor="val_accuracy",
        save_best_only=True,
    ),
    tf.keras.callbacks.EarlyStopping(
        monitor="val_loss", patience=5, restore_best_weights=True,
    ),
    tf.keras.callbacks.ReduceLROnPlateau(
        monitor="val_loss", factor=0.5, patience=3, min_lr=1e-6,
    ),
    tf.keras.callbacks.TensorBoard(log_dir="artifacts/logs"),
]

train_ds = build_dataset("data/train-*.tfrecord", batch_size=64, shuffle=True)
val_ds = build_dataset("data/val-*.tfrecord", batch_size=64, shuffle=False)

history = model.fit(
    train_ds,
    validation_data=val_ds,
    epochs=50,
    callbacks=callbacks,
)

• 在拟合任何归一化层或增强参数之前，先将数据拆分为训练集、验证集和测试集——在完整数据集上拟合会将测试集统计信息泄漏到训练中。
• 使用验证集进行调优（架构、学习率、正则化）；单独保留一个测试集，仅用于最终报告，且只使用一次。
• 为每次运行跟踪损失曲线、指标、学习率和资源利用率（TensorBoard 或实验跟踪器）。
• 保存验证指标最佳时的检查点（save_best_only=True），而不只是最后一个 epoch 的权重——最后一个 epoch 不一定表现最佳，尤其是在使用 EarlyStopping 时。

评估

• 报告适合任务的指标：分类任务使用 AUROC/F1/校准指标，语言任务使用困惑度或 BLEU/ROUGE，回归任务使用 MAE/RMSE。
• 对分类任务提供混淆矩阵或按类别划分的错误切片——整体准确率会掩盖类别层面的失败。
• 数据允许时，在边缘案例和已知分布偏移上评估（例如，留出来自不同时间段或来源的数据切片）。
• 当数据集较小或属于表格数据时，与非神经网络基线（逻辑回归、梯度提升树）进行比较——深度模型表现不如简单基线是一个重要信号，而非可有可无的注释。

test_ds = build_dataset("data/test-*.tfrecord", batch_size=64, shuffle=False)
results = model.evaluate(test_ds, return_dict=True)
print(results)  # {'loss': ..., 'accuracy': ...}

部署

• 使用明确的输入签名导出模型，以便服务基础设施（TF Serving、Vertex AI 等）了解预期的输入形状和数据类型：

model.export("artifacts/saved_model/v1")

# Or with an explicit serving signature:
@tf.function(input_signature=[tf.TensorSpec(shape=[None, 32], dtype=tf.float32, name="features")])
def serve_fn(features):
    return {"predictions": model(features, training=False)}

tf.saved_model.save(model, "artifacts/saved_model/v1", signatures={"serving_default": serve_fn})

• 确保训练和服务过程中的预处理保持一致——将归一化/分词逻辑内置到导出的计算图中（例如，将 tf.keras.layers.Normalization 作为模型的一部分），而不是在独立的服务端脚本中重复实现。
• 添加冒烟测试，加载导出的 SavedModel，并使用少量已知样本输入运行推理，在将模型部署到生产环境之前断言输出处于预期范围/形状。
• 部署后监控延迟、内存、预测漂移和输入架构变化——导出时正确的模型，可能会随着上游数据变化而悄然退化。

常见错误

• 在验证标签和数据质量之前就调优架构——数据错误的影响会比任何架构更改持续得久。
• 在完整数据集上而非仅在训练集上拟合预处理或增强步骤，导致验证集或测试集数据泄漏。
• 对不平衡数据集只依赖准确率——应改用精确率/召回率、F1 或 AUROC。
• 部署只有 Notebook、没有可复现训练脚本或固定依赖项的模型。
• 忽略训练和推理之间批量大小、数据类型或设备（CPU/GPU）的差异；这些差异可能会悄然改变数值结果（尤其是在混合精度下）。