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

📅 2026/7/30 2:01:13 👁️ 阅读次数 📝 编程学习
python神经网络编程入门(十三)——CNN纯 NumPy 实现LeNet-5 反向传播串联与 MNIST 完整训练实战(下)

引言

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

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

具体来说,本文将完成以下工作:

  1. 补齐每一层的反向传播:为Conv2DMaxPoolingFullyConnected实现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.outself.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(纯 Python)PyTorch(C++/CUDA 后端)
循环方式Python 显式for循环底层 C++ 模板元编程 + 高度优化的for循环
数学库OpenBLAS(通用)Intel MKL / cuDNN(硬件指令集特化)
卷积实现直接滑动窗口im2col+ GEMM(矩阵乘)或 Winograd 算法
自动微分需手写反向传播(极易出错)autograd引擎自动构建计算图
多核利用依赖 NumPy 的有限并行主动绑定 CPU 核心(OMP_NUM_THREADS

实测算力对比

  • 纯 NumPy 版本:1000 张图 → 213 秒(有效算力 ≈ 4.7 张/秒)

  • PyTorch CPU 版本:60000 张图 → 35.88 秒(有效算力 ≈1672 张/秒

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

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

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

6.1 关键超参数说明

超参数取值设计考量
激活函数ReLU避免 Sigmoid 在深层网络中的梯度饱和现象
优化器Adam自适应学习率,对初始学习率不敏感,收敛速度快
学习率0.001Adam 的默认推荐值,足够稳定
Batch Size128充分利用 CPU 的 SIMD 指令集,同时保持梯度估计的稳定性
Epoch10模型在第 5~6 个 epoch 后已基本收敛

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

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

Epoch训练损失(Cross-Entropy)测试准确率累计耗时
10.374696.07%3.8s
20.096397.75%7.4s
30.065298.39%10.9s
40.050898.57%15.0s
50.042398.57%18.4s
60.035698.83%21.9s
70.031798.88%25.3s
80.026798.77%28.7s
90.025598.93%32.2s
100.020098.78%35.9s


数据分析

  • 第 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%的准确率。

模型参数量训练数据量10 Epoch 耗时测试准确率
MLP(单隐层)~200k60k极慢(纯 NumPy)~93%
LeNet-5(NumPy 原型)~61k1k213s未收敛(梯度消失)
LeNet-5(PyTorch CPU)~61k60k35.88s98.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])