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

引言在上一篇十三·上中我们完成了 LeNet-5 的前向传播组装与验证——从 32×3232×32 的输入图像一路流到 1010 个类别的概率输出并计算了随机初始化状态下的交叉熵损失作为基线。但一个只能做前向推理、不能学习的网络本质上只是一堆随机数字的排列组合。今天我们要做的就是让这个网络真正“学起来”。具体来说本文将完成以下工作补齐每一层的反向传播为Conv2D、MaxPooling、FullyConnected实现backward方法。串联反向传播链路在LeNet5中实现完整的backward方法让梯度从输出层一路回传到输入层。实现参数更新用随机梯度下降SGD更新所有可训练参数。构建完整训练循环在 MNIST 上训练模型实时监控损失下降与准确率提升。可视化训练过程用 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, axis0) self.db np.sum(dout, axis0) # 对输入的梯度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_size2, stride2): 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*sp, j*s:j*sp] # 找到最大值的位置展平后取 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 的位置就是最大值位置 # 但这里更直接用前向记录的位置来分配梯度 # 我们用另一种方式在对应窗口内找到 mask1 的位置 window_mask self.mask[b, c, i*s:i*sp, j*s:j*sp] # 在窗口内找到值为 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的反向传播卷积层的反向传播是最复杂的部分。我们先回顾前向传播反向传播需要计算三个梯度对权重 W 的梯度dW——用于更新卷积核对偏置 b 的梯度db——用于更新偏置对输入 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, stride1, padding0): # ... 与之前一致 ... 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)), modeconstant) 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*sk, j*s:j*sk] # (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)), modeconstant) else: dout_pad dout # 将卷积核旋转 180° weights_rot np.rot90(self.weights, k2, 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:ik, j:jk] # (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 之后的梯度 # 输出层fc3dout 直接传入 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*sk, j*s:j*sk] out[b, oc, i, j] np.sum(window * weights[oc]) bias[oc]每一对(i, j)坐标都要执行一次 Python 层面的切片、乘法和求和操作。对于 6 万张图这意味着几十亿次 Python 解释器指令CPU 大部分时间耗在解释型语言的循环开销和动态类型检查上而非真正的数值计算。5.2 PyTorch 的三大杀招维度NumPy纯 PythonPyTorchC/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.8s20.096397.75%7.4s30.065298.39%10.9s40.050898.57%15.0s50.042398.57%18.4s60.035698.83%21.9s70.031798.88%25.3s80.026798.77%28.7s90.025598.93%32.2s100.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∣ 较大时例如 x3σ(x) 趋近于 1此时导数趋近于 1×00。在反向传播中梯度是连乘的LeNet-5 有 5 层可训练参数每经过一层 Sigmoid梯度就乘以一个小于 1 的数通常在 0.1~0.25 左右。连乘 5 次后梯度变得极小趋近于 1e-5 甚至更小导致浅层Conv1、Conv2的权重几乎不更新。你的损失卡在 2.3037 不变正是梯度过小、模型“停止生长”的典型表现。7.2 ReLU 如何破解梯度消失ReLU 函数的导数为在正半轴x0导数恒为 1。这意味着梯度在 ReLU 层传播时不会被缩放浅层网络可以接收到来自深层的完整梯度信号彻底解决了“梯度消失”问题。7.3 Adam 相对于 SGD 的优越性SGD 使用全局统一的学习率需要精细调节如本次若用 SGD ReLU可能需要调整到 lr0.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-5NumPy 原型~61k1k213s未收敛梯度消失LeNet-5PyTorch CPU~61k60k35.88s98.78%结论结构决定上限CNN 的局部连接和权值共享使其在图像任务上天生优于全连接网络。工程决定下限再好的算法如果用低效的工程实现如纯 NumPy 循环在有限时间内根本无法收敛。PyTorch 将算法从“理论玩具”变成了“生产力工具”。九、总结与展望本文基于你的两次实际实验完成了一次完整的“诊断-优化-验证”闭环诊断Sigmoid 导致梯度消失SGD 在深层网络上力不从心。优化切换至 ReLU 激活 Adam 优化器并采用 PyTorch 后端加速。验证仅用 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(fUsing device: CPU, cores: {os.cpu_count()}) # -------------------- 2. 数据集类不变-------------------- class MNISTCSVDataset(Dataset): def __init__(self, csv_path, limitNone, pad_to32): 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)), modeconstant) self.images images def __len__(self): return len(self.labels) def __getitem__(self, idx): return torch.tensor(self.images[idx], dtypetorch.float32), torch.tensor(self.labels[idx], dtypetorch.long) # -------------------- 3. LeNet-5ReLU Adam 版-------------------- class LeNet5(nn.Module): def __init__(self): super(LeNet5, self).__init__() self.conv1 nn.Conv2d(1, 6, kernel_size5, stride1, padding0) self.pool1 nn.MaxPool2d(kernel_size2, stride2) self.conv2 nn.Conv2d(6, 16, kernel_size5, stride1, padding0) self.pool2 nn.MaxPool2d(kernel_size2, stride2) 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_dim1) 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, limitNone, pad_to32) test_dataset MNISTCSVDataset(mnist_test.csv, limitNone, pad_to32) train_loader DataLoader(train_dataset, batch_size128, shuffleTrue, num_workers0) test_loader DataLoader(test_dataset, batch_size128, shuffleFalse, num_workers0) print(fTrain: {len(train_dataset)}, Test: {len(test_dataset)}) # -------------------- 5. 初始化模型、损失、优化器 -------------------- model LeNet5().to(device) criterion nn.CrossEntropyLoss() optimizer optim.Adam(model.parameters(), lr0.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(fEpoch {epoch1}/{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])