第 12 章 — 可复现性保障（set_seed）

第 11 章把代码拆成了多文件工程，跑起来清爽多了。但你大概率已经发现一个尴尬的问题：同一份代码、同一个配置，跑两次结果却不一样——验证集切分、数据洗牌、模型初始化、数据增强，处处都在"随机"。本章把能固定下来的随机源全部固定住，从此同配置 = 同结果：对比实验终于敢下结论，复现论文、排查 bug 也有了底气。

一、本章要解决的问题

• 之前：第 11 章拆完工程后跑起来很清爽，但同一份代码、同一个配置跑两次，random_split 的划分、模型的随机初始化、DataLoader 的洗牌顺序、数据增强的随机裁剪/翻转全在"随机"，两次结果必然不同。
• 现在：我们要封装一个 set_seed 工具，在训练入口、任何随机操作之前调用一次，把 Python random、numpy、torch、CUDA、cuDNN、DataLoader worker 的全部随机源固定下来，让"同配置 = 同结果"。
• 判断标准：连续两次 python train.py --epochs 5，把标准输出分别重定向到 run1.log 和 run2.log，diff run1.log run2.log 没有任何输出。

二、核心概念速览

下面 7 个概念是本章的全部"生词"。先花 5 分钟读完，再看代码会轻松很多。

1. 随机性藏在哪里（随机性的来源）

代码里看起来"不用管"的地方其实到处是随机：random_split 切分训练/验证集、DataLoader 的 shuffle=True 洗牌、模型的权重初始化（nn.Conv2d/nn.Linear 默认随机初始化）、训练集的数据增强（随机裁剪、水平翻转、色彩抖动）。这四类随机源混在一起，任何一个没固定，最终结果都会飘。类比做菜：盐、糖、酱油全凭手感放，就算菜谱写得一模一样，两次做出来的味道也差一截——要把每一处"手感"都换成"量杯"。

2. 随机种子是什么（随机数的"初始状态"）

计算机的随机其实是伪随机：用一个初始数字（种子）喂给一个固定算法，生成一串"看起来随机"的数字。给定同一个种子，生成的那串数字序列永远相同。所以"固定种子"就是给随机数发生器设定同一个起点，之后所有随机操作都从同一个序列里取数。类比一副洗好的牌：种子是"洗牌前的初始顺序"，只要初始顺序一样，后面切牌、发牌的流程相同，结果就相同；不固定种子，等于每次都用一副随机洗过、完全不同的牌。

3. 为什么要把 random / numpy / torch 三套随机源分别固定

Python 生态里至少有三套互不相通的随机发生器：内置 random、numpy.random、torch（CPU 和 GPU 还各是各的）。它们各自维护自己的内部状态，固定其中任何一个，另外几个照样乱跑。所以 set_seed 要挨个调 random.seed(seed)、np.random.seed(seed)、torch.manual_seed(seed)。类比三张牌桌：每桌各洗各的牌，你想让三桌结果都稳定，就得给每桌都指定同一个初始状态——少指定一桌，那一桌照样随机。

4. torch.manual_seed 与 manual_seed_all 的区别

torch.manual_seed(seed) 一次固定 CPU 上的随机发生器，以及当前 GPU 的默认发生器；torch.cuda.manual_seed_all(seed) 则覆盖所有 GPU 的默认发生器（多卡时每张卡都要有种子）。单卡时只调 manual_seed 就够用，但为了以后上多卡（第 13 章）不用回头改代码，我们习惯两个都调——manual_seed_all 只在 torch.cuda.is_available() 为真时才执行。

5. cudnn.deterministic 与 benchmark（确定性 vs 性能的权衡）

即使随机数全固定了，cuDNN 底层实现卷积等算子时，会根据输入形状在多个算法里"挑最快的那个"——这个挑选过程本身有随机性，会导致同一份输入在不同运行里出现微小差异。benchmark=False 关掉"自动选优"，deterministic=True 强制用确定性算法，结果就能逐位复现，代价是部分算子慢一点。类比导航：benchmark 是"每次看实时路况选最快路线"，deterministic 是"规定永远走同一条路"——慢一点，但每一步都确定。

6. DataLoader worker 为什么需要 worker_init_fn

DataLoader(num_workers=2) 会 fork 出子进程来并行预加载数据。子进程从主进程那里继承到的只是"fork 那一刻"的随机状态，之后主进程的随机序列它完全不知道；而且多个子进程如果从同一个继承状态出发，还可能产生互相重复的随机数。worker_init_fn 会在每个子进程启动时被调用，PyTorch 会用主进程种子派生出每个 worker 独有的 torch.initial_seed()，我们再拿它去重置 numpy/random，这样每个 worker 内的随机源既确定、又互不重复。

7. 固定种子的调用时机

set_seed 必须放在任何随机操作之前——random_split 切分、模型初始化、DataLoader 洗牌、数据增强，全都不能先于它。因为"固定种子"只对之后发生的随机操作生效，之前的已经发生了，现在固定也白搭。所以它在 train.py 里紧跟在参数覆盖之后、构建任何组件之前调用。类比发牌前先定洗牌顺序：等牌都发完了再改规则，已经来不及了。

---

三、解决思路

1. 新增 utils/seed.py：封装两个函数——set_seed(seed) 一次固定 random/numpy/torch/CUDA/cuDNN 这五处随机源；worker_init_fn(worker_id) 在每个 DataLoader 子进程启动时重置该进程的 numpy/random。一个文件只干一件事，正是第 11 章说的"单一职责"。
2. 配置与入口：config/config.yaml 顶部新增 seed: 42；train.py 新增 --seed 命令行参数（传了就覆盖配置），并在参数覆盖之后、构建任何组件之前调用 set_seed(cfg.seed)。
3. 数据侧透传：dataset/datasets.py 的 build_dataloaders 增加 worker_init_fn=None 参数，透传给三个 DataLoader；train.py 把 worker_init_fn 传进去。调用方不提需求就不影响原有行为（默认 None）。
4. 依赖清单：requirements.txt 显式加上 numpy>=1.24——set_seed 里要用它，而且明确列出来，别人复现时才不会踩"隐式依赖"的坑。
5. 不做什么：不做多卡训练（第 13 章）、不做实验管理与结果对比表、不锁定依赖的精确版本（重度复现可自己 pip freeze）。本章只解决"单机单卡下逐位可复现"。

trade-off：cudnn.deterministic=True + benchmark=False 会让部分算子变慢（cuDNN 不再按形状自动选最快算法）；数据增强一旦固定，每个 epoch 看到的样本顺序也固定，等于少了一点"免费随机"，正则化效果略降。对 CIFAR-10 这种小任务，性能损失可忽略，完全值得付；对一次训练要跑几周的大模型，通常只在发布前复现关键实验时才开。

四、代码变更

相对第 11 章的改动，一共 1 个新文件、4 个修改文件：

# 新增 utils/seed.py（见下方完整代码）
+ utils/seed.py                 # 新增：set_seed + worker_init_fn

# config/config.yaml：顶部新增一行
+ seed: 42                      # 全局随机种子（命令行 --seed 可覆盖）
  device: auto

# dataset/datasets.py：build_dataloaders 增加可选参数并透传给三个 DataLoader
- def build_dataloaders(cfg):
+ def build_dataloaders(cfg, worker_init_fn=None):
      ...
      return (make_loader(train_dataset, shuffle=True),
              make_loader(val_dataset, shuffle=False),
              make_loader(test_dataset, shuffle=False))

# train.py：新增 --seed 参数、set_seed 调用、worker_init_fn 透传
  parser.add_argument("--resume", type=str, default=None,
                      help="从指定 checkpoint 恢复训练")
+ parser.add_argument("--seed", type=int, default=None,
+                     help="覆盖配置文件中的 seed")
  ...
  if args.batch_size is not None:
      cfg.data.batch_size = args.batch_size
+ if args.seed is not None:
+     cfg.seed = args.seed
+
+ # 关键顺序：set_seed 必须在 random_split、模型初始化、DataLoader 之前
+ set_seed(cfg.seed)
  ...
- train_loader, val_loader, test_loader = build_dataloaders(cfg)
+ train_loader, val_loader, test_loader = build_dataloaders(
+     cfg, worker_init_fn=worker_init_fn)

# requirements.txt：显式声明 numpy
  torch>=2.0
  torchvision>=0.15
+ numpy>=1.24
  pyyaml>=6.0
  ...

五、完整代码

本章完整代码 = 第 11 章的全部文件 + 下面 5 个新增/修改文件。其余文件（models/classifier.py、utils/logger.py、utils/checkpoint.py、utils/metrics.py、utils/early_stopping.py、engine/trainer.py）与第 11 章完全相同，一个字都不用动。目录结构如下，utils/ 下多了 seed.py：

pytorch_learning/
├── train.py
├── requirements.txt
├── config/
│   └── config.yaml              # 修改：新增 seed
├── dataset/
│   └── datasets.py              # 修改：build_dataloaders 支持 worker_init_fn
├── models/
│   └── classifier.py            # 与第 11 章完全相同
├── utils/
│   ├── logger.py                # 与第 11 章完全相同
│   ├── checkpoint.py            # 与第 11 章完全相同
│   ├── metrics.py               # 与第 11 章完全相同
│   ├── early_stopping.py        # 与第 11 章完全相同
│   └── seed.py                  # 本章新增
└── engine/
    └── trainer.py               # 与第 11 章完全相同

utils/seed.py（新增）：

"""随机种子工具：固定所有随机源，保证实验可复现。"""
import random        # Python 内置随机源（random_split 内部靠它打乱）
import numpy as np   # numpy 随机源（数据增强/预处理常用它）
import torch         # torch 随机源（模型初始化、DataLoader 洗牌靠它）


def set_seed(seed):
    """固定 Python/numpy/torch/CUDA 的随机源，须在任何随机操作之前调用。"""
    random.seed(seed)                     # 固定内置 random
    np.random.seed(seed)                  # 固定 numpy
    torch.manual_seed(seed)               # 固定 torch（CPU + 当前 GPU 一起）
    if torch.cuda.is_available():
        torch.cuda.manual_seed_all(seed)  # 多卡时覆盖所有 GPU 的默认发生器
    torch.backends.cudnn.deterministic = True   # cuDNN 用确定性算法
    torch.backends.cudnn.benchmark = False      # 关闭"按形状自动选最快算法"
    # 上面两行合起来：结果逐位可复现，代价是部分算子略慢（见"核心概念速览"第 5 条）


def worker_init_fn(worker_id):
    """DataLoader worker 进程内重置 numpy/random。
    torch.initial_seed() 已由主进程 seed 派生。"""
    # worker_id 是第几个子进程；torch.initial_seed() 由主进程种子派生而来
    worker_seed = torch.initial_seed() % 2**32
    np.random.seed(worker_seed)           # 每个 worker 拿到自己独立的随机序列
    random.seed(worker_seed)

config/config.yaml（修改，仅加 seed）：

# CIFAR-10 图像分类实验配置
seed: 42                       # 全局随机种子（命令行 --seed 可覆盖）
device: auto                   # auto = 有 GPU 用 GPU，没有用 CPU

data:
  root: ./data
  batch_size: 64
  num_workers: 2               # 数据加载子进程数（配合 worker_init_fn 固定随机源）
  val_ratio: 0.1               # 从训练集划出 10% 当验证集（random_split 用）
  augmentation:
    random_crop_padding: 4     # 随机裁剪前先四周补 4 像素
    horizontal_flip: true      # 随机水平翻转
    color_jitter: [0.2, 0.2, 0.2, 0.1]   # 亮度/对比度/饱和度/色相抖动幅度

model:
  num_classes: 10

train:
  epochs: 30
  lr: 0.001
  lr_scheduler: step
  lr_step_size: 15
  lr_gamma: 0.1
  patience: 7                  # 早停：连续 7 轮 val_acc 无提升就停
  use_amp: true                # 混合精度（第 10 章学的）

checkpoint:
  dir: ./checkpoints
log:
  dir: ./runs/cifar10

dataset/datasets.py（修改，build_dataloaders 支持固定 worker 种子）：

"""数据集与数据加载：build_transforms 构建变换，build_dataloaders 组装三个 loader。"""
from torch.utils.data import DataLoader, random_split
from torchvision import datasets, transforms


def build_transforms(cfg, train=False):
    """按配置构建训练/验证两套变换。训练集做增强，评估集只归一化。"""
    aug = cfg.data.augmentation
    transforms_list = []
    if train:
        # 只有训练集做增强；下面三个变换都是随机的，属于本章要固定的随机源
        if aug.random_crop_padding > 0:
            transforms_list.append(
                transforms.RandomCrop(32, padding=aug.random_crop_padding))
        if aug.horizontal_flip:
            transforms_list.append(transforms.RandomHorizontalFlip())
        if aug.color_jitter is not None:
            transforms_list.append(transforms.ColorJitter(*aug.color_jitter))
    transforms_list += [
        transforms.ToTensor(),
        transforms.Normalize((0.4914, 0.4822, 0.4465), (0.2470, 0.2435, 0.2616)),
    ]
    return transforms.Compose(transforms_list)


def build_dataloaders(cfg, worker_init_fn=None):
    """返回 (train_loader, val_loader, test_loader) 三元组。
    worker_init_fn 传入 utils.seed 的 worker_init_fn 可固定 worker 随机源。"""
    train_dataset = datasets.CIFAR10(root=cfg.data.root, train=True, download=True,
                                     transform=build_transforms(cfg, train=True))
    test_dataset = datasets.CIFAR10(root=cfg.data.root, train=False, download=True,
                                    transform=build_transforms(cfg, train=False))
    # random_split 用的是 Python 内置 random，必须发生在 set_seed 之后（概念 7）
    val_size = int(len(train_dataset) * cfg.data.val_ratio)
    train_dataset, val_dataset = random_split(
        train_dataset, [len(train_dataset) - val_size, val_size]
    )

    # 局部小函数：避免把 batch_size/num_workers 重复写三遍
    def make_loader(ds, shuffle):
        return DataLoader(ds, batch_size=cfg.data.batch_size, shuffle=shuffle,
                          num_workers=cfg.data.num_workers,
                          worker_init_fn=worker_init_fn)

    return (make_loader(train_dataset, shuffle=True),
            make_loader(val_dataset, shuffle=False),
            make_loader(test_dataset, shuffle=False))

train.py（修改，入口固定种子）：

"""训练入口：解析参数 -> 固定种子 -> 加载配置 -> 构建组件 -> 启动训练。"""
import argparse
import torch
import torch.nn as nn
import torch.optim as optim
from torch.utils.tensorboard import SummaryWriter
import yaml

# 唯一的"上帝文件"：import 所有模块，import 方向决定整个工程的依赖方向
from dataset.datasets import build_dataloaders
from models.classifier import build_model
from engine.trainer import Trainer
from utils.logger import setup_logger
from utils.seed import set_seed, worker_init_fn   # 本章新增：固定种子 + worker 种子


def parse_args():
    # 命令行参数：全部有默认值，不传就用配置文件里的
    parser = argparse.ArgumentParser(description="CIFAR-10 图像分类训练")
    parser.add_argument("--config", type=str, default="config/config.yaml",
                        help="YAML 配置文件路径")
    parser.add_argument("--lr", type=float, default=None, help="覆盖配置文件中的 train.lr")
    parser.add_argument("--epochs", type=int, default=None, help="覆盖配置文件中的 train.epochs")
    parser.add_argument("--batch-size", type=int, default=None,
                        help="覆盖配置文件中的 data.batch_size")
    parser.add_argument("--resume", type=str, default=None,
                        help="从指定 checkpoint 恢复训练")
    parser.add_argument("--seed", type=int, default=None, help="覆盖配置文件中的 seed")
    return parser.parse_args()


class Config(dict):
    # 让 dict 支持 cfg.train.lr 这样的点号访问，而不是 cfg["train"]["lr"]
    def __getattr__(self, key):
        try:
            return self[key]
        except KeyError as e:
            raise AttributeError(key) from e


def _to_config(obj):
    # 把 yaml 读出来的普通 dict/list 递归转成 Config
    if isinstance(obj, dict):
        return Config({k: _to_config(v) for k, v in obj.items()})
    if isinstance(obj, list):
        return [_to_config(v) for v in obj]
    return obj


def load_config(path):
    with open(path, "r", encoding="utf-8") as f:
        return _to_config(yaml.safe_load(f))


def build_scheduler(optimizer, cfg):
    # 按配置选择学习率调度器；不支持的名字返回 None（等价于不用调度器）
    if cfg.train.lr_scheduler == "step":
        return optim.lr_scheduler.StepLR(
            optimizer, step_size=cfg.train.lr_step_size, gamma=cfg.train.lr_gamma)
    if cfg.train.lr_scheduler == "cosine":
        return optim.lr_scheduler.CosineAnnealingLR(
            optimizer, T_max=cfg.train.epochs)
    return None


def main():
    args = parse_args()
    cfg = load_config(args.config)
    # 命令行参数优先：传了就覆盖配置文件里的值
    if args.lr is not None:
        cfg.train.lr = args.lr
    if args.epochs is not None:
        cfg.train.epochs = args.epochs
    if args.batch_size is not None:
        cfg.data.batch_size = args.batch_size
    if args.seed is not None:
        cfg.seed = args.seed

    # 关键顺序：set_seed 必须在 random_split、模型初始化、DataLoader 之前
    # 种子只对"之后"的随机操作生效，放晚一步等于没固定（见"核心概念速览"第 7 条）
    set_seed(cfg.seed)

    logger = setup_logger(cfg.log.dir)
    writer = SummaryWriter(log_dir=cfg.log.dir)
    # device: auto = 有 GPU 用 GPU，否则用 CPU
    device = torch.device("cuda" if cfg.device == "auto" and torch.cuda.is_available()
                          else cfg.device)
    logger.info(f"Using device: {device}")

    # worker_init_fn 让 DataLoader 子进程也确定性（num_workers>0 时必须有）
    train_loader, val_loader, test_loader = build_dataloaders(
        cfg, worker_init_fn=worker_init_fn)
    model = build_model(cfg).to(device)
    optimizer = optim.Adam(model.parameters(), lr=cfg.train.lr)
    scheduler = build_scheduler(optimizer, cfg)
    use_amp = cfg.train.use_amp and device.type == "cuda"
    scaler = torch.amp.GradScaler("cuda", enabled=use_amp)
    if use_amp:
        logger.info("AMP 已启用（FP16 混合精度）")

    # 组装阶段：把"配件"一个个造好，最后一次性注入 Trainer
    trainer = Trainer(cfg, model, optimizer, scheduler, scaler,
                      train_loader, val_loader, test_loader, logger, writer)
    trainer.train(resume=args.resume)
    writer.close()


if __name__ == "__main__":
    main()

requirements.txt（修改，显式加入 numpy）：

torch>=2.0
torchvision>=0.15
numpy>=1.24
pyyaml>=6.0
tensorboard>=2.13
matplotlib>=3.7
tqdm>=4.66
rich>=13.0
torchsummary>=1.5

运行方式与第 11 章一致，但多了一个可复现的验证姿势：

pip install -r requirements.txt
python train.py --epochs 5 > run1.log
python train.py --epochs 5 > run2.log
diff run1.log run2.log

diff 没有任何输出 = 两个文件逐字节相同 = 两次训练从数据切分、模型初始化、洗牌顺序到每个 epoch 的 loss/acc 逐位一致。注意：runs/cifar10/train.log 是追加模式，两次运行会写进同一个文件，不适合直接 diff，所以这里把标准输出重定向到两个独立文件再比较。

六、本章小结

• 学到了什么
  • 随机源有 5 处：random、numpy、torch（CPU/GPU）、cuDNN 的算法选择、DataLoader worker，缺一不可。
  • torch.manual_seed 一次固定 CPU 与当前 GPU；manual_seed_all 覆盖所有 GPU（多卡）。
  • cudnn.deterministic=True + benchmark=False 是"逐位可复现"的最后一块拼图，代价是部分算子略慢。
  • worker 子进程的随机源必须用 worker_init_fn 单独重置：子进程不继承主进程后续的随机序列，但会从主进程种子派生出自己的一份。
  • 固定种子的时机必须在一切随机操作之前：random_split、模型初始化、DataLoader 洗牌、数据增强都不能先于它。
• 常见坑
  • seed 固定太晚：random_split 或模型初始化发生在 set_seed 之前，等于没固定，白白产生"复现不了的实验"。
  • 只固定 torch 不固定 random/numpy：random_split 用的是 Python 内置 random，照样随机。
  • 依赖版本不同步：同一 seed 在 torch 1.x 与 2.x 下结果仍可能不同，重度复现建议用 pip freeze 锁定环境。
  • 可复现 ≠ 指标更好：数据增强固定后，每个 epoch 看到的样本顺序也固定了，相当于少了一点"免费随机"，正则化效果略降。
  • 多卡下 seed 只保证"每张卡内的序列"，跨卡同步仍可能引入随机性（第 13 章再讨论）。
• 下一章预告：单卡训练大模型太慢，一块显卡的利用率上不去——第 13 章先上 DataParallel，再上 DistributedDataParallel 多卡训练，届时你会理解为什么"多卡 + 可复现"要更小心。

七、动手练习

1. 验证复现：连续两次 python train.py --epochs 5 > run1.log / > run2.log，用 diff run1.log run2.log 确认逐位一致；然后注释掉 set_seed(cfg.seed) 再对比两次日志，观察差异并解释原因。
2. 只固定一半：把 set_seed 里 random.seed 和 np.random.seed 两行去掉，只留 torch 相关，跑两次看 random_split 的划分是否仍可复现（思考：random_split 用的是哪个随机源？）。
3. 成本测量：对比 cudnn.benchmark=True/False 两种配置下单 epoch 的平均耗时差，量化"确定性"的成本，并说说什么情况下值得付、什么情况下不值得。
4. 换种子：分别用 --seed 0、--seed 42、--seed 2024 各跑一遍，对比最终 val_acc，理解"种子不同 = 不同的随机序列 = 不同的小数点后几位"。
5. 折腾 num_workers：把 num_workers 改成 0 和 2 各跑两次，验证两种配置下都能复现，体会 worker_init_fn 在 num_workers=0（无子进程）时为何不影响结果。