python神经网络编程入门（十三）——CNN纯 NumPy 实现LeNet-5 反向传播串联与 MNIST 完整训练实战（下）

引言

在上一篇（十三·上）中，我们完成了 LeNet-5 的前向传播组装与验证——从 32×3232×32 的输入图像一路流到 1010 个类别的概率输出，并计算了随机初始化状态下的交叉熵损失作为基线。

但一个只能做前向推理、不能学习的网络，本质上只是一堆随机数字的排列组合。今天我们要做的，就是让这个网络真正“学起来”。

具体来说，本文将完成以下工作：

1. 补齐每一层的反向传播：为Conv2D、MaxPooling、FullyConnected实现backward方法。

2. 串联反向传播链路：在LeNet5中实现完整的backward方法，让梯度从输出层一路回传到输入层。

3. 实现参数更新：用随机梯度下降（SGD）更新所有可训练参数。

4. 构建完整训练循环：在 MNIST 上训练模型，实时监控损失下降与准确率提升。

5. 可视化训练过程：用 Matplotlib 绘制损失曲线和准确率曲线，直观感受 CNN 的学习过程。

数据一致性声明：本文使用的 MNIST 数据、加载方式、预处理（零填充到 32×3232×32）与上一篇完全一致，代码可直接拼接运行。

一、反向传播的理论基础：链式法则

在动手写代码之前，我们先回顾一个核心数学工具——链式法则（Chain Rule）。

1.1 什么是链式法则？

神经网络本质上是一个复合函数：输入 x 经过层层变换，最终得到输出。如果用 f1,f2,...,fL表示每一层的变换，那么整个网络就是：

训练网络的目标是最小化损失函数L(,y)。为此，我们需要知道损失函数对每一个可训练参数（权重 W 和偏置 b）的偏导数，然后沿着负梯度方向更新参数。

链式法则告诉我们：后一层的梯度可以传递给前一层。具体来说，如果损失 L 对某一层输出 a[l]的梯度已知，那么：

• 该层参数的梯度 = 上游梯度 × 该层输出对 W[l] 的偏导数

• 该层输入的梯度 = 上游梯度 × 该层输出对输入的偏导数（这个梯度要继续往回传）

用一句话概括：反向传播就是“从后往前，一层一层地计算梯度，并把梯度回传给前一层”。

1.2 三大模块的反向传播策略

层类型	反向传播的核心任务	关键要点
全连接层	计算 dW、db、dx	矩阵乘法的逆运算：
池化层（MaxPooling）	将梯度分配给前向传播时被选中的那个神经元	需要记录前向时最大值的位置（掩码），梯度独享给该位置
卷积层	计算 dW、db、dx	本质上是卷积操作的逆过程，需要对梯度做“反卷积”式的填充与滑动

二、各层反向传播的代码实现

2.1 全连接层（FullyConnected）的反向传播

全连接层的前向传播是：

其中 X形状为 (N,D)，W 形状为(D,M)，b 形状为(M,)。

假设上游传来的梯度是（形状 (N,M)），即损失对 Z 的偏导数。那么：

• 对权重 W 的梯度：，形状 (D,M)

• 对偏置 b 的梯度：，形状(M,)（对 batch 维度求和）

• 对输入 X 的梯度：，形状 (N,D)（要传回给前一层）

class FullyConnected: # ... __init__ 和 forward 与之前保持一致 ... def backward(self, dout): """ 反向传播 dout: 来自上一层的梯度，形状 (batch, out_size) 返回: 传给前一层的梯度 dx，形状 (batch, in_size) """ # 对权重的梯度：dW = X^T @ dout self.dW = np.dot(self.x.T, dout) # 对偏置的梯度：db = sum(dout, axis=0) self.db = np.sum(dout, axis=0) # 对输入的梯度：dx = dout @ W^T dx = np.dot(dout, self.weights.T) return dx

⚠️注意：在forward中我们需要保存输入self.x，因为反向传播计算时需要用到它。

2.2 最大池化层（MaxPooling）的反向传播

最大池化层在前向传播时，每个2×2 窗口只输出了最大值，其他三个值被丢弃了。

反向传播时，梯度只能传给前向时被选中的那个最大值的位置，其他位置的梯度为 0。

因此，我们需要在前向传播时记录每个窗口最大值的位置（称为“掩码”mask）。

class MaxPooling: def __init__(self, pool_size=2, stride=2): self.pool_size = pool_size self.stride = stride self.mask = None # 用于记录最大值的位置 def forward(self, x): self.x = x # 保存输入，反向传播时需要 batch, channels, h, w = x.shape p, s = self.pool_size, self.stride out_h = (h - p) // s + 1 out_w = (w - p) // s + 1 out = np.zeros((batch, channels, out_h, out_w)) self.mask = np.zeros_like(x) # 掩码与输入同形状 for b in range(batch): for c in range(channels): for i in range(out_h): for j in range(out_w): window = x[b, c, i*s:i*s+p, j*s:j*s+p] # 找到最大值的位置（展平后取 argmax，再还原为二维坐标） flat_idx = np.argmax(window) max_h = flat_idx // p max_w = flat_idx % p # 记录最大值的位置（在掩码中置 1） self.mask[b, c, i*s + max_h, j*s + max_w] = 1 out[b, c, i, j] = window[max_h, max_w] return out def backward(self, dout): """ 反向传播：将梯度分配给最大值的位置 dout: 来自上一层的梯度，形状 (batch, channels, out_h, out_w) 返回: 传给前一层的梯度 dx，形状 (batch, channels, h, w) """ # 先将 dout 上采样（放大）到与输入相同的尺寸 batch, channels, out_h, out_w = dout.shape p, s = self.pool_size, self.stride h, w = self.x.shape[2], self.x.shape[3] dx = np.zeros_like(self.x) for b in range(batch): for c in range(channels): for i in range(out_h): for j in range(out_w): # 找到前向时最大值的位置 # mask 中值为 1 的位置就是最大值位置 # 但这里更直接：用前向记录的位置来分配梯度 # 我们用另一种方式：在对应窗口内找到 mask=1 的位置 window_mask = self.mask[b, c, i*s:i*s+p, j*s:j*s+p] # 在窗口内找到值为 1 的位置（即最大值位置） pos = np.where(window_mask == 1) if len(pos[0]) > 0: max_h, max_w = pos[0][0], pos[1][0] dx[b, c, i*s + max_h, j*s + max_w] = dout[b, c, i, j] return dx

📝原理说明：self.mask是一个与输入x同形状的矩阵，在最大值位置记为 1，其余为 0。反向传播时，我们只需把dout中的每个值放到mask中对应最大值的位置即可。

2.3 卷积层（Conv2D）的反向传播

卷积层的反向传播是最复杂的部分。我们先回顾前向传播：

反向传播需要计算三个梯度：

1. 对权重 W 的梯度：dW——用于更新卷积核

2. 对偏置 b 的梯度：db——用于更新偏置

3. 对输入 X 的梯度：dX——传回给前一层

2.3.1 对偏置的梯度 dbdb

偏置会加到所有输出位置 (i,j) 上，所以：

即对 batch、高度、宽度三个维度求和。

2.3.2 对权重的梯度 dW

观察前向公式，权重会与所有输入中对应位置的窗口相乘。具体来说：

这个计算本质上就是：把当作卷积核，在输入 X 上做“卷积”。

2.3.3 对输入的梯度 dX

这是最 tricky 的部分。前向时，输入 X 中的一个像素可能会被多个卷积窗口覆盖到（当步长小于卷积核大小时）。反向传播时，这些位置的梯度需要累加。

更直观的理解是：卷积层的反向传播，相当于把零填充后，与旋转 180° 的卷积核做卷积。

具体实现时，我们采用窗口遍历的方式，与 forward 保持对称：

class Conv2D: def __init__(self, in_channels, out_channels, kernel_size, stride=1, padding=0): # ... 与之前一致 ... self.x = None # 保存输入，用于反向传播 def forward(self, x): self.x = x # 保存输入 # ... 与之前一致 ... return out def backward(self, dout): """ 反向传播 dout: 来自上一层的梯度，形状 (batch, out_channels, out_h, out_w) 返回: 传给前一层的梯度 dx，形状 (batch, in_channels, h, w) """ batch, _, h, w = self.x.shape _, _, out_h, out_w = dout.shape k, s, p = self.kernel_size, self.stride, self.padding # 1. 计算对偏置的梯度 db self.db = np.sum(dout, axis=(0, 2, 3)) # 对 batch、高、宽求和 # 2. 计算对权重的梯度 dW self.dW = np.zeros_like(self.weights) # 如果有填充，先对输入做填充（与前向一致） if p > 0: x_pad = np.pad(self.x, ((0,0), (0,0), (p,p), (p,p)), mode='constant') else: x_pad = self.x for b in range(batch): for oc in range(self.out_channels): for i in range(out_h): for j in range(out_w): # 提取输入窗口 window = x_pad[b, :, i*s:i*s+k, j*s:j*s+k] # (in_channels, k, k) # dout[b, oc, i, j] 是一个标量 self.dW[oc] += window * dout[b, oc, i, j] # 3. 计算对输入的梯度 dx（要传回前一层） # 先对 dout 做零填充，使得卷积后的尺寸与输入一致 # 对于无填充的情况：需要在 dout 四周填充 (k-1) 圈 0 # 对于有填充的情况：填充量为 (k-1 - p) pad_dout = (k - 1) - p if pad_dout > 0: dout_pad = np.pad(dout, ((0,0), (0,0), (pad_dout, pad_dout), (pad_dout, pad_dout)), mode='constant') else: dout_pad = dout # 将卷积核旋转 180° weights_rot = np.rot90(self.weights, k=2, axes=(2, 3)) # 在每个 (k,k) 窗口内旋转 # 交换输入通道和输出通道的维度：原形状 (out_channels, in_channels, k, k) # 旋转后需要交换前两维，变成 (in_channels, out_channels, k, k) weights_rot = np.transpose(weights_rot, (1, 0, 2, 3)) # 现在用 dout_pad 和 weights_rot 做卷积 dx = np.zeros_like(self.x) batch, in_c, h, w = self.x.shape out_h_dx = (h + 2*0 - k) // 1 + 1 # 这里步长固定为 1 for b in range(batch): for ic in range(in_c): for i in range(h): for j in range(w): # 提取 dout_pad 中的窗口 window = dout_pad[b, :, i:i+k, j:j+k] # (out_channels, k, k) dx[b, ic, i, j] = np.sum(window * weights_rot[ic]) return dx

⚠️上述实现为了可读性采用了显式循环，在实际训练中效率较低。但作为教学实现，它清晰地展示了卷积反向传播的每一步。你可以后续用im2col等技术优化。

三、串联 LeNet-5 的完整反向传播

现在我们把各层的backward串联起来。在LeNet5类中添加backward方法：

class LeNet5: # ... __init__ 和 forward 与之前一致 ... def backward(self, dout): """ 反向传播：从输出层开始，逐层回传梯度 dout: 损失函数对输出层（Softmax 之后）的梯度 """ # 输出层（fc3）：dout 直接传入 dout = self.fc3.backward(dout) dout = dout * sigmoid_derivative(self.fc2.out) # 穿过 Sigmoid # FC2 dout = self.fc2.backward(dout) dout = dout * sigmoid_derivative(self.fc1.out) # FC1 dout = self.fc1.backward(dout) # 展平层的反向：将 (batch, 400) 重塑为 (batch, 16, 5, 5) dout = dout.reshape(self.pool2.out.shape) # Pool2 dout = self.pool2.backward(dout) # Conv2 -> Sigmoid dout = dout * sigmoid_derivative(self.conv2.out) dout = self.conv2.backward(dout) # Pool1 dout = self.pool1.backward(dout) # Conv1 -> Sigmoid dout = dout * sigmoid_derivative(self.conv1.out) dout = self.conv1.backward(dout) # dout 此时是损失对输入的梯度，但我们不需要再往前传了

这里需要一个sigmoid_derivative函数：

def sigmoid_derivative(x): """Sigmoid 函数的导数：σ'(x) = σ(x) * (1 - σ(x))""" return x * (1 - x)

📝注意：在forward中，我们需要保存每一层的输出（如self.conv1.out、self.fc1.out等），以便在反向传播时计算激活函数的导数。

四、参数更新：随机梯度下降（SGD）

有了梯度，我们就可以更新参数了。最简单的优化算法是随机梯度下降（SGD）：

其中 η 是学习率（Learning Rate）。

我们为每个可训练层添加update方法：

class Conv2D: # ... def update(self, lr): self.weights -= lr * self.dW self.bias -= lr * self.db class FullyConnected: # ... def update(self, lr): self.weights -= lr * self.dW self.bias -= lr * self.db

在LeNet5中统一调用：

class LeNet5: # ... def update(self, lr): self.conv1.update(lr) self.conv2.update(lr) self.fc1.update(lr) self.fc2.update(lr) self.fc3.update(lr)

五、开始训练：从 NumPy 原型到 PyTorch 工程化：为什么非换不可？

在纯 NumPy 实现中，我们用 1000 张图跑 3 个 epoch 就耗时 213 秒。而在本次 PyTorch 实验中，6 万张图跑 10 个 epoch 仅耗时 35.88 秒。性能差距如此悬殊，根本原因在于底层计算范式的本质区别。

5.1 NumPy 的致命瓶颈：Python 层的显式循环

NumPy 虽然底层用 C 实现了向量化运算，但卷积操作无法通过简单的向量化表达。我们在 NumPy 版本中必须写四重循环：

for b in range(batch): for oc in range(out_channels): for i in range(out_h): for j in range(out_w): window = x_pad[b, :, i*s:i*s+k, j*s:j*s+k] out[b, oc, i, j] = np.sum(window * weights[oc]) + bias[oc]

每一对(i, j)坐标都要执行一次 Python 层面的切片、乘法和求和操作。对于 6 万张图，这意味着几十亿次 Python 解释器指令，CPU 大部分时间耗在解释型语言的循环开销和动态类型检查上，而非真正的数值计算。

5.2 PyTorch 的三大杀招

实测算力对比：

• 纯 NumPy 版本：1000 张图 → 213 秒（有效算力 ≈ 4.7 张/秒）

• PyTorch CPU 版本：60000 张图 → 35.88 秒（有效算力 ≈1672 张/秒）

性能提升355 倍。这正是工业界用 PyTorch/TensorFlow，而非 NumPy 搭建生产级模型的根本原因。

六、完整训练循环：超参数配置与实时监控

我们最终运行的LeNet5模型采用ReLU + Adam组合，具体训练配置如下：

6.1 关键超参数说明

6.2 训练过程日志解读（基于你的实际输出）

下面是你在实验中记录的关键数据：

数据分析：

• 第 1 个 epoch：损失从随机的 2.30 骤降至 0.37，准确率直接跳到 96%。这说明 ReLU 的梯度在初始阶段非常通畅，模型迅速捕捉到了 MNIST 的核心特征（边缘、轮廓）。

• 第 3~5 个 epoch：准确率突破 98%，损失下降趋缓，表明模型已经进入“精调”阶段。

• 第 10 个 epoch：最佳准确率达到98.93%（第 9 个 epoch），最终稳定在 98.78%。相比我们最初设想的 97%~98%，实际表现更优，这也验证了 Adam 优化器的强大。

6.3 可视化训练曲线（代码及解读）

曲线解读：

• 左图（损失）：呈现标准的“陡降-缓降”模式。第 1 个 epoch 下降最剧烈（2.30 → 0.37），随后平滑趋近于 0。没有震荡，说明学习率设置合理。

• 右图（准确率）：几乎是一条向上倾斜的直线。值得注意的是，第 4 个 epoch 之后准确率稳定在 98.5% 以上，波动极小（±0.2%），表明模型已经充分泛化，没有明显的过拟合迹象。

七、为什么 Sigmoid 会“死掉”，而 ReLU + Adam 能“救活”？

结合你第一次实验（Sigmoid + SGD）的失败结果，我们可以从数学上找到根本原因。

7.1 Sigmoid 的梯度饱和

Sigmoid 函数​ 的导数满足：

当 ∣x∣ 较大时（例如 x>3），σ(x) 趋近于 1，此时导数趋近于 1×0=0。在反向传播中，梯度是连乘的：

LeNet-5 有 5 层可训练参数，每经过一层 Sigmoid，梯度就乘以一个小于 1 的数（通常在 0.1~0.25 左右）。连乘 5 次后，梯度变得极小（趋近于 1e-5 甚至更小），导致浅层（Conv1、Conv2）的权重几乎不更新。你的损失卡在 2.3037 不变，正是梯度过小、模型“停止生长”的典型表现。

7.2 ReLU 如何破解梯度消失？

ReLU 函数的导数为：

在正半轴（x>0），导数恒为 1。这意味着梯度在 ReLU 层传播时不会被缩放，浅层网络可以接收到来自深层的完整梯度信号，彻底解决了“梯度消失”问题。

7.3 Adam 相对于 SGD 的优越性

SGD 使用全局统一的学习率，需要精细调节（如本次若用 SGD + ReLU，可能需要调整到 lr=0.001）。而Adam（自适应矩估计）为每个参数维护独立的学习率：

（一阶矩）

（二阶矩）

简单说，更新步长大的参数会自动降低学习率，更新步长小的参数会自动提升学习率。这种自适应性使得 Adam 在大部分 CV 任务上“开箱即用”，省去了大量的调参时间。

八、CNN 与 MLP 的实战对比（基于你的实测数据）

我们在第七篇中用纯 NumPy 实现的 MLP（单隐藏层 256 个神经元）在 MNIST 上仅能达到约92%~93%的准确率（且训练 10 个 epoch 耗时极长）。

而你现在用 LeNet-5（仅 6 万参数）在35.88 秒内就达到了98.78%的准确率。

结论：

1. 结构决定上限：CNN 的局部连接和权值共享，使其在图像任务上天生优于全连接网络。

2. 工程决定下限：再好的算法，如果用低效的工程实现（如纯 NumPy 循环），在有限时间内根本无法收敛。PyTorch 将算法从“理论玩具”变成了“生产力工具”。

九、总结与展望

本文基于你的两次实际实验，完成了一次完整的“诊断-优化-验证”闭环：

1. 诊断：Sigmoid 导致梯度消失，SGD 在深层网络上力不从心。

2. 优化：切换至 ReLU 激活 + Adam 优化器，并采用 PyTorch 后端加速。

3. 验证：仅用 35.88 秒，在 6 万张 MNIST 图片上达到了98.78%的准确率。

优化后完整代码：

import numpy as np import pandas as pd import torch import torch.nn as nn import torch.optim as optim from torch.utils.data import Dataset, DataLoader import time import os # -------------------- 1. CPU 优化设置 -------------------- os.environ["OMP_NUM_THREADS"] = str(os.cpu_count()) os.environ["MKL_NUM_THREADS"] = str(os.cpu_count()) torch.set_num_threads(os.cpu_count()) device = torch.device("cpu") print(f"Using device: CPU, cores: {os.cpu_count()}") # -------------------- 2. 数据集类（不变）-------------------- class MNISTCSVDataset(Dataset): def __init__(self, csv_path, limit=None, pad_to=32): data = pd.read_csv(csv_path).values if limit: data = data[:limit] self.labels = data[:, 0].astype(np.int64) images = data[:, 1:].astype(np.float32) images = images / 255.0 images = images.reshape(-1, 1, 28, 28) if pad_to > 28: pad_size = (pad_to - 28) // 2 images = np.pad(images, ((0,0), (0,0), (pad_size, pad_size), (pad_size, pad_size)), mode='constant') self.images = images def __len__(self): return len(self.labels) def __getitem__(self, idx): return torch.tensor(self.images[idx], dtype=torch.float32), torch.tensor(self.labels[idx], dtype=torch.long) # -------------------- 3. LeNet-5（ReLU + Adam 版）-------------------- class LeNet5(nn.Module): def __init__(self): super(LeNet5, self).__init__() self.conv1 = nn.Conv2d(1, 6, kernel_size=5, stride=1, padding=0) self.pool1 = nn.MaxPool2d(kernel_size=2, stride=2) self.conv2 = nn.Conv2d(6, 16, kernel_size=5, stride=1, padding=0) self.pool2 = nn.MaxPool2d(kernel_size=2, stride=2) self.fc1 = nn.Linear(16 * 5 * 5, 120) self.fc2 = nn.Linear(120, 84) self.fc3 = nn.Linear(84, 10) self.relu = nn.ReLU() def forward(self, x): x = self.relu(self.conv1(x)) x = self.pool1(x) x = self.relu(self.conv2(x)) x = self.pool2(x) x = torch.flatten(x, start_dim=1) x = self.relu(self.fc1(x)) x = self.relu(self.fc2(x)) x = self.fc3(x) # 没有激活，CrossEntropyLoss 自带 Softmax return x # -------------------- 4. 加载数据 -------------------- print("Loading data...") train_dataset = MNISTCSVDataset('mnist_train.csv', limit=None, pad_to=32) test_dataset = MNISTCSVDataset('mnist_test.csv', limit=None, pad_to=32) train_loader = DataLoader(train_dataset, batch_size=128, shuffle=True, num_workers=0) test_loader = DataLoader(test_dataset, batch_size=128, shuffle=False, num_workers=0) print(f"Train: {len(train_dataset)}, Test: {len(test_dataset)}") # -------------------- 5. 初始化模型、损失、优化器 -------------------- model = LeNet5().to(device) criterion = nn.CrossEntropyLoss() optimizer = optim.Adam(model.parameters(), lr=0.001) # 关键改动 epochs = 10 train_losses = [] test_accuracies = [] print("Training with ReLU + Adam...") start = time.time() for epoch in range(epochs): model.train() epoch_loss = 0.0 for images, labels in train_loader: images, labels = images.to(device), labels.to(device) optimizer.zero_grad() outputs = model(images) loss = criterion(outputs, labels) loss.backward() optimizer.step() epoch_loss += loss.item() avg_loss = epoch_loss / len(train_loader) train_losses.append(avg_loss) # 测试 model.eval() correct = 0 total = 0 with torch.no_grad(): for images, labels in test_loader: images, labels = images.to(device), labels.to(device) outputs = model(images) _, predicted = torch.max(outputs, 1) total += labels.size(0) correct += (predicted == labels).sum().item() test_acc = correct / total test_accuracies.append(test_acc) print(f"Epoch {epoch+1}/{epochs}, Loss: {avg_loss:.6f}, Test Acc: {test_acc:.4f}, Time: {time.time()-start:.1f}s") print(f"\nDone in {time.time()-start:.2f}s") print("Train Losses:", [f"{l:.4f}" for l in train_losses]) print("Test Accuracies:", [f"{a:.4f}" for a in test_accuracies])