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

Lee_jerome

1223人浏览 · 2026-07-29 23:46:47

Lee_jerome · 2026-07-29 23:46:47 发布

📌 本文属于《Python神经网络入门：零基础保姆级路线图》专栏

上一篇：python神经网络编程入门（十三）——CNN纯 NumPy 实现 LeNet-5 全流程拆解与 MNIST 前向验证（上）
下一篇：python神经网络编程入门（十四）——RNN序列数据与循环思想，为什么你的模型需要“记忆”？

完整目录 & 更新记录：
《Python神经网络入门：零基础保姆级路线图（附全系列免费源码）》

引言

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

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

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

补齐每一层的反向传播：为 Conv2D、MaxPooling、FullyConnected 实现 backward 方法。
串联反向传播链路：在 LeNet5 中实现完整的 backward 方法，让梯度从输出层一路回传到输入层。
实现参数更新：用随机梯度下降（SGD） 更新所有可训练参数。
构建完整训练循环：在 MNIST 上训练模型，实时监控损失下降与准确率提升。
可视化训练过程：用 Matplotlib 绘制损失曲线和准确率曲线，直观感受 CNN 的学习过程。

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

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

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

1.1 什么是链式法则？

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

$\hat{y} = f_L \left( f_{L-1} \left( \cdots f_1 (x) \cdots \right) \right)$

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

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

该层参数 $W^{[l]}$ 的梯度 = 上游梯度 × 该层输出对 W[l] 的偏导数
该层输入 $a^{[l-1]}$ 的梯度 = 上游梯度 × 该层输出对输入的偏导数（这个梯度要继续往回传）

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

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

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

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

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

全连接层的前向传播是：

$Z = X \cdot W + b$

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

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

对权重 W 的梯度： $dW = X^T \cdot \text{dout}$ ，形状 (D,M)
对偏置 b 的梯度： $db = \sum_{i=1}^{N} \text{dout}[i, :]$ ，形状(M,)（对 batch 维度求和）
对输入 X 的梯度： $dX = \text{dout} \cdot W^T$ ，形状 (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，因为反向传播计算 $dW$ 时需要用到它。

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）的反向传播

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

$Y[b, oc, i, j] = \sum_{c=0}^{C-1} \sum_{u=0}^{K-1} \sum_{v=0}^{K-1} X[b, c, i \cdot s + u, j \cdot s + v] \cdot W[oc, c, u, v] + b[oc]$

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

对权重 W 的梯度：dW——用于更新卷积核
对偏置 b 的梯度：db——用于更新偏置
对输入 X 的梯度：dX——传回给前一层

2.3.1 对偏置的梯度 dbdb

偏置 $b[oc]$ 会加到所有输出位置 (i,j) 上，所以：

$db[oc] = \sum_{b=0}^{N-1} \sum_{i=0}^{H_{\text{out}}-1} \sum_{j=0}^{W_{\text{out}}-1} \text{dout}[b, oc, i, j]$

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

2.3.2 对权重的梯度 dW

观察前向公式，权重 $W_{oc, c, u, v}$ 会与所有输入中对应位置的窗口相乘。具体来说：

$dW[oc, c, u, v] = \sum_{b=0}^{N-1} \sum_{i=0}^{H_{\text{out}}-1} \sum_{j=0}^{W_{\text{out}}-1} \text{dout}[b, oc, i, j] \cdot X[b, c, i \cdot s + u, j \cdot s + v]$

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

2.3.3 对输入的梯度 dX

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

更直观的理解是：卷积层的反向传播，相当于把 $\text{dout}$ 零填充后，与旋转 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）：

$W = W - \eta \cdot dW, \quad b = b - \eta \cdot db$

其中 η 是学习率（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.001	Adam 的默认推荐值，足够稳定
Batch Size	128	充分利用 CPU 的 SIMD 指令集，同时保持梯度估计的稳定性
Epoch	10	模型在第 5~6 个 epoch 后已基本收敛

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

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

Epoch	训练损失（Cross-Entropy）	测试准确率	累计耗时
1	0.3746	96.07%	3.8s
2	0.0963	97.75%	7.4s
3	0.0652	98.39%	10.9s
4	0.0508	98.57%	15.0s
5	0.0423	98.57%	18.4s
6	0.0356	98.83%	21.9s
7	0.0317	98.88%	25.3s
8	0.0267	98.77%	28.7s
9	0.0255	98.93%	32.2s
10	0.0200	98.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 函数 $\sigma(x) = \frac{1}{1 + e^{-x}}$ 的导数满足：

$\sigma'(x) = \sigma(x) \cdot (1 - \sigma(x))$

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

$\frac{\partial \mathcal{L}}{\partial W^{[1]}} = \frac{\partial \mathcal{L}}{\partial Z^{[3]}} \cdot W^{[3]} \cdot \sigma'(Z^{[3]}) \cdot W^{[2]} \cdot \sigma'(Z^{[2]}) \cdot \frac{\partial Z^{[1]}}{\partial W^{[1]}}$

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

7.2 ReLU 如何破解梯度消失？

ReLU 函数 $\text{ReLU}(x) = \max(0, x)$ 的导数为：

$\text{ReLU}'(x) = \begin{cases} 1, & x > 0 \\ 0, & x \leq 0 \end{cases}$

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

7.3 Adam 相对于 SGD 的优越性

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

$m_t = \beta_1 m_{t-1} + (1 - \beta_1) g_t$ （一阶矩）

$v_t = \beta_2 v_{t-1} + (1 - \beta_2) g_t^2$ （二阶矩）

$\theta_{t+1} = \theta_t - \eta \cdot \frac{m_t}{\sqrt{v_t} + \epsilon}$

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

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

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

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

模型	参数量	训练数据量	10 Epoch 耗时	测试准确率
MLP（单隐层）	~200k	60k	极慢（纯 NumPy）	~93%
LeNet-5（NumPy 原型）	~61k	1k	213s	未收敛（梯度消失）
LeNet-5（PyTorch CPU）	~61k	60k	35.88s	98.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(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])