# 勤奋的凯尔森同学10000h刻意练习Day21:训练工程化与现代CNN实战 —— 数据管道、残差、迁移学习到CIFAR-10架构对决

所属专栏: 凯尔森的10000h刻意练习
Slogan: 一个36岁程序员, 研究生(学硕),前世界500强top20软件开发Java工程师,用10000小时深耕AI, 目标Top10 AI资深工程师, 全程日更记录。

模型能跑了,可怎么让每次结果可复现?怎么在数据不够时站到ImageNet的肩膀上?

Day20用PyTorch跑通了训练闭环、学完了CNN五大件。今天先把训练从"能跑"升级到"工程可用":用Dataset和DataLoader搭数据管道,train/val/test三分各司其职,设备无关代码一份跑遍CPU/CUDA/MPS,再把训练循环沉淀成可复用的fit函数——加上学习曲线、早停、模型存取和超参搜索,工程闭环就齐了。然后回到CNN本身:现代CNN的残差连接让网络敢深,数据增强凭空变出更多训练数据,迁移学习让你站在ImageNet的肩膀上。最后在CIFAR-10上亲手训练AlexNet、VGG和ResNet,三家正面对决。

# 1.1 数据管道:Dataset与DataLoader

Day20里我们用过TensorDataset和DataLoader,但只当它们是"打包工具"。今天先看清这两个抽象各自的职责。

Dataset负责按索引取一个样本,最核心的三个方法是:__init__里准备好数据、__len__返回样本总数、__getitem__按下标返回(x, y)。DataLoader则负责在Dataset之上做分批、打乱、并行加载,它拿到的是一个个batch而不是单个样本。

import numpy as np
import torch
from torch.utils.data import Dataset, DataLoader
from sklearn.datasets import load_digits

# 8x8手写数字: 1797个样本, 像素0~16, 归一化到0~1
digits = load_digits()
X = digits.data.astype("float32") / 16.0
y = digits.target.astype("int64")
print(X.shape, X.min(), X.max())

class DigitsDataset(Dataset):
    def __init__(self, X, y, transform=None):
        # 数据只在内存里存一份, transform在取样时才施加
        self.X = torch.tensor(X)
        self.y = torch.tensor(y)
        self.transform = transform

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

    def __getitem__(self, idx):
        x, label = self.X[idx], int(self.y[idx])
        if self.transform is not None:
            x = self.transform(x)
        return x, label

dataset = DigitsDataset(X, y)
print("样本总数:", len(dataset))
x0, y0 = dataset[0]  # 像列表一样按下标取样
print("单样本形状:", x0.shape, "标签:", y0)
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
(1797, 64) 0.0 1.0
样本总数: 1797
单样本形状: torch.Size([64]) 标签: 0
1
2
3

transform是数据管道的关键钩子:预处理在取样瞬间完成,而不是提前复制出多份数据。最简单的"变换"就是一个接收张量、返回张量的函数,用组合器串起来:

class Compose:
    """把多个变换按顺序串起来, torchvision.transforms.Compose是它的图像增强版"""
    def __init__(self, transforms):
        self.transforms = transforms

    def __call__(self, x):
        for t in self.transforms:
            x = t(x)
        return x

def add_gaussian_noise(x, sigma=0.05):
    # 加噪是最朴素的数据增强, Day21会用torchvision做翻转/裁剪等正经增强
    return x + sigma * torch.randn_like(x)

# 训练集带噪声增强, 验证/测试集不带
train_dataset_aug = DigitsDataset(X, y, transform=Compose([add_gaussian_noise]))
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16

DataLoader的几个高频参数:batch_size每批多少条、shuffle是否每轮打乱(训练True,验证测试False)、drop_last最后一批不足batch时是否丢弃(BatchNorm依赖较大批量时建议True)、num_workers并行进程数(Mac/Windows上用0最稳,Linux服务器可调大)。

loader = DataLoader(dataset, batch_size=64, shuffle=True)
for x_batch, y_batch in loader:
    print("一个batch:", x_batch.shape, y_batch.shape)
    break
1
2
3
4
一个batch: torch.Size([64, 64]) torch.Size([64])
1

注意Dataset吐出的单样本是64维向量,DataLoader自动在最前面拼出了batch维——这就是为什么模型里见到的永远是(N, 64)。

# 1.2 train/val/test三分:三份数据三种职责

Day20我们用测试集准确率一边训练一边观察,这在工程上是违规的:测试集一旦参与了任何决策(选超参、选epoch、选模型),它就不再是"没见过的数据",报出的准确率会虚高。规范做法是切三份:

  • 训练集(train)(约60%):梯度下降真正拟合的数据;
  • 验证集(val)(约20%):训练过程中反复看,用来选epoch、选超参、选模型;
  • 测试集(test)(约20%):整个实验只在最后评估一次,模拟真实世界的未见数据。

切分用两次train_test_split,两次都按标签分层抽样(stratify)、固定random_state保证可复现:

from sklearn.model_selection import train_test_split

# 先切出20%作为测试集, 剩下80%再按1/4切出验证集(0.8*0.25=0.2)
X_tv, X_test, y_tv, y_test = train_test_split(
    X, y, test_size=0.2, stratify=y, random_state=42)
X_train, X_val, y_train, y_val = train_test_split(
    X_tv, y_tv, test_size=0.25, stratify=y_tv, random_state=42)
print(len(X_train), len(X_val), len(X_test))

train_dataset = DigitsDataset(X_train, y_train)
val_dataset = DigitsDataset(X_val, y_val)
test_dataset = DigitsDataset(X_test, y_test)
1
2
3
4
5
6
7
8
9
10
11
12
1077 360 360
1

一条纪律:测试集在所有超参和训练轮数敲定之前不许碰。最后跑出的测试准确率,才是可以写进报告的数字。

# 1.3 设备无关:同一份代码跑遍CPU/CUDA/MPS

Day19已经验证过本机的设备情况。工程上的标准写法是只在一处决定device,之后所有模型和数据都通过.to(device)搬运,代码本身不出现任何设备硬编码:

device = torch.device(
    "cuda" if torch.cuda.is_available()
    else "mps" if torch.backends.mps.is_available()
    else "cpu"
)
print(device)  # N卡机上输出cuda, Apple Silicon上输出mps, 其余输出cpu
1
2
3
4
5
6
mps
1

两个容易忘的动作:①模型和数据要搬到同一个设备,模型在MPS、张量在CPU会直接报错;②GPU上的张量不能直接.numpy(),要先.cpu()。下面的训练循环会把这些规矩固化进去。

# 1.4 规范训练循环:一个fit函数打天下

把Day20反复抄写的训练流程沉淀成两个可复用函数:run_epoch跑一个epoch(训练/评估两用),fit负责组装模型、优化器、循环和历史记录。顺手把"可复现"的种子设置也固化下来——随机初始化、DataLoader洗牌、数据增强里的随机数,全靠同一个种子钉死。

import copy
import random
import torch.nn as nn

def set_seed(seed=42):
    """让numpy/random/torch三者的随机数全部可复现"""
    random.seed(seed)
    np.random.seed(seed)
    torch.manual_seed(seed)
    if torch.cuda.is_available():
        torch.cuda.manual_seed_all(seed)
    torch.backends.cudnn.benchmark = False  # 固定卷积算法选择

def build_model(hidden=128, dropout=0.2):
    return nn.Sequential(
        nn.Linear(64, hidden), nn.ReLU(), nn.Dropout(dropout),
        nn.Linear(hidden, hidden // 2), nn.ReLU(), nn.Dropout(dropout),
        nn.Linear(hidden // 2, 10)
    )

def run_epoch(model, loader, loss_fn, optimizer=None):
    """训练/评估两用: 传optimizer就是训练, 不传就是评估"""
    is_train = optimizer is not None
    model.train(is_train)          # train()/eval()开关, 决定Dropout/BN行为
    total_loss, correct, n = 0.0, 0, 0
    for x_batch, y_batch in loader:
        x_batch, y_batch = x_batch.to(device), y_batch.to(device)
        with torch.set_grad_enabled(is_train):  # 评估时关掉梯度, 省内存还提速
            output = model(x_batch)
            loss = loss_fn(output, y_batch)
        if is_train:
            optimizer.zero_grad()
            loss.backward()
            optimizer.step()
        total_loss += loss.item() * len(y_batch)
        correct += (output.argmax(dim=1) == y_batch).sum().item()
        n += len(y_batch)
    return total_loss / n, correct / n

def fit(hidden=128, dropout=0.2, lr=1e-3, weight_decay=0.0,
        epochs=40, batch_size=64, patience=None, seed=42):
    set_seed(seed)
    model = build_model(hidden, dropout).to(device)
    optimizer = torch.optim.Adam(model.parameters(), lr=lr, weight_decay=weight_decay)
    loss_fn = nn.CrossEntropyLoss()
    # 给DataLoader单独的随机数生成器并播种, shuffle顺序也可复现
    generator = torch.Generator()
    generator.manual_seed(seed)
    train_loader = DataLoader(train_dataset, batch_size, shuffle=True, generator=generator)
    val_loader = DataLoader(val_dataset, batch_size)  # 验证集不打乱
    history = {"train_loss": [], "val_loss": [], "train_acc": [], "val_acc": []}
    # 为早停预做准备: 记录最佳验证损失与对应权重
    best_val_loss, best_epoch, best_state, bad_epochs = float("inf"), -1, None, 0
    for epoch in range(epochs):
        tr_loss, tr_acc = run_epoch(model, train_loader, loss_fn, optimizer)
        va_loss, va_acc = run_epoch(model, val_loader, loss_fn)
        history["train_loss"].append(tr_loss)
        history["val_loss"].append(va_loss)
        history["train_acc"].append(tr_acc)
        history["val_acc"].append(va_acc)
        if va_loss < best_val_loss - 1e-4:
            best_val_loss, best_epoch, bad_epochs = va_loss, epoch, 0
            best_state = copy.deepcopy(model.state_dict())  # 深拷当前最佳权重
        else:
            bad_epochs += 1
            if patience is not None and bad_epochs >= patience:
                break
    if best_state is not None:
        model.load_state_dict(best_state)  # 返回最佳时刻的模型
    return model, history, best_val_loss, best_epoch, epoch + 1
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70

跑一次标准配置(MPS上约6秒):

model, history, best_val_loss, best_epoch, n_ran = fit(epochs=40, seed=42)
print(f"训练/验证准确率: {history['train_acc'][-1]:.4f} / {history['val_acc'][-1]:.4f}")
print(f"最佳验证损失: {best_val_loss:.4f} @ epoch {best_epoch}")
1
2
3
训练/验证准确率: 0.9694 / 0.9722
最佳验证损失: 0.0861 @ epoch 39
1
2

这里有个反直觉但正确的细节:best_state用copy.deepcopy而不是直接赋值——state_dict()里装的是张量引用,直接保存的"最佳权重"会随后续训练继续被原地改写。

# 1.5 学习曲线:模型状态的体检报告

fit返回的history不是可有可无的日志,它是判断模型状态的依据。把训练/验证的loss曲线画出来:

import matplotlib.pyplot as plt

fig, axes = plt.subplots(1, 2, figsize=(11, 4))
axes[0].plot(history["train_loss"], label="train")
axes[0].plot(history["val_loss"], label="val")
axes[0].set_xlabel("epoch"); axes[0].set_ylabel("loss"); axes[0].legend()
axes[1].plot(history["train_acc"], label="train")
axes[1].plot(history["val_acc"], label="val")
axes[1].set_xlabel("epoch"); axes[1].set_ylabel("accuracy"); axes[1].legend()
plt.show()
1
2
3
4
5
6
7
8
9
10

看曲线有三种典型病征,处方完全不同:

曲线形态 诊断 处方
train和val都在高位徘徊、loss都降不下去 欠拟合(容量不足) 加大模型/加深、训练更久、调大lr
train一路降到0,val先降后升、两条线张口 过拟合(背训练集) 加正则/Dropout/数据增强、早停、更多数据
两条线贴合着一起下降并趋于平稳 健康 可以再小步调超参榨性能

故意把网络加大、去掉Dropout和权值衰减训60轮,就能观察过拟合的苗头:

_, history_overfit, _, best_ep, _ = fit(
    hidden=256, dropout=0.0, weight_decay=0.0, epochs=60, seed=42)
for ep in [5, 30, 59]:
    print(f"epoch{ep:>2}: train_loss={history_overfit['train_loss'][ep]:.4f}, "
          f"val_loss={history_overfit['val_loss'][ep]:.4f}, "
          f"train_acc={history_overfit['train_acc'][ep]:.3f}, "
          f"val_acc={history_overfit['val_acc'][ep]:.3f}")
1
2
3
4
5
6
7
epoch 5: train_loss=0.2823, val_loss=0.3187, train_acc=0.923, val_acc=0.911
epoch30: train_loss=0.0274, val_loss=0.0940, train_acc=0.997, val_acc=0.967
epoch59: train_loss=0.0061, val_loss=0.0800, train_acc=1.000, val_acc=0.972
1
2
3

训练loss从0.28压到0.006、训练准确率100%,验证loss在第56轮触底0.0791后已不再下降——模型开始用极致拟合训练集换取泛化,两条线之间的缝隙就是过拟合。手写数字数据集太小太"干净",缝隙不算夸张;在复杂图像任务上这个张口能大到十几个百分点,一眼可见。

# 1.6 早停:让fit自动收手

1.4的fit里已经埋好了早停逻辑:给patience传一个整数,验证损失连续patience轮不刷新最低纪录就停训,并自动回滚到最佳权重。这正是Day18讲早停法时许下的"完整PyTorch实现":

  • 判定信号是验证损失(不是训练loss,也不是准确率波动);
  • 停下后返回的是最佳时刻深拷的权重,不是反弹后的权重;
  • patience给了容忍波动的缓冲带,不会因为一次抖动误杀。
model_es, history_es, best_val, best_epoch, n_ran = fit(
    epochs=200, patience=15, seed=42)
print(f"第{n_ran}个epoch触发早停, 回滚到第{best_epoch}轮, "
      f"验证损失{best_val:.4f}, 验证准确率{history_es['val_acc'][best_epoch]:.4f}")
1
2
3
4
第160个epoch触发早停, 回滚到第144轮, 验证损失0.0396, 验证准确率0.9833
1

省下了40个无效epoch,最终留下的还是最佳权重。工程上patience常设10~20:太小容易误停,太大又省不下时间。

# 1.7 模型存取:state_dict与断点续训

训练好的模型不能只活在内存里。PyTorch存模型的规范做法是存**state_dict**(一个参数名到张量的字典),而不是把整个模型对象pickle下来——后者强绑定当时的类定义路径,换个目录就加载不了。

# 只存权重: 加载时需要先实例化同样结构的模型
torch.save(model_es.state_dict(), "digits_mlp.pt")

loaded_model = build_model().to(device)
loaded_model.load_state_dict(torch.load("digits_mlp.pt", weights_only=True))
1
2
3
4
5

长时间训练还要防"训练到一半进程挂了"——这时只存权重不够,要把优化器状态、当前epoch、最佳指标一起打包成checkpoint,恢复后从断点继续,动量等优化器内部累积量也不会丢。

保存时机也有讲究:真实项目里checkpoint应在训练过程中验证损失刷新最佳时写盘——这样磁盘上永远躺着历史最优断点,进程挂掉后恢复的也是它。为此把1.4的fit升级一版:新增ckpt_path参数,默认None不写盘,传入路径时在"验证损失刷新最佳"的分支里把权重、优化器状态、当前epoch、最佳指标打包落盘。完整的fit代码如下:

def fit(hidden=128, dropout=0.2, lr=1e-3, weight_decay=0.0,
        epochs=40, batch_size=64, patience=None, seed=42, ckpt_path=None):
    set_seed(seed)
    model = build_model(hidden, dropout).to(device)
    optimizer = torch.optim.Adam(model.parameters(), lr=lr, weight_decay=weight_decay)
    loss_fn = nn.CrossEntropyLoss()
    # 给DataLoader单独的随机数生成器并播种, shuffle顺序也可复现
    generator = torch.Generator()
    generator.manual_seed(seed)
    train_loader = DataLoader(train_dataset, batch_size, shuffle=True, generator=generator)
    val_loader = DataLoader(val_dataset, batch_size)  # 验证集不打乱
    history = {"train_loss": [], "val_loss": [], "train_acc": [], "val_acc": []}
    # 为早停预做准备: 记录最佳验证损失与对应权重
    best_val_loss, best_epoch, best_state, bad_epochs = float("inf"), -1, None, 0
    for epoch in range(epochs):
        tr_loss, tr_acc = run_epoch(model, train_loader, loss_fn, optimizer)
        va_loss, va_acc = run_epoch(model, val_loader, loss_fn)
        history["train_loss"].append(tr_loss)
        history["val_loss"].append(va_loss)
        history["train_acc"].append(tr_acc)
        history["val_acc"].append(va_acc)
        if va_loss < best_val_loss - 1e-4:
            best_val_loss, best_epoch, bad_epochs = va_loss, epoch, 0
            best_state = copy.deepcopy(model.state_dict())  # 深拷当前最佳权重
            # checkpoint保存逻辑放这里: 验证损失刷新最佳时写盘
            if ckpt_path is not None:
                checkpoint = {
                    "epoch": epoch,
                    "model_state": best_state,
                    "optimizer_state": optimizer.state_dict(),  # Adam的动量也存在这里
                    "best_val_loss": best_val_loss,
                }
                torch.save(checkpoint, ckpt_path)
        else:
            bad_epochs += 1
            if patience is not None and bad_epochs >= patience:
                break
    if best_state is not None:
        model.load_state_dict(best_state)  # 返回最佳时刻的模型
    return model, history, best_val_loss, best_epoch, epoch + 1
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40

用升级版fit重跑1.6的早停训练:同一个种子下结果与1.6完全一致,区别只是训练过程中digits_checkpoint.pt被自动写盘,且每次刷新最佳纪录都会覆盖更新:

model_es, history_es, best_val, best_epoch, n_ran = fit(
    epochs=200, patience=15, seed=42, ckpt_path="digits_checkpoint.pt")
print(f"第{n_ran}个epoch触发早停, 最佳验证损失{best_val:.4f}, checkpoint已写盘")
1
2
3
第160个epoch触发早停, 最佳验证损失0.0396, checkpoint已写盘
1

断点续训时,把checkpoint里存的东西逐项恢复,就能从下一个epoch接着训:

# 断点续训: 逐项恢复后, 从下一个epoch接着训
ckpt = torch.load("digits_checkpoint.pt", weights_only=False)
model = build_model().to(device)
optimizer = torch.optim.Adam(model.parameters())
model.load_state_dict(ckpt["model_state"])
optimizer.load_state_dict(ckpt["optimizer_state"])
start_epoch = ckpt["epoch"] + 1
print(f"从第{start_epoch}轮续训, 当时最佳验证损失{ckpt['best_val_loss']:.4f}")
1
2
3
4
5
6
7
8
从第145轮续训, 当时最佳验证损失0.0396
1

加载恢复出来的模型直接在测试集上验收——这是测试集第一次也是唯一一次出场:

test_loader = DataLoader(test_dataset, 128)
test_loss, test_acc = run_epoch(model, test_loader, nn.CrossEntropyLoss())
print(f"测试集准确率: {test_acc:.4f}")
1
2
3
测试集准确率: 0.9722
1

# 1.8 超参随机搜索:用验证集系统地挑

最后一环是超参选择。手动一个个试的问题是:学习率的合理取值跨了好几个数量级(0.0001和0.003的差距是30倍),在线性网格上均匀采样会把大量预算浪费在0.002~0.003这种"没区别"的区间。正确做法是在对数尺度上均匀采样学习率,权值衰减同理,然后每组超参都用同一个种子训练、用验证损失排名:

rng = np.random.default_rng(0)
results = []
for trial in range(8):
    # lr在1e-4~3e-3之间对数均匀采样
    lr = 10 ** rng.uniform(np.log10(1e-4), np.log10(3e-3))
    weight_decay = 10 ** rng.uniform(-5, -2)
    _, _, val_loss, best_ep, _ = fit(
        lr=lr, weight_decay=weight_decay, epochs=30, seed=42)
    results.append((trial, lr, weight_decay, val_loss, best_ep))
    print(f"trial{trial}: lr={lr:.5f}, weight_decay={weight_decay:.5f}, "
          f"val_loss={val_loss:.4f}, best_epoch={best_ep}")

best = min(results, key=lambda r: r[3])  # 验证损失最低者获胜
print(f"最优: trial{best[0]}, lr={best[1]:.5f}, weight_decay={best[2]:.5f}")
1
2
3
4
5
6
7
8
9
10
11
12
13
14
trial0: lr=0.00087, weight_decay=0.00006, val_loss=0.1223, best_epoch=28
trial1: lr=0.00011, weight_decay=0.00001, val_loss=0.9084, best_epoch=29
trial2: lr=0.00159, weight_decay=0.00547, val_loss=0.1467, best_epoch=28
trial3: lr=0.00079, weight_decay=0.00154, val_loss=0.1468, best_epoch=28
trial4: lr=0.00064, weight_decay=0.00639, val_loss=0.2312, best_epoch=28
trial5: lr=0.00160, weight_decay=0.00001, val_loss=0.0837, best_epoch=28
trial6: lr=0.00185, weight_decay=0.00001, val_loss=0.0825, best_epoch=28
trial7: lr=0.00120, weight_decay=0.00003, val_loss=0.1009, best_epoch=28
最优: trial6, lr=0.00185, weight_decay=0.00001
1
2
3
4
5
6
7
8
9

结果很有信息量:trial1的学习率太小(0.00011),30个epoch还在0.9的高损失区间没走出来;trial4的权值衰减过重(0.006)把模型约束得太死;trial6以较大学习率配极小衰减胜出。挑出最优后,最后用全部训练数据、最优超参重新训练一次,在测试集上报告最终成绩:

final_model, _, _, _, _ = fit(lr=best[1], weight_decay=best[2], epochs=30, seed=42)
_, final_test_acc = run_epoch(final_model, test_loader, nn.CrossEntropyLoss())
print(f"最终测试集准确率: {final_test_acc:.4f}")
1
2
3
最终测试集准确率: 0.9722
1

这就是一轮完整、规范的实验闭环:固定种子保证可复现→三分数据防止作弊→验证集+随机搜索选超参→早停定轮数→checkpoint防中断→测试集只报一次数。随机搜索只有8次试验,看着朴素,但在超参维度不超过3、4个时,它的效率往往不输昂贵的贝叶斯优化,而且结果一目了然。

# 1.9 卷积的工程实现:im2col直觉

先回答一个很多人忽略的问题:卷积窗口在特征图上挪来挪去做乘积累加,这种"跳着访问内存"的运算对GPU其实很不友好。框架们的真实做法是im2col(image to column):把每个滑动窗口覆盖的小区域提前"展开"成矩阵的一行,卷积就变成了两个大矩阵相乘(GEMM),而矩阵乘法正是GPU最擅长、库优化最充分的运算。

以4×4单通道输入、3×3卷积核、无填充为例:窗口一共停在个位置,im2col把每个窗口的9个数展成一行,得到4行9列的矩阵。PyTorch里可以用F.unfold直接看到这个展开过程:

import torch
import torch.nn.functional as F

x = torch.arange(16, dtype=torch.float32).reshape(1, 1, 4, 4)
print(x[0, 0])
patches = F.unfold(x, kernel_size=3)  # 输出(N, 通道*核高*核宽, 窗口数)
print(patches.shape, patches[0].T.shape)
1
2
3
4
5
6
7
tensor([[ 0.,  1.,  2.,  3.],
        [ 4.,  5.,  6.,  7.],
        [ 8.,  9., 10., 11.],
        [12., 13., 14., 15.]])
torch.Size([1, 9, 4]) torch.Size([4, 9])
1
2
3
4
5

patches的4列(转置后是4行)就是4个窗口。多通道时每列长度变成;卷积核同样展成列向量,一次矩阵乘法就完成所有窗口、所有输出通道的计算。代价是展开后有数据重复(重叠区域被复制),属于"用显存换速度"。理解im2col就理解了为什么深度学习如此依赖矩阵乘法硬件。

# 1.10 现代CNN的四大标准件

# 1.10.1 小卷积堆叠:VGG的参数经济学

AlexNet里还有11×11的大核,VGG之后业界全面转向3×3小核。原因是数学上的等价与参数上的节省:两层3×3卷积的感受野等于一层5×5(从一个像素出发,第一层看到3×3,第二层在3×3的每个点上再看3×3),三层3×3等于7×7。而参数量呢?假设通道数都是:一层5×5有个参数,两层3×3只有;更别说两层之间还多夹了一次非线性激活,表达力更强。所以现代网络的基本单位是"卷积块":若干层Conv3×3 → BN → ReLU堆叠,再用池化或步幅2降分辨率、同时翻倍通道数。

# 1.10.2 1×1卷积:跨通道的信息搅拌机

1×1卷积在空间上不看邻域,初看奇怪,但它作用在通道维上:对每个像素位置,它把所有通道的值做一次线性组合。两大用途:①降维/升维,用很少的计算把通道数压下来(GoogleNet Inception的核心技巧);②跨通道信息融合。

conv11 = nn.Conv2d(in_channels=8, out_channels=2, kernel_size=1)
y = conv11(torch.randn(1, 8, 4, 4))
print(y.shape)  # 空间尺寸不变, 通道8 -> 2
1
2
3
torch.Size([1, 2, 4, 4])
1

# 1.10.3 批标准化的正确位置

Day18讲过BN的原理,现代CNN里它的标准位置是卷积之后、激活之前,并且卷积后的BN通常不需要偏置(BN自带的接管了偏置的作用,所以代码里bias=False):

def conv_bn_relu(in_ch, out_ch, kernel_size=3, stride=1, padding=1):
    return nn.Sequential(
        nn.Conv2d(in_ch, out_ch, kernel_size, stride=stride,
                  padding=padding, bias=False),
        nn.BatchNorm2d(out_ch),
        nn.ReLU()
    )
1
2
3
4
5
6
7

BN让各层输入分布稳定,我们因此敢用更大的学习率、对初始化也更宽容——这是网络能加深的重要前提。

# 1.10.4 全局平均池化:干掉末端的巨大全连接层

LeNet末端把16张5×5特征图拉平成400再接全连接层,经典网络里这种末端FC层常常占掉全模型绝大部分参数(还容易过拟合)。现代网络改用全局平均池化(Global Average Pooling, GAP):对每张特征图直接取整个空间的平均值,一张图变成一个数,(N,C,H,W)变成(N,C,1,1),再Flatten接一个和类别数等宽的线性层即可。

gap = nn.AdaptiveAvgPool2d(output_size=1)
y = gap(torch.randn(1, 64, 7, 7))
print(y.shape)  # (1,64,7,7) -> (1,64,1,1), 每张特征图只留一个数
1
2
3
torch.Size([1, 64, 1, 1])
1

AdaptiveAvgPool2d(1)的好处是不管输入特征图是7×7还是13×13,输出恒为1×1,模型对输入尺寸不再敏感。

# 1.11 残差连接手写:让网络敢深

光靠BN还不够,真正让网络能堆到几十上百层的是ResNet的残差连接。核心思想:把网络要学的目标从"直接输出"改成学残差,前向时把输入通过一条恒等捷径直接加到输出上。捷径上如果通道数或尺寸对不上(比如通道翻倍、步幅2),就用一个带步幅的1×1卷积做投影。下面的BasicBlock就是ResNet-18的标准积木:

class BasicBlock(nn.Module):
    def __init__(self, in_ch, out_ch, stride=1):
        super().__init__()
        # 主路: 3x3卷积(可能带步幅) -> BN -> ReLU -> 3x3卷积 -> BN
        self.conv1 = nn.Conv2d(in_ch, out_ch, 3, stride=stride, padding=1, bias=False)
        self.bn1 = nn.BatchNorm2d(out_ch)
        self.conv2 = nn.Conv2d(out_ch, out_ch, 3, padding=1, bias=False)
        self.bn2 = nn.BatchNorm2d(out_ch)
        # 捷径: 尺寸/通道对不上时用1x1卷积投影, 对得上就直接用恒等映射
        self.downsample = None
        if stride != 1 or in_ch != out_ch:
            self.downsample = nn.Sequential(
                nn.Conv2d(in_ch, out_ch, 1, stride=stride, bias=False),
                nn.BatchNorm2d(out_ch))

    def forward(self, x):
        identity = x if self.downsample is None else self.downsample(x)
        out = torch.relu(self.bn1(self.conv1(x)))
        out = self.bn2(self.conv2(out))
        return torch.relu(out + identity)  # 相加后再激活

# 通道32->64且步幅2: 14x14 -> 7x7, 靠1x1投影捷径对齐
block = BasicBlock(32, 64, stride=2)
print(block(torch.randn(1, 32, 14, 14)).shape)
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
torch.Size([1, 64, 7, 7])
1

把积木拼起来,定义一个迷你ResNet:stem卷一次 → 两对残差块(每对的第一个块负责降采样+升通道)→ GAP → 分类层。

modern_net = nn.Sequential(
    conv_bn_relu(1, 32),
    BasicBlock(32, 32),
    BasicBlock(32, 64, stride=2),    # 28x28 -> 14x14, 通道32 -> 64
    BasicBlock(64, 64),
    BasicBlock(64, 128, stride=2),   # 14x14 -> 7x7, 通道64 -> 128
    BasicBlock(128, 128),
    nn.AdaptiveAvgPool2d(1),
    nn.Flatten(),
    nn.Linear(128, 10)
)
x = torch.randn(2, 1, 28, 28)
for layer in modern_net:
    x = layer(x)
print(x.shape, "参数量:", sum(p.numel() for p in modern_net.parameters()))
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
torch.Size([2, 10]) 参数量: 677482
1

用Day20的Fashion MNIST流程(Adam、lr=1e-3、8个epoch)和LeNet-5(Sigmoid+SGD lr=0.9+Xavier初始化,同样8个epoch)正面对比,训练循环沿用Day21的train/eval写法:

import random
import numpy as np
import torch
import torch.nn as nn
from torch.utils.data import DataLoader, TensorDataset
from torchvision import datasets

# ---- 数据加载(与Day20一致) ----
def load_fashion_mnist():
    train_csv = "fashion-mnist_train.csv"
    test_csv = "fashion-mnist_test.csv"
    import os
    if os.path.exists(train_csv) and os.path.exists(test_csv):
        import pandas as pd
        train = pd.read_csv(train_csv)
        test = pd.read_csv(test_csv)
        X_train = torch.tensor(train.iloc[:, 1:].values, dtype=torch.float32).reshape(-1, 1, 28, 28) / 255.
        y_train = torch.tensor(train.iloc[:, 0].values, dtype=torch.int64)
        X_test = torch.tensor(test.iloc[:, 1:].values, dtype=torch.float32).reshape(-1, 1, 28, 28) / 255.
        y_test = torch.tensor(test.iloc[:, 0].values, dtype=torch.int64)
    else:
        train = datasets.FashionMNIST(root=".", train=True, download=True)
        test = datasets.FashionMNIST(root=".", train=False, download=True)
        X_train = train.data.float().reshape(-1, 1, 28, 28) / 255.
        y_train = train.targets
        X_test = test.data.float().reshape(-1, 1, 28, 28) / 255.
        y_test = test.targets
    return X_train, y_train, X_test, y_test

X_train, y_train, X_test, y_test = load_fashion_mnist()
train_dataset = TensorDataset(X_train, y_train)
test_dataset = TensorDataset(X_test, y_test)

device = torch.device("cuda" if torch.cuda.is_available()
                      else "mps" if torch.backends.mps.is_available() else "cpu")

# ---- 训练/评估函数(与Day21的run_epoch一致) ----
def run_epoch(model, loader, loss_fn, optimizer=None):
    is_train = optimizer is not None
    model.train(is_train)
    total_loss, correct, n = 0.0, 0, 0
    for x_batch, y_batch in loader:
        x_batch, y_batch = x_batch.to(device), y_batch.to(device)
        with torch.set_grad_enabled(is_train):
            output = model(x_batch)
            loss = loss_fn(output, y_batch)
        if is_train:
            optimizer.zero_grad()
            loss.backward()
            optimizer.step()
        total_loss += loss.item() * len(y_batch)
        correct += (output.argmax(dim=1) == y_batch).sum().item()
        n += len(y_batch)
    return total_loss / n, correct / n

def set_seed(seed=42):
    random.seed(seed)
    np.random.seed(seed)
    torch.manual_seed(seed)
    if torch.cuda.is_available():
        torch.cuda.manual_seed_all(seed)

# ---- 模型定义 ----
# LeNet-5: Sigmoid + AvgPool, Xavier均匀初始化
lenet = nn.Sequential(
    nn.Conv2d(1, 6, kernel_size=5, padding=2),
    nn.Sigmoid(),
    nn.AvgPool2d(kernel_size=2, stride=2),
    nn.Conv2d(6, 16, kernel_size=5),
    nn.Sigmoid(),
    nn.AvgPool2d(kernel_size=2, stride=2),
    nn.Flatten(),
    nn.Linear(400, 120), nn.Sigmoid(),
    nn.Linear(120, 84),  nn.Sigmoid(),
    nn.Linear(84, 10),
)

# 迷你ResNet: BN + ReLU + 残差连接 + GAP(定义已在1.11节给出, 这里复用)
def conv_bn_relu(in_ch, out_ch, kernel_size=3, stride=1, padding=1):
    return nn.Sequential(
        nn.Conv2d(in_ch, out_ch, kernel_size, stride=stride,
                  padding=padding, bias=False),
        nn.BatchNorm2d(out_ch),
        nn.ReLU()
    )

class BasicBlock(nn.Module):
    def __init__(self, in_ch, out_ch, stride=1):
        super().__init__()
        self.conv1 = nn.Conv2d(in_ch, out_ch, 3, stride=stride, padding=1, bias=False)
        self.bn1 = nn.BatchNorm2d(out_ch)
        self.conv2 = nn.Conv2d(out_ch, out_ch, 3, padding=1, bias=False)
        self.bn2 = nn.BatchNorm2d(out_ch)
        self.downsample = None
        if stride != 1 or in_ch != out_ch:
            self.downsample = nn.Sequential(
                nn.Conv2d(in_ch, out_ch, 1, stride=stride, bias=False),
                nn.BatchNorm2d(out_ch))
    def forward(self, x):
        identity = x if self.downsample is None else self.downsample(x)
        out = torch.relu(self.bn1(self.conv1(x)))
        out = self.bn2(self.conv2(out))
        return torch.relu(out + identity)

modern_net = nn.Sequential(
    conv_bn_relu(1, 32),
    BasicBlock(32, 32),
    BasicBlock(32, 64, stride=2),
    BasicBlock(64, 64),
    BasicBlock(64, 128, stride=2),
    BasicBlock(128, 128),
    nn.AdaptiveAvgPool2d(1),
    nn.Flatten(),
    nn.Linear(128, 10)
)

# ---- 训练对比: 同样8个epoch, 种子相同 ----
def train_compare(model, name, epochs=8, batch_size=256, seed=42, use_sgd=False):
    set_seed(seed)
    model = model.to(device)
    def init_weights(layer):
        if type(layer) == nn.Linear or type(layer) == nn.Conv2d:
            nn.init.xavier_uniform_(layer.weight)
    model.apply(init_weights)
    lr = 0.9 if use_sgd else 1e-3
    optimizer = (torch.optim.SGD(model.parameters(), lr=lr) if use_sgd
                 else torch.optim.Adam(model.parameters(), lr=lr))
    loss_fn = nn.CrossEntropyLoss()
    train_loader = DataLoader(train_dataset, batch_size, shuffle=True)
    test_loader = DataLoader(test_dataset, batch_size, shuffle=False)
    for epoch in range(epochs):
        tr_loss, tr_acc = run_epoch(model, train_loader, loss_fn, optimizer)
        te_loss, te_acc = run_epoch(model, test_loader, loss_fn)
        print(f"[{name}] epoch {epoch+1}/{epochs}  "
              f"train_acc {tr_acc:.4f}  test_acc {te_acc:.4f}")
    return tr_acc, te_acc

# LeNet-5: SGD lr=0.9 + Xavier初始化 (与Day20一致)
tr1, te1 = train_compare(lenet, "LeNet-5", use_sgd=True)
# 迷你ResNet: Adam lr=1e-3 (现代CNN标配)
tr2, te2 = train_compare(modern_net, "迷你ResNet", use_sgd=False)
print(f"\nLeNet-5:   train_acc {tr1:.4f}, test_acc {te1:.4f}")
print(f"迷你ResNet: train_acc {tr2:.4f}, test_acc {te2:.4f}")
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
LeNet-5:   train_acc 0.8001, test_acc 0.7935
迷你ResNet: train_acc 0.9657, test_acc 0.8984
1
2

同样的数据、同样的epoch数,现代构件带来12个百分点的测试准确率提升(MPS上约3分钟)。注意LeNet这里如果忘了Xavier初始化,Sigmoid会大面积饱和、8个epoch后还停在10%瞎猜——Day18初始化实验里的警告在实战中再次应验。

# 1.12 数据增强:用变换"凭空"变出更多训练数据

过拟合的根源是数据不够模型背。除了Day18的正则化,计算机视觉还有一招独门武器数据增强(data augmentation):对训练图片施加不改变语义的随机变换(翻转、旋转、裁剪、变色),等于每个epoch都给模型看一批"新"图片。torchvision的transforms把这些变换封装成了可组合的管道。

from torchvision import datasets, transforms

# 训练集: 随机旋转+翻转做增强, 再ToTensor并标准化
train_transform = transforms.Compose([
    transforms.RandomRotation(degrees=10),            # 随机旋转±10度
    transforms.RandomHorizontalFlip(p=0.5),           # 50%概率水平翻转
    transforms.ToTensor(),                            # PIL -> 张量并/255
    transforms.Normalize(mean=(0.5,), std=(0.5,))     # 标准化到-1~1
])
# 验证/测试集: 只做确定性的ToTensor+标准化, 绝不能加随机变换!
eval_transform = transforms.Compose([
    transforms.ToTensor(),
    transforms.Normalize(mean=(0.5,), std=(0.5,))
])

train_dataset = datasets.FashionMNIST(
    root="data", train=True, download=True, transform=train_transform)
test_dataset = datasets.FashionMNIST(
    root="data", train=False, download=True, transform=eval_transform)
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19

三条纪律:

  1. 增强只加在训练集上。验证/测试集必须用确定性变换,否则评估结果带随机噪声、不可复现;
  2. 变换不能改变语义。Fashion MNIST里鞋子、裤子左右翻转一般不影响类别,但数字"6"翻转就成了别的;真实猫狗照片水平翻转安全,垂直翻转就不行——增强策略要跟着数据语义走;
  3. Normalize要统一。训练集用什么均值方差,验证测试就用什么;用预训练模型时则要用它训练时的ImageNet统计量(下一节)。

工程机制上,这些变换发生在__getitem__取样瞬间,不额外占磁盘;配合DataLoader的num_workers还能并行完成。

# 1.13 迁移学习:站在ImageNet的肩膀上

现实里你几乎不可能拥有ImageNet那样120万张标注图。**迁移学习(transfer learning)**的洞察是:在ImageNet上训出的CNN,前面的层学到的是边缘、纹理、颜色等通用视觉特征,这些特征对猫狗、病灶、零件缺陷同样有用——我们只需借用这个"见多识广的骨干",在自己的小数据集上做轻量适配。

torchvision一行代码即可加载预训练模型:

from torchvision.models import resnet18, ResNet18_Weights

model = resnet18(weights=ResNet18_Weights.DEFAULT)  # 自动下载ImageNet权重
print(sum(p.numel() for p in model.parameters()))   # 约1169万参数
1
2
3
4

# 1.13.1 两种适配策略

策略一:冻结骨干,只训新分类头。 把骨干参数的requires_grad全部设为False,再把最后的1000类fc层替换成自己任务的分类层(新层默认可训练)。优点是训练极快、小数据上不易过拟合:

model = resnet18(weights=ResNet18_Weights.DEFAULT)
for param in model.parameters():
    param.requires_grad_(False)            # 冻结全部骨干
model.fc = nn.Linear(model.fc.in_features, 10)  # 替换分类头(参数可训练)
# 优化器只收编requires_grad=True的参数
optimizer = torch.optim.Adam(filter(lambda p: p.requires_grad, model.parameters()), lr=1e-3)
1
2
3
4
5
6

策略二:整体微调。 全部参数保持可训练,但用更小的学习率(预训练权重已经很好,大lr会把它打坏):

model = resnet18(weights=ResNet18_Weights.DEFAULT)
model.fc = nn.Linear(model.fc.in_features, 10)
optimizer = torch.optim.Adam(model.parameters(), lr=1e-4)  # 比从头训练小一个量级
1
2
3

还可以做分层学习率:骨干用小lr慢慢磨,新分类头用大lr快速学,通过参数组实现:

optimizer = torch.optim.Adam([
    {"params": [p for n, p in model.named_parameters() if "fc" not in n], "lr": 1e-5},
    {"params": model.fc.parameters(), "lr": 1e-3}
])
1
2
3
4

另外两个接口细节:ResNet要求224×224左右的3通道输入,灰度图要复制成3通道(Lambda(lambda t: t.repeat(3,1,1))),尺寸可先Resize((64,64))以在本地快速试验;用预训练模型时Normalize要换成ImageNet的均值方差[0.485,0.456,0.406]/[0.229,0.224,0.225]。

# 1.13.2 实测:只用1200张图,三种方案对比

从Fashion MNIST各取1200张训练、400张测试(缩放到64×64、复制成3通道、按ImageNet统计量标准化),对比从零训练、冻结骨干、整体微调三种方案:

from torchvision import datasets, transforms
from torchvision.models import resnet18, ResNet18_Weights
from torch.utils.data import DataLoader, Subset

# ---- 数据准备: 1200张训练 + 400张测试 ----
set_seed(42)  # 数据子集抽样可复现
tf = transforms.Compose([
    transforms.Resize((64, 64)),                       # 本地快速试验用小尺寸
    transforms.Grayscale(num_output_channels=3),       # 灰度复制成3通道
    transforms.ToTensor(),
    transforms.Normalize([0.485, 0.456, 0.406],        # ImageNet统计量
                         [0.229, 0.224, 0.225]),
])
full_train = datasets.FashionMNIST(root=".", train=True, download=True, transform=tf)
full_test = datasets.FashionMNIST(root=".", train=False, download=True, transform=tf)
# 随机抽固定索引的子集, 三种方案用同一份数据才公平
train_idx = torch.randperm(len(full_train))[:1200]
test_idx = torch.randperm(len(full_test))[:400]
small_train = Subset(full_train, train_idx)
small_test = Subset(full_test, test_idx)
train_loader = DataLoader(small_train, batch_size=64, shuffle=True)
test_loader = DataLoader(small_test, batch_size=64)

# ---- 三种方案统一训练函数 ----
def run_transfer(model, name, epochs=3, lr=1e-3):
    set_seed(42)                       # 同种子初始化fc层, 公平对比
    model = model.to(device)
    # 只收编requires_grad=True的参数(冻结时骨干不进优化器)
    optimizer = torch.optim.Adam(
        filter(lambda p: p.requires_grad, model.parameters()), lr=lr)
    loss_fn = nn.CrossEntropyLoss()
    for epoch in range(epochs):
        tr_loss, tr_acc = run_epoch(model, train_loader, loss_fn, optimizer)
        te_loss, te_acc = run_epoch(model, test_loader, loss_fn)
        print(f"[{name}] epoch {epoch+1}/{epochs}  "
              f"train {tr_acc:.4f}  test {te_acc:.4f}")
    return tr_acc, te_acc

# 方案一: 从零训练(不加载预训练权重)
m_scratch = resnet18(weights=None)
m_scratch.fc = nn.Linear(m_scratch.fc.in_features, 10)
run_transfer(m_scratch, "从零训练", lr=1e-3)

# 方案二: 冻结骨干, 只训新分类头
m_frozen = resnet18(weights=ResNet18_Weights.DEFAULT)
for param in m_frozen.parameters():
    param.requires_grad_(False)
m_frozen.fc = nn.Linear(m_frozen.fc.in_features, 10)   # 新层默认可训练
run_transfer(m_frozen, "冻结骨干", lr=1e-3)

# 方案三: 整体微调(全部参数可训练, 小lr保护预训练权重)
m_finetune = resnet18(weights=ResNet18_Weights.DEFAULT)
m_finetune.fc = nn.Linear(m_finetune.fc.in_features, 10)
run_transfer(m_finetune, "整体微调", lr=1e-4)
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
[从零训练] epoch 1/3  train 0.6067  test 0.6100
[从零训练] epoch 2/3  train 0.7817  test 0.7125
[从零训练] epoch 3/3  train 0.8533  test 0.7375
[冻结骨干] epoch 1/3  train 0.4775  test 0.3675
[冻结骨干] epoch 2/3  train 0.5500  test 0.4125
[冻结骨干] epoch 3/3  train 0.6142  test 0.4725
[整体微调] epoch 1/3  train 0.8125  test 0.7575
[整体微调] epoch 2/3  train 0.8892  test 0.7900
[整体微调] epoch 3/3  train 0.9417  test 0.8250
1
2
3
4
5
6
7
8
9

结论分两层看:

  • 整体微调一骑绝尘:3个epoch、1200张图就到82.5%,且每个epoch只要1秒多——这就是迁移学习的威力,实际项目里90%的图像任务都应该从这个起点开始;
  • 冻结策略这次反而最慢,但不是失效:把上面run_transfer(m_frozen, ...)的epochs改成10再跑一次,第1、3、10个epoch测试准确率分别为0.3675、0.4725、0.7800,稳步爬升,10个epoch后追平了从零训练。原因是Fashion MNIST(放大的灰度服装图)与ImageNet(自然彩色照片)域差异较大,冻结的特征需要分类头单方面迁就;在与ImageNet更接近的自然照片任务上(猫狗、花草、车型),冻结策略通常第1个epoch就能拿到极好的成绩,训练还最快。

经验选择:数据少、任务与ImageNet接近→先冻结训头,不够再微调;数据有一定量或域差异大→直接整体微调,小lr配分层学习率;任务域差异极大(如医学显微图像、X光)→可以在浅层冻结、深层放开,或考虑从零训练。

# 1.14 为什么换数据集:CIFAR-10登场

MNIST/FashionMNIST是深度学习的"hello world",但它们太简单也太特殊:灰度、居中、背景干净,LeNet就能拿下八成多。真实世界的图像是彩色的、有复杂背景的、类间差异细微的(猫和狗比T恤和裤子难分多了)。

CIFAR-10就是为此准备的标准进阶数据集:

属性 FashionMNIST CIFAR-10
图像尺寸 28×28 32×32
通道 1(灰度) 3(RGB彩色)
训练/测试 60000 / 10000 50000 / 10000
类别 T恤、裤子、包…… 飞机、汽车、鸟、猫、鹿、狗、蛙、马、船、卡车
难度 LeNet约84% LeNet明显吃力,现代CNN 85%~95%
from torchvision import datasets, transforms
from torch.utils.data import DataLoader

# CIFAR-10官方三通道均值标准差(在50000张训练图上统计得到)
mean = (0.4914, 0.4822, 0.4465)
std = (0.247, 0.243, 0.261)

train_tf = transforms.Compose([
    transforms.RandomCrop(32, padding=4),      # 四周补4像素零边再随机裁回32x32
    transforms.RandomHorizontalFlip(),         # 水平翻转
    transforms.ToTensor(),
    transforms.Normalize(mean, std)
])
test_tf = transforms.Compose([
    transforms.ToTensor(),                     # 测试集不做任何随机增强
    transforms.Normalize(mean, std)
])

train_dataset = datasets.CIFAR10("data", train=True, download=True, transform=train_tf)
test_dataset = datasets.CIFAR10("data", train=False, download=True, transform=test_tf)
train_loader = DataLoader(train_dataset, batch_size=128, shuffle=True)
test_loader = DataLoader(test_dataset, batch_size=256, shuffle=False)
print(len(train_dataset), len(test_dataset), train_dataset[0][0].shape)
print(train_dataset.classes)
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
50000 10000 torch.Size([3, 32, 32])
['airplane', 'automobile', 'bird', 'cat', 'deer', 'dog', 'frog', 'horse', 'ship', 'truck']
1
2

两个增强都是Day21讲过的标准件:RandomCrop(padding=4)制造轻微位移,HorizontalFlip制造左右镜像,等于每个epoch给模型看不同的"新图"。测试集只用确定性变换,保证评估可复现。

# 1.15 AlexNet的CIFAR改造版:大网络的第一次实战

2012年的原版AlexNet吃的是224×224图,第一层是11×11、步幅4的大卷积,末端连着两个4096维的巨型全连接层,共约6000万参数。直接照搬到32×32上,第一层大卷积一步就把图缩没了。改造原则是保留它的灵魂(ReLU、Dropout、卷积堆叠加宽、大FC分类头),适配输入尺寸:

  • 第一层改为3×3、padding=1,保住分辨率,让后面的池化逐步降尺寸;
  • 加上BatchNorm(原版用的是LRN,已被BN全面取代);
  • 卷积通道按64→192→384→256→256递增,3次2×2池化后特征图降到4×4。
import torch.nn as nn

class AlexCIFAR(nn.Module):
    def __init__(self, num_classes=10):
        super().__init__()
        self.features = nn.Sequential(
            nn.Conv2d(3, 64, 3, padding=1), nn.BatchNorm2d(64), nn.ReLU(),
            nn.MaxPool2d(2, 2),                                  # 32 -> 16
            nn.Conv2d(64, 192, 3, padding=1), nn.BatchNorm2d(192), nn.ReLU(),
            nn.MaxPool2d(2, 2),                                  # 16 -> 8
            nn.Conv2d(192, 384, 3, padding=1), nn.ReLU(),
            nn.Conv2d(384, 256, 3, padding=1), nn.ReLU(),
            nn.Conv2d(256, 256, 3, padding=1), nn.ReLU(),
            nn.MaxPool2d(2, 2),                                  # 8 -> 4
        )
        self.classifier = nn.Sequential(
            nn.Dropout(0.5),
            nn.Linear(256 * 4 * 4, 4096), nn.ReLU(),
            nn.Dropout(0.5),
            nn.Linear(4096, 4096), nn.ReLU(),
            nn.Linear(4096, num_classes),
        )

    def forward(self, x):
        x = self.features(x)
        x = torch.flatten(x, 1)
        return self.classifier(x)
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27

注意参数量分布:这个模型有3586万参数,其中绝大部分在两个4096维全连接层里——这正是Day21全局平均池化要解决的问题,先记住这个数字,看完VGG你会更有体感。

# 1.16 VGG11:用一份配置表堆出网络

VGG的精髓是"全用3×3卷积,靠深度取胜"。它的结构规整到可以用一个数字列表描述:64表示输出64通道的卷积,"M"表示池化。今天实现的是VGG-11(8个卷积层+3个全连接层,共11个带权重层)的CIFAR版:

class VGG(nn.Module):
    def __init__(self, cfg, num_classes=10):
        super().__init__()
        layers = []
        in_channels = 3
        for v in cfg:
            if v == "M":
                layers.append(nn.MaxPool2d(2, 2))
            else:
                layers += [
                    nn.Conv2d(in_channels, v, 3, padding=1, bias=False),
                    nn.BatchNorm2d(v),
                    nn.ReLU()]
                in_channels = v
        self.features = nn.Sequential(*layers)
        # CIFAR版用512->512的轻量分类头(原版ImageNet是4096->4096)
        self.classifier = nn.Sequential(
            nn.Dropout(0.5), nn.Linear(512, 512), nn.ReLU(),
            nn.Dropout(0.5), nn.Linear(512, 512), nn.ReLU(),
            nn.Linear(512, num_classes))

    def forward(self, x):
        return self.classifier(torch.flatten(self.features(x), 1))

# 11个权重层的配置: 5次池化, 32->16->8->4->2->1
cfg_vgg11 = [64, "M", 128, "M", 256, 256, "M", 512, 512, "M", 512, 512, "M"]
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26

这个配置表是VGG留给业界的遗产:换VGG-13/VGG-16/VGG-19只需换一张表,代码一行不改。卷积全部bias=False,因为后面紧跟BN(Day21的规则)。5次池化后32×32变成1×1,正好接512维分类头。参数量只有975万——不到AlexNet改造版的三分之一。

# 1.17 ResNet18:直接用torchvision,只做32×32适配

残差连接已经在Day21手写过,这次不再重复造轮子,直接加载torchvision的标准ResNet18(1117万参数)。但它是为224×224的ImageNet设计的,stem部分会把32×32缩得太狠,做两处外科手术:

from torchvision.models import resnet18

def make_resnet18_cifar():
    model = resnet18(weights=None, num_classes=10)
    # 原版: 7x7卷积stride2 + 3x3 maxpool, 32x32进stem就变8x8, 信息损失太大
    model.conv1 = nn.Conv2d(3, 64, 3, stride=1, padding=1, bias=False)
    model.maxpool = nn.Identity()      # 去掉stem后的最大池化
    return model
1
2
3
4
5
6
7
8

nn.Identity()是个"什么都不做"的占位层——输出等于输入。这样网络不再在stem处猛降分辨率,而是靠layer2/3/4里的stride2逐级把32→16→8→4,残差结构原封不动。这是社区在CIFAR上训练ResNet的标准改法。

# 1.18 三家正面对比:同样15个epoch

统一条件:相同的增强与标准化、batch size 128、固定随机种子42、各训15个epoch。优化器按各网络的"主流配方"给:AlexNet/VGG用Adam(lr=1e-3),ResNet用SGD(lr=0.1,momentum=0.9,weight_decay=5e-4)+余弦退火——配方差异本身就是工程经验(深层残差网用大初始学习率的SGD最稳)。训练循环沿用Day21的train/eval写法,每个epoch结束后在训练集(增强关闭误差可忽略)和测试集上各评一次:

import torch, time

def train_eval(model, optimizer, epochs, scheduler=None):
    for epoch in range(epochs):
        t0 = time.time()
        model.train()
        for x, y in train_loader:
            x, y = x.to(device), y.to(device)
            optimizer.zero_grad()
            loss = nn.functional.cross_entropy(model(x), y)
            loss.backward()
            optimizer.step()
        if scheduler is not None:
            scheduler.step()
        model.eval()
        correct_train = correct_test = 0
        with torch.no_grad():
            for x, y in train_loader:
                correct_train += (model(x.to(device)).argmax(1).cpu() == y).sum()
            for x, y in test_loader:
                correct_test += (model(x.to(device)).argmax(1).cpu() == y).sum()
        dt = time.time() - t0
        print(f"ep{epoch:02d} train {correct_train/50000:.4f} "
              f"test {correct_test/10000:.4f} ({dt:.1f}s)")
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24

三个网络各配一套"主流配方"后依次开训(模型类在1.15~1.17节已定义):

# 每个网络重置一次种子, 权重初始化与数据打乱顺序都公平一致
# ---- AlexNet改造版: Adam lr=1e-3 ----
torch.manual_seed(42)
alex = AlexCIFAR().to(device)
print("===== AlexNet ("
      f"{sum(p.numel() for p in alex.parameters())/1e6:.2f}M参数, Adam lr=1e-3) =====")
train_eval(alex, torch.optim.Adam(alex.parameters(), lr=1e-3), epochs=15)

# ---- VGG11: Adam lr=1e-3 ----
torch.manual_seed(42)
vgg = VGG(cfg_vgg11).to(device)
print(f"===== VGG11 ({sum(p.numel() for p in vgg.parameters())/1e6:.2f}M参数, Adam lr=1e-3) =====")
train_eval(vgg, torch.optim.Adam(vgg.parameters(), lr=1e-3), epochs=15)

# ---- ResNet18: SGD lr=0.1 + momentum + weight_decay + 余弦退火 ----
torch.manual_seed(42)
resnet = make_resnet18_cifar().to(device)
resnet_optimizer = torch.optim.SGD(resnet.parameters(), lr=0.1,
                                   momentum=0.9, weight_decay=5e-4)
# 余弦退火: lr从0.1按余弦曲线衰减到近0, T_max=15表示15个epoch走完一个周期
resnet_scheduler = torch.optim.lr_scheduler.CosineAnnealingLR(
    resnet_optimizer, T_max=15)
print(f"===== ResNet18 ({sum(p.numel() for p in resnet.parameters())/1e6:.2f}M参数, SGD lr=0.1+余弦退火) =====")
train_eval(resnet, resnet_optimizer, epochs=15, scheduler=resnet_scheduler)

# 训练结束把这份权重存盘: Day22 1.1的混淆矩阵会直接加载它
torch.save(resnet.state_dict(), "cifar_resnet18.pt")
print("已保存 cifar_resnet18.pt")
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28

以下是在Apple M4(MPS加速)上的真实结果:

===== AlexNet (35.86M参数, Adam lr=1e-3) =====
ep00 train 0.4383 test 0.4437 (77.2s)
ep01 train 0.5424 test 0.5505 (92.1s)
ep02 train 0.6011 test 0.6061 (74.1s)
ep03 train 0.6392 test 0.6333 (73.1s)
ep04 train 0.6785 test 0.6840 (77.9s)
ep05 train 0.6982 test 0.7020 (80.4s)
ep06 train 0.7234 test 0.7260 (90.3s)
ep07 train 0.7252 test 0.7210 (74.7s)
ep08 train 0.7483 test 0.7386 (79.9s)
ep09 train 0.7574 test 0.7495 (85.2s)
ep10 train 0.7533 test 0.7510 (80.6s)
ep11 train 0.7743 test 0.7673 (79.0s)
ep12 train 0.7761 test 0.7660 (75.2s)
ep13 train 0.7769 test 0.7676 (71.5s)
ep14 train 0.7918 test 0.7799 (71.8s)

===== VGG11 (9.75M参数, Adam lr=1e-3) =====
ep00 train 0.4599 test 0.4470 (78.2s)
ep01 train 0.6184 test 0.6167 (78.3s)
ep02 train 0.6720 test 0.6776 (73.8s)
ep03 train 0.7203 test 0.7154 (81.4s)
ep04 train 0.7354 test 0.7415 (78.3s)
ep05 train 0.7792 test 0.7724 (76.3s)
ep06 train 0.8003 test 0.7888 (77.2s)
ep07 train 0.8143 test 0.7954 (76.7s)
ep08 train 0.8304 test 0.8131 (77.4s)
ep09 train 0.8392 test 0.8200 (80.2s)
ep10 train 0.8434 test 0.8258 (75.8s)
ep11 train 0.8548 test 0.8287 (74.4s)
ep12 train 0.8503 test 0.8272 (76.2s)
ep13 train 0.8587 test 0.8286 (77.8s)
ep14 train 0.8880 test 0.8528 (78.2s)

===== ResNet18 (11.17M参数, SGD lr=0.1+余弦退火) =====
ep00 train 0.3167 test 0.3303 (124.3s)
ep01 train 0.4705 test 0.4665 (122.5s)
ep02 train 0.5456 test 0.5487 (126.7s)
ep03 train 0.6183 test 0.6259 (125.0s)
ep04 train 0.6175 test 0.6162 (125.3s)
ep05 train 0.7441 test 0.7424 (125.3s)
ep06 train 0.7569 test 0.7470 (129.9s)
ep07 train 0.7916 test 0.7814 (125.2s)
ep08 train 0.8337 test 0.8119 (126.0s)
ep09 train 0.8679 test 0.8445 (123.5s)
ep10 train 0.8688 test 0.8325 (122.3s)
ep11 train 0.9099 test 0.8799 (158.0s)
ep12 train 0.9273 test 0.8907 (156.5s)
ep13 train 0.9415 test 0.8983 (145.3s)
ep14 train 0.9487 test 0.9024 (134.4s)
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50

汇总对比:

网络 参数量 第1个epoch测试 15轮后训练 15轮后测试 总训练时长
AlexNet改造版 35.86M 0.4437 0.7918 0.7799 19.7分钟
VGG11(BN) 9.75M 0.4470 0.8880 0.8528 19.3分钟
ResNet18 11.17M 0.3303 0.9487 0.9024 32.8分钟

# 1.19 数字背后的四个结论

  1. 参数量不等于准确率。AlexNet参数是VGG的3.7倍(3586万对975万),测试准确率反而低7.3个百分点(0.7799对0.8528)。它多出来的2600万参数几乎全堆在4096维全连接层里,这些参数在"背诵"而不是"理解",性价比极低——VGG用3×3小卷积和轻量分类头完胜。

  2. "加深"的收益被实验证实。FashionMNIST上84%的LeNet搬到CIFAR-10会明显吃力;VGG11在15轮就到85.3%,ResNet18更是冲过90%——Day21讲的3×3堆叠(参数更少、感受野等效、非线性更多)和残差连接在这里拿到了实证。

  3. 起跑快的不一定跑得远。AlexNet第1个epoch就有44.4%的测试准确率、VGG11有44.7%,得益于巨大全连接层的拟合能力和Adam的快速起步;ResNet18第1个epoch只有33.0%(SGD初始lr大、预热慢)。但AlexNet后半程在78%附近反复震荡、明显饱和(第7轮测试准确率从0.7260回落到0.7210,第10轮训练准确率也从0.7574回落),VGG11稳步爬到85.3%,ResNet18在第10轮短暂回落到0.8325后,靠余弦退火在末段连跳(第11~14轮测试准确率0.8799→0.8907→0.8983→0.9024),最终90.24%居首。训练曲线的形状比单轮数字更重要。

  4. 残差网络更慢但更值。ResNet18每轮约131秒,约为VGG11(每轮约77秒)的1.7倍(18层、每个残差块两条路都要算),总时长32.8分钟也接近另两家的两倍。它的训练/测试差距最大(0.9487对0.9024,4.6个百分点),但这不代表过拟合失控——末段测试准确率仍在快速上升,说明是容量充足、数据还没吃完;反倒是AlexNet差距最小(0.7918对0.7799,1.2个百分点),而它的训练和测试双双停在78%,本质是容量没被充分利用的欠拟合。还要说明:ResNet的SGD配方通常要训40~60个epoch才完全展开(那套配置在CIFAR-10上可达93%以上);只给15轮时座次已是ResNet>VGG>AlexNet,继续训差距只会拉大。

# 小结

训练工程化、现代CNN和CIFAR-10实战今天一次讲透。可模型训完了,怎么知道它到底在哪类样本上栽了跟头?Day22从混淆矩阵和精确率/召回率开始,把学习率调度、混合精度、TensorBoard、Grad-CAM、自编码器、物体检测和语义分割一气讲完——评估、监控、可解释到应用。

最近更新: 2026/10/14
 |  Copyright © -Present Kelsen All Rights Reserved. | Author KelsenJing