案例 10：从零实现二维卷积、后向传播与 LeNet#
学习目标与记号#
本案例把二维互相关的前向计算、卷积核梯度、输入梯度和 LeNet 训练放在同一个可以逐步检查的过程中，主要讨论以下内容：
区分深度学习框架使用的互相关与严格数学卷积；
根据输入、卷积核、填充和步幅计算输出维度；
用 NumPy 实现多通道二维互相关的前向传播和后向传播；
用有限差分检查输入、核和偏置梯度；
在 Fashion-MNIST 上比较展平线性基线与 LeNet；
结合参数量和容易混淆的类别解释卷积神经网络（convolutional neural network，CNN）的作用，而不是只看准确率。
记批量大小、输入通道数和输出通道数分别为 \(N\)、\(C_{\mathrm{in}}\) 和 \(C_{\mathrm{out}}\)，输入图像的高和宽分别为 \(H\) 和 \(W\)，卷积核的高和宽分别为 \(K_h\) 和 \(K_w\)。
数据与下载#
Fashion-MNIST 包含 60,000 张训练图像和 10,000 张官方测试图像，每张图像都是 \(28\times28\) 的灰度服饰图像，共有 10 个类别。
数据说明、许可和 MD5 校验值：https://github.com/zalandoresearch/fashion-mnist；
许可：MIT；
本案例使用该项目列出的四个匿名 HTTPS 文件，并把它们缓存到课程源码目录之外。
官方测试集只用于最终评价；官方训练文件再按类别比例划分为训练集和验证集。这里的像素归一化只把 0--255 的像素值缩放到固定范围，不需要根据全部数据估计均值或方差，因此不会提前使用测试集信息。
from __future__ import annotations
import gzip
import os
import random
import struct
import urllib.request
from pathlib import Path
import numpy as np
import pandas as pd
import matplotlib.pyplot as plt
from sklearn.metrics import accuracy_score, confusion_matrix, f1_score
from sklearn.model_selection import train_test_split
SEED = 20260812
FAST_MODE = os.getenv("FAST_MODE", "1") != "0"
random.seed(SEED)
np.random.seed(SEED)
try:
import torch
from torch import nn
from torch.utils.data import DataLoader, TensorDataset
except ImportError as exc:
raise ImportError("本案例需要 PyTorch。") from exc
torch.manual_seed(SEED)
DEVICE = torch.device("cuda" if torch.cuda.is_available() else "cpu")
CACHE_ROOT = Path(os.getenv("AI_COURSE_DATA_DIR", Path.home() / ".cache" / "ai-course"))
CACHE_DIR = CACHE_ROOT / "fashion_mnist"
CACHE_DIR.mkdir(parents=True, exist_ok=True)
print({"快速模式": FAST_MODE, "计算设备": str(DEVICE), "缓存目录": str(CACHE_DIR)})
{'快速模式': True, '计算设备': 'cpu', '缓存目录': '/private/tmp/ai-course-case-data/fashion_mnist'}
BASE = "https://storage.googleapis.com/tensorflow/tf-keras-datasets/"
FILES = {
"train-images-idx3-ubyte.gz": BASE + "train-images-idx3-ubyte.gz",
"train-labels-idx1-ubyte.gz": BASE + "train-labels-idx1-ubyte.gz",
"t10k-images-idx3-ubyte.gz": BASE + "t10k-images-idx3-ubyte.gz",
"t10k-labels-idx1-ubyte.gz": BASE + "t10k-labels-idx1-ubyte.gz",
}
for name, url in FILES.items():
path = CACHE_DIR / name
if not path.exists():
urllib.request.urlretrieve(url, path)
assert path.stat().st_size > 1_000
def read_idx(path: Path):
with gzip.open(path, "rb") as handle:
zero, dtype_code, ndim = struct.unpack(">HBB", handle.read(4))
assert zero == 0 and dtype_code == 8
shape = struct.unpack(">" + "I" * ndim, handle.read(4 * ndim))
data = np.frombuffer(handle.read(), dtype=np.uint8)
assert data.size == int(np.prod(shape))
return data.reshape(shape)
X_full = read_idx(CACHE_DIR / "train-images-idx3-ubyte.gz")
y_full = read_idx(CACHE_DIR / "train-labels-idx1-ubyte.gz")
X_test = read_idx(CACHE_DIR / "t10k-images-idx3-ubyte.gz")
y_test = read_idx(CACHE_DIR / "t10k-labels-idx1-ubyte.gz")
assert X_full.shape == (60_000, 28, 28) and X_test.shape == (10_000, 28, 28)
assert y_full.shape == (60_000,) and set(np.unique(y_full)) == set(range(10))
print({"训练图像维度": X_full.shape, "训练标签维度": y_full.shape, "测试图像维度": X_test.shape, "测试标签维度": y_test.shape})
{'训练图像维度': (60000, 28, 28), '训练标签维度': (60000,), '测试图像维度': (10000, 28, 28), '测试标签维度': (10000,)}
LABELS = ["T恤/上衣", "裤子", "套衫", "连衣裙", "外套", "凉鞋", "衬衫", "运动鞋", "包", "短靴"]
fig, axes = plt.subplots(2, 5, figsize=(10, 4))
for cls, ax in enumerate(axes.flat):
idx = np.flatnonzero(y_full == cls)[0]
ax.imshow(X_full[idx], cmap="gray")
ax.set_title(f"{cls}: {LABELS[cls]}")
ax.axis("off")
plt.tight_layout()
class_counts = np.bincount(y_full, minlength=10)
print(pd.Series(class_counts, index=LABELS, name="训练文件类别数"))
assert class_counts.min() == class_counts.max() # 原始训练文件类别均衡
T恤/上衣 6000
裤子 6000
套衫 6000
连衣裙 6000
外套 6000
凉鞋 6000
衬衫 6000
运动鞋 6000
包 6000
短靴 6000
Name: 训练文件类别数, dtype: int64
train_idx, valid_idx = train_test_split(
np.arange(len(y_full)),
test_size=10_000,
random_state=SEED,
stratify=y_full,
)
if FAST_MODE:
# 从已定义的训练部分分层抽样；验证与官方测试也取固定分层子集，以缩短 CPU 演示。
train_idx, _ = train_test_split(
train_idx, train_size=8_000, random_state=SEED, stratify=y_full[train_idx]
)
valid_idx, _ = train_test_split(
valid_idx, train_size=2_000, random_state=SEED, stratify=y_full[valid_idx]
)
test_idx, _ = train_test_split(
np.arange(len(y_test)), train_size=2_000, random_state=SEED, stratify=y_test
)
else:
test_idx = np.arange(len(y_test))
X_train = X_full[train_idx].astype(np.float32) / 255.0
y_train = y_full[train_idx].astype(np.int64)
X_valid = X_full[valid_idx].astype(np.float32) / 255.0
y_valid = y_full[valid_idx].astype(np.int64)
X_eval = X_test[test_idx].astype(np.float32) / 255.0
y_eval = y_test[test_idx].astype(np.int64)
assert set(train_idx).isdisjoint(set(valid_idx))
assert X_train.ndim == 3 and X_train.shape[1:] == (28, 28)
print({"训练集": X_train.shape, "验证集": X_valid.shape, "测试集": X_eval.shape})
{'训练集': (8000, 28, 28), '验证集': (2000, 28, 28), '测试集': (2000, 28, 28)}
二维互相关的输出维度与参数共享#
对输入 \(\boldsymbol{X}\in\mathbb{R}^{N\times C_{\mathrm{in}}\times H\times W}\) 和卷积核 \(\boldsymbol{K}\in\mathbb{R}^{C_{\mathrm{out}}\times C_{\mathrm{in}}\times K_h\times K_w}\)，若两个空间方向的填充均为 \(P\)、步幅均为 \(S\)，则输出图像的高和宽分别为
深度学习框架通常不会把卷积核翻转，因此实际计算是互相关。一个输出通道在所有空间位置重复使用同一组卷积核参数，这体现了局部连接和参数共享。下面的 NumPy 实现只用于很小的数组和梯度检查；训练实际模型时使用经过优化的 PyTorch 运算。
def conv2d_forward(x, w, b, stride=1, padding=0):
x = np.asarray(x, dtype=np.float64)
w = np.asarray(w, dtype=np.float64)
b = np.asarray(b, dtype=np.float64)
assert x.ndim == 4 and w.ndim == 4 and b.shape == (w.shape[0],)
n, c_in, h, width = x.shape
c_out, c_w, kh, kw = w.shape
assert c_in == c_w
h_out = (h + 2 * padding - kh) // stride + 1
w_out = (width + 2 * padding - kw) // stride + 1
assert h_out > 0 and w_out > 0
xp = np.pad(x, ((0, 0), (0, 0), (padding, padding), (padding, padding)))
out = np.empty((n, c_out, h_out, w_out), dtype=np.float64)
for i in range(h_out):
for j in range(w_out):
patch = xp[:, :, i*stride:i*stride+kh, j*stride:j*stride+kw]
out[:, :, i, j] = np.einsum("nchw,ochw->no", patch, w) + b
return out
rng = np.random.default_rng(SEED)
x_probe = rng.normal(size=(2, 2, 5, 6))
w_probe = rng.normal(size=(3, 2, 3, 2))
b_probe = rng.normal(size=3)
z_probe = conv2d_forward(x_probe, w_probe, b_probe, stride=2, padding=1)
assert z_probe.shape == (2, 3, 3, 4)
print("输出维度：", z_probe.shape)
输出维度： (2, 3, 3, 4)
def conv2d_backward(dout, x, w, stride=1, padding=0):
x = np.asarray(x, dtype=np.float64)
w = np.asarray(w, dtype=np.float64)
dout = np.asarray(dout, dtype=np.float64)
n, c_in, h, width = x.shape
c_out, _, kh, kw = w.shape
xp = np.pad(x, ((0, 0), (0, 0), (padding, padding), (padding, padding)))
dxp = np.zeros_like(xp)
dw = np.zeros_like(w)
db = dout.sum(axis=(0, 2, 3))
for i in range(dout.shape[2]):
for j in range(dout.shape[3]):
hs, ws = i * stride, j * stride
patch = xp[:, :, hs:hs+kh, ws:ws+kw]
dw += np.einsum("no,nchw->ochw", dout[:, :, i, j], patch)
dxp[:, :, hs:hs+kh, ws:ws+kw] += np.einsum(
"no,ochw->nchw", dout[:, :, i, j], w
)
dx = dxp[:, :, padding:padding+h, padding:padding+width] if padding else dxp
assert dx.shape == x.shape and dw.shape == w.shape
return dx, dw, db
dout = rng.normal(size=z_probe.shape)
dx, dw, db = conv2d_backward(dout, x_probe, w_probe, stride=2, padding=1)
print({"输入梯度维度": dx.shape, "卷积核梯度维度": dw.shape, "偏置梯度维度": db.shape})
{'输入梯度维度': (2, 2, 5, 6), '卷积核梯度维度': (3, 2, 3, 2), '偏置梯度维度': (3,)}
# 有限差分检查标量目标 L=sum(conv(x,w,b)*dout) 的若干坐标。
def scalar_loss(x, w, b):
return float(np.sum(conv2d_forward(x, w, b, stride=2, padding=1) * dout))
def finite_difference(array, index, loss_fn, eps=1e-6):
old = array[index]
array[index] = old + eps
plus = loss_fn()
array[index] = old - eps
minus = loss_fn()
array[index] = old
return (plus - minus) / (2 * eps)
checks = []
for name, array, grad, indices in [
("x", x_probe, dx, [(0,0,0,0), (1,1,4,5)]),
("w", w_probe, dw, [(0,0,0,0), (2,1,2,1)]),
("b", b_probe, db, [(0,), (2,)]),
]:
for index in indices:
numerical = finite_difference(
array, index, lambda: scalar_loss(x_probe, w_probe, b_probe)
)
analytical = grad[index]
rel_error = abs(numerical - analytical) / max(1.0, abs(numerical), abs(analytical))
checks.append((name, index, analytical, numerical, rel_error))
check_table = pd.DataFrame(checks, columns=["variable", "index", "analytical", "numerical", "relative_error"])
check_display = check_table.rename(columns={
"variable": "变量", "index": "索引", "analytical": "解析梯度",
"numerical": "数值梯度", "relative_error": "相对误差",
})
check_display["变量"] = check_display["变量"].map({"x": "输入", "w": "卷积核", "b": "偏置"})
display(check_display)
assert check_table["relative_error"].max() < 1e-6
| 变量 | 索引 | 解析梯度 | 数值梯度 | 相对误差 | |
|---|---|---|---|---|---|
| 0 | 输入 | (0, 0, 0, 0) | 1.888034 | 1.888034 | 6.682904e-11 |
| 1 | 输入 | (1, 1, 4, 5) | 1.922231 | 1.922231 | 1.254125e-09 |
| 2 | 卷积核 | (0, 0, 0, 0) | -2.752577 | -2.752577 | 1.290833e-10 |
| 3 | 卷积核 | (2, 1, 2, 1) | -1.776828 | -1.776828 | 7.763989e-10 |
| 4 | 偏置 | (0,) | -0.027033 | -0.027033 | 4.672517e-09 |
| 5 | 偏置 | (2,) | 7.035649 | 7.035649 | 1.475482e-10 |
def make_loader(X, y, batch_size, shuffle):
ds = TensorDataset(
torch.from_numpy(X[:, None, :, :]),
torch.from_numpy(y),
)
generator = torch.Generator().manual_seed(SEED)
return DataLoader(ds, batch_size=batch_size, shuffle=shuffle, generator=generator if shuffle else None)
train_loader = make_loader(X_train, y_train, 128, True)
valid_loader = make_loader(X_valid, y_valid, 256, False)
test_loader = make_loader(X_eval, y_eval, 256, False)
class LinearPixels(nn.Module):
def __init__(self):
super().__init__()
self.net = nn.Sequential(nn.Flatten(), nn.Linear(28 * 28, 10))
def forward(self, x):
return self.net(x)
class LeNet(nn.Module):
def __init__(self):
super().__init__()
self.features = nn.Sequential(
nn.Conv2d(1, 6, 5, padding=2), nn.Tanh(), nn.AvgPool2d(2),
nn.Conv2d(6, 16, 5), nn.Tanh(), nn.AvgPool2d(2),
)
self.classifier = nn.Sequential(
nn.Flatten(), nn.Linear(16 * 5 * 5, 120), nn.Tanh(),
nn.Linear(120, 84), nn.Tanh(), nn.Linear(84, 10),
)
def forward(self, x):
z = self.features(x)
assert z.shape[1:] == (16, 5, 5)
return self.classifier(z)
assert LinearPixels()(torch.zeros(2,1,28,28)).shape == (2,10)
assert LeNet()(torch.zeros(2,1,28,28)).shape == (2,10)
def fit(model, epochs):
torch.manual_seed(SEED)
model = model.to(DEVICE)
optimizer = torch.optim.Adam(model.parameters(), lr=1e-3)
history = []
for epoch in range(epochs):
model.train()
losses = []
for xb, yb in train_loader:
xb, yb = xb.to(DEVICE), yb.to(DEVICE)
optimizer.zero_grad()
logits = model(xb)
loss = nn.functional.cross_entropy(logits, yb)
assert logits.shape == (len(xb), 10) and torch.isfinite(loss)
loss.backward()
optimizer.step()
losses.append(loss.item())
model.eval()
correct = total = 0
with torch.no_grad():
for xb, yb in valid_loader:
pred = model(xb.to(DEVICE)).argmax(1).cpu()
correct += int((pred == yb).sum())
total += len(yb)
history.append((np.mean(losses), correct / total))
return model, np.asarray(history)
epochs = 2 if FAST_MODE else 8
trained, histories = {}, {}
model_labels = {"linear": "线性像素分类器", "lenet": "LeNet 卷积网络"}
for name, factory in {"linear": LinearPixels, "lenet": LeNet}.items():
trained[name], histories[name] = fit(factory(), epochs)
print(model_labels[name], "最后一轮训练损失/验证准确率：", histories[name][-1])
线性像素分类器 最后一轮训练损失/验证准确率： [0.79186671 0.763 ]
LeNet 卷积网络 最后一轮训练损失/验证准确率： [0.78103431 0.7465 ]
def evaluate(model):
model.eval()
ys, preds = [], []
with torch.no_grad():
for xb, yb in test_loader:
logits = model(xb.to(DEVICE))
ys.append(yb.numpy())
preds.append(logits.argmax(1).cpu().numpy())
y_true, y_pred = np.concatenate(ys), np.concatenate(preds)
return y_true, y_pred
rows, predictions = [], {}
for name, model in trained.items():
yt, yp = evaluate(model)
predictions[name] = yp
rows.append({
"model": name,
"parameters": sum(p.numel() for p in model.parameters()),
"accuracy": accuracy_score(yt, yp),
"macro_f1": f1_score(yt, yp, average="macro"),
})
results = pd.DataFrame(rows).set_index("model")
results_display = results.rename(
index=model_labels,
columns={"parameters": "参数量", "accuracy": "准确率", "macro_f1": "宏平均 F1"},
)
results_display.index.name = "模型"
display(results_display)
assert np.isfinite(results.to_numpy()).all()
cm = confusion_matrix(y_eval, predictions["lenet"])
per_class_recall = cm.diagonal() / cm.sum(axis=1)
print(pd.Series(per_class_recall, index=LABELS, name="LeNet 各类别召回率").sort_values())
print("容易混淆的类别反映图像外观相似性；参数量更大或准确率更高都不能单独证明架构更好。")
| 参数量 | 准确率 | 宏平均 F1 | |
|---|---|---|---|
| 模型 | |||
| 线性像素分类器 | 7850 | 0.7485 | 0.742128 |
| LeNet 卷积网络 | 61706 | 0.7080 | 0.691887 |
衬衫 0.080
套衫 0.550
连衣裙 0.680
外套 0.685
T恤/上衣 0.720
运动鞋 0.780
凉鞋 0.790
裤子 0.920
包 0.930
短靴 0.945
Name: LeNet 各类别召回率, dtype: float64
容易混淆的类别反映图像外观相似性；参数量更大或准确率更高都不能单独证明架构更好。
逐步理解本案例#
1. 从图像文件到四维张量#
Fashion-MNIST 中每幅图像都是 \(28\times28\) 的灰度图。读入像素后先除以 255，把数值缩放到固定范围。单幅灰度图仍需保留通道轴，因此批量输入的维度为 \((批量,1,28,28)\)，四个轴依次表示样本、通道、高和宽。观测标签向量的维度为 \((批量,)\)，每个整数表示一个服饰类别。若误把高度轴当作通道轴，卷积运算有时仍可执行，但参数含义已经改变，所以程序会在图像进入网络前检查维度。
从官方训练数据中再划分验证集，测试集不参与网络结构、停止轮数和学习率的选择。抽样、参数初始化、批次顺序和数值梯度抽查均使用确定的随机种子。固定随机种子是复现实验的基础，但不能保证不同计算设备得到逐位相同的结果，因此正式比较还应使用多个随机种子重复实验。
2. 手工二维互相关在计算什么#
对输入 \(\boldsymbol{X}\)、卷积核 \(\boldsymbol{K}\) 和偏置 \(\boldsymbol{b}\)，输出位置 \((i,j)\) 等于局部输入窗口与卷积核逐元素相乘后求和，再加上相应偏置。输入、卷积核和输出的维度分别为 \((N,C_{\mathrm{in}},H,W)\)、\((C_{\mathrm{out}},C_{\mathrm{in}},K_h,K_w)\) 和 \((N,C_{\mathrm{out}},H_{\mathrm{out}},W_{\mathrm{out}})\)。输出的高和宽由输入尺寸、填充、步幅和卷积核大小共同决定。先用小张量手算，再与框架结果比较，可以把滑动窗口的过程转化为能够逐项核对的数值运算。
多通道卷积先在每个输入通道上计算局部乘积，再对所有输入通道求和；每个输出通道有自己的一组卷积核。一个输出通道的偏置会加到该通道的所有空间位置。这里的广播方式由模型结构明确规定，并不表示任意维度的数组都可以相加。
3. 为什么要分别检查三类后向梯度#
后续计算传回的梯度 \(\mathrm{d}\boldsymbol{Y}\) 到达卷积层后，卷积核梯度 \(\mathrm{d}\boldsymbol{K}\) 要累加所有样本和空间位置对同一共享参数的贡献；输入梯度 \(\mathrm{d}\boldsymbol{X}\) 要把每个输出位置的影响加回对应的输入窗口；偏置梯度 \(\mathrm{d}\boldsymbol{b}\) 则要对批量轴和两个空间轴求和。三者的维度必须分别与 \(\boldsymbol{K}\)、\(\boldsymbol{X}\) 和 \(\boldsymbol{b}\) 一致。遗漏批量求和、错误处理卷积核方向或填充区域，都可能得到维度正确但数值错误的梯度。
中心差分在某个参数位置使用 \([\mathcal{J}(\theta+\epsilon)-\mathcal{J}(\theta-\epsilon)]/(2\epsilon)\) 近似真实梯度。\(\epsilon\) 太大会使近似不够精细，太小又容易受到浮点舍入误差影响，因此本案例使用 64 位浮点小张量，并比较相对误差。数值梯度检查只适合抽查少量位置，不能替代完整的模型训练。
4. 池化与 LeNet 怎样改变数据表示#
卷积层保留局部空间关系，激活函数引入非线性，池化降低空间分辨率，并扩大后续神经元相对于原图的有效感受野。程序在每一层之后打印或检查张量维度，直到将卷积特征展开并送入用于输出类别的全连接层。模型参数通常有很大一部分位于展开后的全连接层；改变输入尺寸时，如果仍手工写死展开后的特征数，程序就会报错或建立错误的网络结构。
LeNet 是用于教学的基础模型，并不代表现代图像分类中的最佳结构。它帮助我们理解经过手工检查的卷积运算怎样与框架层组成完整分类器。训练时，交叉熵函数直接接收输出层的线性运算结果，不需要先手工计算 Softmax；框架会在内部采用更稳定的计算方法。
5. 怎样避免只看总体准确率#
Fashion-MNIST 的各类别样本数大致均衡，但衬衫、套衫和外套在视觉上较为相似。除总体准确率外，还应查看每个类别的召回率和混淆矩阵。某一类别召回率较低时，可以先检查它最常被误判成哪些类别，再查看相应图像和预测概率。测试集上的错误分析只能用于说明模型局限，不能据此反复调整模型后，仍把同一测试集结果当作独立的最终评价。
可以从多个方面核对结果：手工互相关应与框架输出一致，公式得到的梯度应与数值梯度接近，复杂模型应与简单分类基线比较，还应同时观察训练曲线、验证曲线和最终分类指标。这些检查分别针对运算、梯度、参数更新和模型在新样本上的表现，不能相互替代。
6. 常见错误与使用范围#
当步幅大于 1 或卷积核大小为偶数时，same 填充在两侧可能不同，不能简单理解为每侧都填 \(K//2\) 个位置。最大池化的后向传播只把梯度传回前向窗口中取得最大值的位置；若有多个位置取得相同最大值，应遵循框架规定的处理方式。对于彩色图像，输入通道数从 1 变为 3，但类别数不会因此改变。
发布时，源 Notebook 不保存运行输出，数据写入独立缓存目录。图题、坐标轴、类别名称、表格列名和运行提示均使用中文；Python 接口、变量名和 LeNet 名称保留原文。
本案例小结与局限#
Fashion-MNIST 是居中、灰度、低分辨率商品图，不能代表自然图像的尺度、背景和光照变化。
快速模式只训练少量轮次和样本，目的是验证管线；不应用其分数宣称模型达到稳定性能。
NumPy 卷积实现用于清楚展示计算过程，不支持分组卷积、空洞卷积等完整功能，也不适合大规模训练；
有限差分只抽查若干坐标；工程测试还应覆盖不同步幅、填充、核大小和数据类型。
官方测试集只评价一次。类别图像和错误样本的人工查看也应避免反复据此调参。
LeNet 的局部归纳偏置适合图像，但它不是所有视觉问题的最佳模型。
发布前检查#
运行后应确认训练集、验证集和测试集的索引互不重叠，卷积输出维度与公式一致，数值梯度的相对误差小于预先规定的允许范围。静态网页只保存图表和解释，不随网页分发原始数据或训练得到的模型参数。
综合练习#
扩展 NumPy 实现，使填充可分别指定高度和宽度，并为非法输出尺寸抛错。
对池化层手工推导输入梯度；分别实现平均池化和最大池化的后向传播。
保持训练预算不变，把 LeNet 的 tanh 改成 ReLU，报告激活分布、梯度范数和结果。
构造一张平移 1 像素的图，比较卷积特征的平移等变性与分类输出的近似不变性。
对每一类展示最多 3 个错误样本，描述最常见混淆，但不要把单张图解释为总体规律。