CNN.ipynb

{
 "cells": [
  {
   "cell_type": "markdown",
   "id": "5d07440c-1127-4114-b604-8d8799c41633",
   "metadata": {},
   "source": [
    "# 卷积神经网络\n",
    "\n",
    "卷积神经网络（Convolutional Neural Network，简称 CNN）是一类专门用于处理具有网格结构数据的深度学习模型，最典型的应用场景是图像处理与计算机视觉领域。CNN 通过模拟人类视觉皮层的工作方式，自动从原始数据中提取有用特征，从而实现分类、检测、分割等任务。\n",
    "\n",
    "CNN 的核心组成部分包括卷积层、池化层和全连接层，通常按以下方式堆叠组成网络：\n",
    "* 卷积层（Convolutional Layer）：\n",
    "卷积层是 CNN 的核心，通过若干可学习的卷积核（滤波器）在输入数据上滑动，进行局部感受野的特征提取。每个卷积核可以检测到输入中的某种局部特征（如边缘、纹理等）。卷积操作能够保留输入的空间结构信息，并且参数共享大幅减少了模型参数数量。\n",
    "\n",
    "* 激活函数（Activation Function）：\n",
    "在卷积层之后通常会跟随非线性激活函数，如 ReLU（Rectified Linear Unit），它能够引入非线性特性，帮助网络学习更复杂的模式。\n",
    "\n",
    "* 池化层（Pooling Layer）：\n",
    "池化层用于对卷积层提取的特征进行下采样，减少特征图的尺寸和计算量，同时增强特征的平移不变性。常见的池化方法有最大池化（Max Pooling）和平均池化（Average Pooling）。\n",
    "\n",
    "* 全连接层（Fully Connected Layer）：\n",
    "在网络的后端，通常使用一到多个全连接层将前面提取的高层次特征映射到最终的输出空间，例如分类的各个类别。全连接层类似于传统的神经网络层。\n",
    "\n",
    "CNN中新增了 Convolution 层和 Pooling 层。 CNN的层的连接顺序是“Convolution - ReLU - (Pooling)”（Pooling 层有时会被省略）。这可以理解为之前的“Affine - ReLU” 连接被替换成了“Convolution ReLU - (Pooling)”连接。"
   ]
  },
  {
   "cell_type": "markdown",
   "id": "0918da71-4f26-4700-a83c-a81d0909f416",
   "metadata": {},
   "source": [
    "## 卷积层\n",
    "* **卷积核**（Kernel）/ **滤波器**（Filter）：卷积层使用的小型权重矩阵，用于在输入数据上滑动并提取局部特征。每个卷积核可以看作一个特征探测器，用于识别特定的模式（如边缘、纹理等）。\n",
    "* **步幅**（Stride）：卷积核在输入数据上滑动的步长。步幅决定了卷积操作的移动速度，影响输出特征图的尺寸。较大的步幅会导致输出尺寸减小。\n",
    "* **填充**（Padding）：在输入数据的边界周围添加额外的像素（通常填充为零），以控制输出特征图的空间尺寸和保留边缘信息。常见的填充方式有“same”填充（输出尺寸与输入相同）和“valid”填充（不填充，使输出尺寸缩小）。\n",
    "* **感受野**（Receptive Field）：指卷积神经网络中某个神经元能“看到”的输入区域大小。感受野随着网络层数的增加而增大，决定了神经元能够捕捉的上下文范围。\n",
    "* **通道**（Channel）：输入或输出特征图的深度维度。比如，彩色图像有3个通道（RGB）。卷积核的深度通常与输入通道数相同，以便在所有通道上进行卷积运算。\n",
    "* **特征图**（Feature Map）：卷积操作后得到的输出，即激活图。特征图反映了输入数据中被卷积核检测到的特征和模式。"
   ]
  },
  {
   "cell_type": "markdown",
   "id": "05142862-fe00-481b-b5da-43636f91d740",
   "metadata": {},
   "source": [
    "在卷积操作中，输出特征图的空间尺寸（高度和宽度）由输入尺寸、滤波器尺寸、填充和步幅共同决定。给定：\n",
    "* 输入尺寸：$H \\times W$\n",
    "* 滤波器尺寸：$F_H \\times F_W$\n",
    "* 填充（每边添加的像素数）：$P$\n",
    "* 步幅：$S$\n",
    "\n",
    "输出尺寸 $(O_H, O_W)$ 的计算公式为：\n",
    "\n",
    "$$\n",
    "O_H = \\left\\lfloor \\frac{H + 2P - F_H}{S} \\right\\rfloor + 1\n",
    "$$\n",
    "\n",
    "$$\n",
    "O_W = \\left\\lfloor \\frac{W + 2P - F_W}{S} \\right\\rfloor + 1\n",
    "$$\n"
   ]
  },
  {
   "cell_type": "markdown",
   "id": "2ee78a93-da15-45ea-82cf-d7ac6895cef4",
   "metadata": {},
   "source": [
    "公式推导说明：\n",
    "1. 填充：填充 P 表示在输入的高度和宽度两边各添加 $P$ 行和 $P$ 列，从而有效输入尺寸变为 $H + 2P$ 和 $W + 2P$。\n",
    "2. 有效滑动次数：滤波器在填充后的输入上滑动。滤波器尺寸为 $F_H \\times F_W$，每移动一步高度或宽度增加 S，直到滤波器刚好能放置在输入边界内。\n",
    "    * 在高度方向，可移动次数 $= \\dfrac{(H + 2P) - F_H}{S} + 1$；\n",
    "    * 在宽度方向，可移动次数 $= \\dfrac{(W + 2P) - F_W}{S} + 1$。\n",
    "3. 取整：由于滑动次数通常要求为整数，所以需要对上述结果取下整（floor），即取最大的整数值保证滤波器完全位于输入范围内。\n"
   ]
  },
  {
   "cell_type": "markdown",
   "id": "097fe143-9940-4549-9c74-b83d12ffd98f",
   "metadata": {},
   "source": [
    "示例说明：\n",
    "\n",
    "假设：\n",
    "* 输入尺寸 $H = 32, W = 32$\n",
    "* 滤波器尺寸 $F_H = 5, F_W = 5$\n",
    "* 填充 $P = 2$\n",
    "* 步幅 $S = 1$\n",
    "\n",
    "则计算输出尺寸：\n",
    "\n",
    "$$\n",
    "O_H = \\left\\lfloor \\frac{32 + 2 \\times 2 - 5}{1} \\right\\rfloor + 1 = \\left\\lfloor \\frac{32 + 4 - 5}{1} \\right\\rfloor + 1 = \\left\\lfloor 31 \\right\\rfloor + 1 = 31 + 1 = 32\n",
    "$$\n",
    "\n",
    "$$\n",
    "O_W = \\left\\lfloor \\frac{32 + 2 \\times 2 - 5}{1} \\right\\rfloor + 1 = 32\n",
    "$$"
   ]
  },
  {
   "cell_type": "markdown",
   "id": "726aae11-b047-4214-9986-294bfd73cc69",
   "metadata": {},
   "source": [
    "## im2col\n",
    "\n",
    "im2col 是一种常用的图像处理函数，特别是在卷积神经网络中进行高效卷积计算时。它的作用是将输入图像的局部块（如卷积核大小的子区域）展开为列向量。通过这种展开，卷积运算可以转化为矩阵乘法，从而利用 BLAS 库的优化。\n",
    "\n",
    "下面是一个相对通用的 im2col 伪代码，假设我们有以下输入：\n",
    "* Input: 形状 (N, C, H, W)\n",
    "* N 表示批大小（batch size）\n",
    "* C 表示通道数（channels）\n",
    "* H, W 表示图像的高和宽\n",
    "* Kernel: 卷积核大小 (kH, kW)\n",
    "* stride: 步幅\n",
    "* padding: 填充数量（四周填充同样大小）\n"
   ]
  },
  {
   "cell_type": "markdown",
   "id": "a6164095-f7f8-4365-9065-b78e0b992e42",
   "metadata": {},
   "source": [
    "function IM2COL(input, kernel_h, kernel_w, stride, pad):\n",
    "    # 1. 获取输入大小 (N, C, H, W)\n",
    "    N, C, H, W = input.shape\n",
    "\n",
    "    # 2. 计算输出的高度和宽度\n",
    "    out_h = (H + 2*pad - kernel_h) // stride + 1\n",
    "    out_w = (W + 2*pad - kernel_w) // stride + 1\n",
    "\n",
    "    # 3. 对输入进行 zero-padding（如果 pad > 0）\n",
    "    #    创建一个新的 padded_input，形状 (N, C, H + 2*pad, W + 2*pad)\n",
    "    padded_input = zero_pad(input, pad)\n",
    "\n",
    "    # 4. 创建用于存放 im2col 结果的空数组\n",
    "    #    每一个卷积核感受野的大小是 (C * kernel_h * kernel_w),\n",
    "    #    每个输出位置对应一列，共 out_h * out_w 列。\n",
    "    col = zeros((N, C * kernel_h * kernel_w, out_h * out_w))\n",
    "\n",
    "    # 5. 双层循环, 遍历 kernel 的每个位置 (dy, dx)\n",
    "    idx = 0\n",
    "    for dy in range(kernel_h):\n",
    "        for dx in range(kernel_w):\n",
    "            # 当前 (dy, dx) 位置能滑动到多少个有效输出位置:\n",
    "            #   纵向范围: [dy, dy + stride*out_h : stride]\n",
    "            #   横向范围: [dx, dx + stride*out_w : stride]\n",
    "\n",
    "            # (N, C, out_h, out_w)\n",
    "            patch = padded_input[:, \n",
    "                                 :,\n",
    "                                 dy : dy + stride*out_h : stride, \n",
    "                                 dx : dx + stride*out_w : stride]\n",
    "\n",
    "            # 将 patch 形状从 (N, C, out_h, out_w) 拉直为 (N, C*out_h*out_w)\n",
    "            patch = reshape(patch, (N, C, -1))  # -1 会自动计算\n",
    "\n",
    "            # 将 patch 放到 col 的第 idx 行 (对应 kernel_h*kernel_w 中的第 idx 个)\n",
    "            col[:, idx*C : (idx+1)*C, :] = patch  # 注意这里要适配索引\n",
    "\n",
    "            idx += 1\n",
    "\n",
    "    return col"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 10,
   "id": "eb1e4fe8-4fe1-4cae-821c-a12a431440bd",
   "metadata": {},
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "im2col 的结果 col_x.shape: (1, 4, 9)\n",
      "[[[ 0.  1.  2.  4.  5.  6.  8.  9. 10.]\n",
      "  [ 1.  2.  3.  5.  6.  7.  9. 10. 11.]\n",
      "  [ 4.  5.  6.  8.  9. 10. 12. 13. 14.]\n",
      "  [ 5.  6.  7.  9. 10. 11. 13. 14. 15.]]]\n",
      "\n",
      "卷积核 w:\n",
      " [[[[1. 2.]\n",
      "   [3. 4.]]]]\n",
      "卷积偏置 b: [1.]\n",
      "\n",
      "卷积乘法后 conv_out.shape: (1, 1, 3, 3)\n",
      "[[ 35.  45.  55.]\n",
      " [ 75.  85.  95.]\n",
      " [115. 125. 135.]]\n"
     ]
    }
   ],
   "source": [
    "import numpy as np\n",
    "\n",
    "def zero_pad(x, pad):\n",
    "    \"\"\"\n",
    "    对输入 x (N, C, H, W) 在 H 和 W 维度做零填充。\n",
    "    pad 为整型时，表示在高和宽的两边各 pad。\n",
    "    如果 pad=0，则直接返回 x。\n",
    "    \"\"\"\n",
    "    if pad == 0:\n",
    "        return x\n",
    "    return np.pad(x, ((0, 0), (0, 0), (pad, pad), (pad, pad)), mode='constant')\n",
    "\n",
    "def im2col(input_data, kernel_h, kernel_w, stride=1, pad=0):\n",
    "    \"\"\"\n",
    "    input_data: 形状 (N, C, H, W)\n",
    "    kernel_h, kernel_w: 卷积核高宽\n",
    "    stride: 步幅\n",
    "    pad: 上下左右的零填充数量\n",
    "\n",
    "    返回值 col: 形状 (N, C*kernel_h*kernel_w, out_h*out_w)\n",
    "    \"\"\"\n",
    "    N, C, H, W = input_data.shape\n",
    "\n",
    "    # 计算输出的高度和宽度\n",
    "    out_h = (H + 2*pad - kernel_h) // stride + 1\n",
    "    out_w = (W + 2*pad - kernel_w) // stride + 1\n",
    "\n",
    "    # 对输入进行 padding\n",
    "    padded_input = zero_pad(input_data, pad)\n",
    "\n",
    "    # 分配 im2col 输出内存\n",
    "    # 行数 = C*kernel_h*kernel_w；列数 = out_h*out_w\n",
    "    col = np.zeros((N, C * kernel_h * kernel_w, out_h * out_w), dtype=input_data.dtype)\n",
    "\n",
    "    out_col_idx = 0\n",
    "    # 遍历卷积核的每个 (dy, dx) 位置\n",
    "    for dy in range(kernel_h):\n",
    "        for dx in range(kernel_w):\n",
    "            # 提取在 padded_input 上的采样结果\n",
    "            # shape: (N, C, out_h, out_w)\n",
    "            patch = padded_input[:,\n",
    "                                 :,\n",
    "                                 dy : dy + stride*out_h : stride,\n",
    "                                 dx : dx + stride*out_w : stride]\n",
    "            \n",
    "            # (N, C, out_h, out_w) -> (N, C, out_h*out_w)\n",
    "            patch = patch.reshape(N, C, -1)\n",
    "\n",
    "            # 存到 col 对应的位置\n",
    "            # 注意：对每个 dy, dx，需要在「kernel_h*kernel_w 维度」上占据一块\n",
    "            col[:, out_col_idx*C : (out_col_idx+1)*C, :] = patch\n",
    "            \n",
    "            out_col_idx += 1\n",
    "\n",
    "    return col\n",
    "\n",
    "\n",
    "# ==================\n",
    "#   测试示例\n",
    "# ==================\n",
    "# ================ 测试代码 ================\n",
    "if __name__ == \"__main__\":\n",
    "    # 1. 构造一个简单输入\n",
    "    x = np.arange(16).reshape(1, 1, 4, 4).astype(np.float32)\n",
    "    # x: shape (1,1,4,4)\n",
    "    # x 的内容:\n",
    "    # [[ 0,  1,  2,  3],\n",
    "    #  [ 4,  5,  6,  7],\n",
    "    #  [ 8,  9, 10, 11],\n",
    "    #  [12,13, 14, 15]]\n",
    "\n",
    "    # 2. 构造一个 2x2 的卷积核 (out_channels=1, in_channels=1, kH=2, kW=2)\n",
    "    #   这里随便给一些值，做个演示\n",
    "    w = np.array([[[[1, 2],\n",
    "                    [3, 4]]]], dtype=np.float32)\n",
    "    # w.shape = (1, 1, 2, 2)\n",
    "    \n",
    "    # 如果有偏置 b (维度 = out_channels=1), 例如\n",
    "    b = np.array([1.0], dtype=np.float32)  # 假设有一个偏置\n",
    "\n",
    "    # 3. 先做 im2col\n",
    "    col_x = im2col(x, kernel_h=2, kernel_w=2, stride=1, pad=0)\n",
    "    # col_x: shape (N=1, C*kH*kW=4, out_h*out_w=9)\n",
    "    # 也即 (1, 4, 9)\n",
    "\n",
    "    # 4. 将卷积核 w 拉直, 形状 = (out_channels, in_channels*kH*kW) = (1, 4)\n",
    "    #   因为 out_channels=1, in_channels=1, kH=2, kW=2 => 1*4 = 4\n",
    "    w_flat = w.reshape(1, -1)  # (1,4)\n",
    "\n",
    "    # 5. 对 im2col 的结果进行乘法\n",
    "    #    由于 col_x.shape = (N, 4, 9)，为了和 (1,4) 做矩阵乘法，需把 batch 维度 N=1 放在外面处理\n",
    "    #    对于每个 batch 条目，都可以做:   out = w_flat(1,4) @ col_x(4,9) = (1,9)\n",
    "    #    再加上偏置 b\n",
    "    #    最后 reshape 为 (1, 1, out_h, out_w) = (1,1,3,3)\n",
    "    \n",
    "    N, _, out_HW = col_x.shape  # N=1, _=4, out_HW=9\n",
    "    out_h = out_w = int(np.sqrt(out_HW))  # = 3\n",
    "\n",
    "    conv_out = np.zeros((N, 1, out_h, out_w), dtype=np.float32)  # (1,1,3,3)\n",
    "    for i in range(N):\n",
    "        # 取出第 i 个样本的 im2col: shape (4,9)\n",
    "        sample_col = col_x[i]  # (4,9)\n",
    "        # 卷积: (1,4) @ (4,9) = (1,9)\n",
    "        out_vec = w_flat @ sample_col  # (1,9)\n",
    "        # 加上偏置 b\n",
    "        out_vec += b[0]  # out_vec shape 依旧 (1,9)\n",
    "        # reshape -> (1,3,3)\n",
    "        out_2d = out_vec.reshape(1, out_h, out_w)\n",
    "        conv_out[i] = out_2d\n",
    "    \n",
    "    print(\"im2col 的结果 col_x.shape:\", col_x.shape)\n",
    "    print(col_x)\n",
    "\n",
    "    print(\"\\n卷积核 w:\\n\", w)\n",
    "    print(\"卷积偏置 b:\", b)\n",
    "\n",
    "    print(\"\\n卷积乘法后 conv_out.shape:\", conv_out.shape)\n",
    "    print(conv_out[0, 0])  # 展示 (3,3)"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 12,
   "id": "a8c04fb3-c08a-4cb7-bbd5-7cb59b32ae71",
   "metadata": {},
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "Epoch [1/3], Step [100/938], Loss: 0.4625\n",
      "Epoch [1/3], Step [200/938], Loss: 0.1340\n",
      "Epoch [1/3], Step [300/938], Loss: 0.0946\n",
      "Epoch [1/3], Step [400/938], Loss: 0.0837\n",
      "Epoch [1/3], Step [500/938], Loss: 0.0822\n",
      "Epoch [1/3], Step [600/938], Loss: 0.0602\n",
      "Epoch [1/3], Step [700/938], Loss: 0.0605\n",
      "Epoch [1/3], Step [800/938], Loss: 0.0503\n",
      "Epoch [1/3], Step [900/938], Loss: 0.0633\n",
      "Epoch [2/3], Step [100/938], Loss: 0.0303\n",
      "Epoch [2/3], Step [200/938], Loss: 0.0376\n",
      "Epoch [2/3], Step [300/938], Loss: 0.0373\n",
      "Epoch [2/3], Step [400/938], Loss: 0.0369\n",
      "Epoch [2/3], Step [500/938], Loss: 0.0399\n",
      "Epoch [2/3], Step [600/938], Loss: 0.0367\n",
      "Epoch [2/3], Step [700/938], Loss: 0.0322\n",
      "Epoch [2/3], Step [800/938], Loss: 0.0342\n",
      "Epoch [2/3], Step [900/938], Loss: 0.0366\n",
      "Epoch [3/3], Step [100/938], Loss: 0.0162\n",
      "Epoch [3/3], Step [200/938], Loss: 0.0208\n",
      "Epoch [3/3], Step [300/938], Loss: 0.0235\n",
      "Epoch [3/3], Step [400/938], Loss: 0.0214\n",
      "Epoch [3/3], Step [500/938], Loss: 0.0231\n",
      "Epoch [3/3], Step [600/938], Loss: 0.0264\n",
      "Epoch [3/3], Step [700/938], Loss: 0.0202\n",
      "Epoch [3/3], Step [800/938], Loss: 0.0189\n",
      "Epoch [3/3], Step [900/938], Loss: 0.0275\n",
      "Test Accuracy: 98.83%\n"
     ]
    }
   ],
   "source": [
    "import torch\n",
    "import torch.nn as nn\n",
    "import torch.optim as optim\n",
    "import torchvision\n",
    "import torchvision.transforms as transforms\n",
    "\n",
    "# ============ 1. 数据准备 ============\n",
    "\n",
    "def load_mnist_data(batch_size=64):\n",
    "    \"\"\"\n",
    "    返回 MNIST 训练集和测试集的 DataLoader\n",
    "    \"\"\"\n",
    "    transform = transforms.Compose([\n",
    "        transforms.ToTensor(), \n",
    "        transforms.Normalize((0.1307,), (0.3081,))  # MNIST 均值/方差\n",
    "    ])\n",
    "\n",
    "    train_dataset = torchvision.datasets.MNIST(\n",
    "        root='./data', train=True, download=True, transform=transform)\n",
    "    test_dataset = torchvision.datasets.MNIST(\n",
    "        root='./data', train=False, download=True, transform=transform)\n",
    "\n",
    "    train_loader = torch.utils.data.DataLoader(\n",
    "        train_dataset, batch_size=batch_size, shuffle=True, num_workers=2)\n",
    "    test_loader = torch.utils.data.DataLoader(\n",
    "        test_dataset, batch_size=batch_size, shuffle=False, num_workers=2)\n",
    "\n",
    "    return train_loader, test_loader\n",
    "\n",
    "\n",
    "# ============ 2. 定义网络结构 ============\n",
    "\n",
    "class SimpleCNN(nn.Module):\n",
    "    def __init__(self, num_classes=10):\n",
    "        super(SimpleCNN, self).__init__()\n",
    "        \n",
    "        # (1) 卷积层 1: 输入通道=1, 输出通道=32, kernel=3, padding=1\n",
    "        self.conv1 = nn.Conv2d(in_channels=1, out_channels=32, \n",
    "                               kernel_size=3, stride=1, padding=1)\n",
    "        self.relu1 = nn.ReLU()\n",
    "        \n",
    "        # (2) 卷积层 2: 输入通道=32, 输出通道=64, kernel=3, padding=1\n",
    "        self.conv2 = nn.Conv2d(in_channels=32, out_channels=64, \n",
    "                               kernel_size=3, stride=1, padding=1)\n",
    "        self.relu2 = nn.ReLU()\n",
    "\n",
    "        # (3) 池化层: kernel=2, stride=2 -> (28x28)->(14x14)\n",
    "        self.pool = nn.MaxPool2d(kernel_size=2, stride=2)\n",
    "        \n",
    "        # 计算池化后的特征图大小: 原图 28x28，池化后 14x14 (只做一次 pooling)。\n",
    "        # 卷积通道=64 -> 平铺后维度 = 64 * 14 * 14 = 12544\n",
    "        self.fc1 = nn.Linear(64 * 14 * 14, 128)\n",
    "        self.relu_fc1 = nn.ReLU()\n",
    "\n",
    "        # 最后一层分类输出\n",
    "        self.fc2 = nn.Linear(128, num_classes)  # 10\n",
    "        \n",
    "    def forward(self, x):\n",
    "        # 输入 x: shape = (N, 1, 28, 28)\n",
    "        out = self.conv1(x)   # (N,32,28,28)\n",
    "        out = self.relu1(out)\n",
    "        \n",
    "        out = self.conv2(out) # (N,64,28,28)\n",
    "        out = self.relu2(out)\n",
    "        \n",
    "        out = self.pool(out)  # (N,64,14,14)\n",
    "        \n",
    "        # 展开\n",
    "        out = out.view(out.size(0), -1)  # (N,64*14*14) = (N,12544)\n",
    "        \n",
    "        out = self.fc1(out)   # (N,128)\n",
    "        out = self.relu_fc1(out)\n",
    "        \n",
    "        out = self.fc2(out)   # (N,10)\n",
    "        return out\n",
    "\n",
    "\n",
    "# ============ 3. 训练与测试流程 ============\n",
    "\n",
    "def train_and_test(epochs=3, batch_size=64, lr=0.001):\n",
    "    # 加载数据\n",
    "    train_loader, test_loader = load_mnist_data(batch_size)\n",
    "    \n",
    "    # 定义网络\n",
    "    model = SimpleCNN(num_classes=10)\n",
    "    \n",
    "    # 定义损失函数 & 优化器\n",
    "    criterion = nn.CrossEntropyLoss()  # 交叉熵损失\n",
    "    optimizer = optim.Adam(model.parameters(), lr=lr)\n",
    "\n",
    "    # ============= 训练 =============\n",
    "    model.train()\n",
    "    for epoch in range(1, epochs+1):\n",
    "        running_loss = 0.0\n",
    "        for i, (images, labels) in enumerate(train_loader):\n",
    "            # 前向\n",
    "            outputs = model(images)\n",
    "            loss = criterion(outputs, labels)\n",
    "            \n",
    "            # 反向\n",
    "            optimizer.zero_grad()\n",
    "            loss.backward()\n",
    "            optimizer.step()\n",
    "            \n",
    "            running_loss += loss.item()\n",
    "            \n",
    "            if (i+1) % 100 == 0:\n",
    "                print(f\"Epoch [{epoch}/{epochs}], Step [{i+1}/{len(train_loader)}], \"\n",
    "                      f\"Loss: {running_loss/100:.4f}\")\n",
    "                running_loss = 0.0\n",
    "\n",
    "    # ============= 测试 =============\n",
    "    model.eval()\n",
    "    correct = 0\n",
    "    total = 0\n",
    "    with torch.no_grad():\n",
    "        for images, labels in test_loader:\n",
    "            outputs = model(images)               # (batch_size,10)\n",
    "            _, predicted = torch.max(outputs, 1)  # 取分类最大概率对应的下标\n",
    "            total += labels.size(0)\n",
    "            correct += (predicted == labels).sum().item()\n",
    "    \n",
    "    print(f\"Test Accuracy: {100 * correct/total:.2f}%\")\n",
    "\n",
    "\n",
    "# ============ 4. 主函数入口 ============\n",
    "\n",
    "if __name__ == \"__main__\":\n",
    "    train_and_test(epochs=3, batch_size=64, lr=0.001)"
   ]
  },
  {
   "cell_type": "markdown",
   "id": "426292d0-cc4a-4e94-b1fb-51a98797663e",
   "metadata": {},
   "source": [
    "1. 带池化层 (Pooling)、不划分验证集 —— cnn_with_pool_no_val.py"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 16,
   "id": "fa6b2144-01e4-459b-804e-e2c7f8ccd0aa",
   "metadata": {},
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "Total steps per epoch: 938\n",
      "Epoch [1/3], Step [100/938], Loss: 0.4233\n",
      "Epoch [1/3], Step [200/938], Loss: 0.1228\n",
      "Epoch [1/3], Step [300/938], Loss: 0.1022\n",
      "Epoch [1/3], Step [400/938], Loss: 0.0763\n",
      "Epoch [1/3], Step [500/938], Loss: 0.0766\n",
      "Epoch [1/3], Step [600/938], Loss: 0.0583\n",
      "Epoch [1/3], Step [700/938], Loss: 0.0643\n",
      "Epoch [1/3], Step [800/938], Loss: 0.0583\n",
      "Epoch [1/3], Step [900/938], Loss: 0.0546\n",
      "Epoch [2/3], Step [100/938], Loss: 0.0297\n",
      "Epoch [2/3], Step [200/938], Loss: 0.0362\n",
      "Epoch [2/3], Step [300/938], Loss: 0.0319\n",
      "Epoch [2/3], Step [400/938], Loss: 0.0391\n",
      "Epoch [2/3], Step [500/938], Loss: 0.0373\n",
      "Epoch [2/3], Step [600/938], Loss: 0.0395\n",
      "Epoch [2/3], Step [700/938], Loss: 0.0408\n",
      "Epoch [2/3], Step [800/938], Loss: 0.0343\n",
      "Epoch [2/3], Step [900/938], Loss: 0.0429\n",
      "Epoch [3/3], Step [100/938], Loss: 0.0159\n",
      "Epoch [3/3], Step [200/938], Loss: 0.0219\n",
      "Epoch [3/3], Step [300/938], Loss: 0.0184\n",
      "Epoch [3/3], Step [400/938], Loss: 0.0185\n",
      "Epoch [3/3], Step [500/938], Loss: 0.0243\n",
      "Epoch [3/3], Step [600/938], Loss: 0.0197\n",
      "Epoch [3/3], Step [700/938], Loss: 0.0245\n",
      "Epoch [3/3], Step [800/938], Loss: 0.0236\n",
      "Epoch [3/3], Step [900/938], Loss: 0.0252\n",
      "Test Accuracy: 98.93%\n"
     ]
    }
   ],
   "source": [
    "\"\"\"\n",
    "cnn_with_pool_no_val.py\n",
    "\n",
    "在 MNIST 上训练一个带池化层的 CNN，不划分验证集。\n",
    "打印类似：\n",
    "Epoch [1/3], Step [100/938], Loss: 0.4625\n",
    "...\n",
    "Test Accuracy: 98.83%\n",
    "\"\"\"\n",
    "\n",
    "import torch\n",
    "import torch.nn as nn\n",
    "import torch.optim as optim\n",
    "import torchvision\n",
    "import torchvision.transforms as transforms\n",
    "\n",
    "# ------------------------------\n",
    "# 1. 加载 MNIST 数据 (无验证集)\n",
    "# ------------------------------\n",
    "def load_mnist_data(batch_size=64):\n",
    "    transform = transforms.Compose([\n",
    "        transforms.ToTensor(),\n",
    "        transforms.Normalize((0.1307,), (0.3081,))\n",
    "    ])\n",
    "    \n",
    "    # 训练集: 60k，全做训练\n",
    "    train_dataset = torchvision.datasets.MNIST(\n",
    "        root='./data', train=True, download=True, transform=transform\n",
    "    )\n",
    "    train_loader = torch.utils.data.DataLoader(\n",
    "        train_dataset, batch_size=batch_size, shuffle=True, num_workers=2\n",
    "    )\n",
    "    \n",
    "    # 测试集: 10k\n",
    "    test_dataset = torchvision.datasets.MNIST(\n",
    "        root='./data', train=False, download=True, transform=transform\n",
    "    )\n",
    "    test_loader = torch.utils.data.DataLoader(\n",
    "        test_dataset, batch_size=batch_size, shuffle=False, num_workers=2\n",
    "    )\n",
    "    \n",
    "    return train_loader, test_loader\n",
    "\n",
    "# ------------------------------\n",
    "# 2. 带池化层的 CNN\n",
    "# ------------------------------\n",
    "class CNNWithPool(nn.Module):\n",
    "    def __init__(self, num_classes=10):\n",
    "        super(CNNWithPool, self).__init__()\n",
    "        \n",
    "        self.conv1 = nn.Conv2d(in_channels=1, out_channels=32,\n",
    "                               kernel_size=3, stride=1, padding=1)\n",
    "        self.relu1 = nn.ReLU()\n",
    "        \n",
    "        self.conv2 = nn.Conv2d(in_channels=32, out_channels=64,\n",
    "                               kernel_size=3, stride=1, padding=1)\n",
    "        self.relu2 = nn.ReLU()\n",
    "\n",
    "        self.pool = nn.MaxPool2d(kernel_size=2, stride=2)  # 28->14\n",
    "        \n",
    "        self.fc1 = nn.Linear(64 * 14 * 14, 128)\n",
    "        self.relu_fc1 = nn.ReLU()\n",
    "        self.fc2 = nn.Linear(128, num_classes)\n",
    "        \n",
    "    def forward(self, x):\n",
    "        x = self.conv1(x)\n",
    "        x = self.relu1(x)\n",
    "        x = self.conv2(x)\n",
    "        x = self.relu2(x)\n",
    "        \n",
    "        x = self.pool(x)  # (N,64,14,14)\n",
    "        \n",
    "        x = x.view(x.size(0), -1)  # (N,64*14*14)\n",
    "        x = self.fc1(x)\n",
    "        x = self.relu_fc1(x)\n",
    "        x = self.fc2(x)\n",
    "        return x\n",
    "\n",
    "# ------------------------------\n",
    "# 3. 训练和测试逻辑\n",
    "# ------------------------------\n",
    "def evaluate_accuracy(model, data_loader):\n",
    "    model.eval()\n",
    "    correct = 0\n",
    "    total = 0\n",
    "    with torch.no_grad():\n",
    "        for images, labels in data_loader:\n",
    "            outputs = model(images)\n",
    "            _, predicted = torch.max(outputs, dim=1)\n",
    "            correct += (predicted == labels).sum().item()\n",
    "            total += labels.size(0)\n",
    "    return 100.0 * correct / total\n",
    "\n",
    "def train_test_model(epochs=3, batch_size=64, lr=1e-3):\n",
    "    train_loader, test_loader = load_mnist_data(batch_size)\n",
    "    model = CNNWithPool(num_classes=10)\n",
    "    \n",
    "    criterion = nn.CrossEntropyLoss()\n",
    "    optimizer = optim.Adam(model.parameters(), lr=lr)\n",
    "    \n",
    "    total_steps = len(train_loader)  # 一共多少个batch\n",
    "    print(f\"Total steps per epoch: {total_steps}\")  # 一般 ~938 (60k/64)\n",
    "    \n",
    "    # 训练\n",
    "    for epoch in range(1, epochs+1):\n",
    "        model.train()\n",
    "        running_loss = 0.0\n",
    "        \n",
    "        for i, (images, labels) in enumerate(train_loader):\n",
    "            outputs = model(images)\n",
    "            loss = criterion(outputs, labels)\n",
    "            \n",
    "            optimizer.zero_grad()\n",
    "            loss.backward()\n",
    "            optimizer.step()\n",
    "            \n",
    "            running_loss += loss.item()\n",
    "            \n",
    "            # 每100 step打印一次\n",
    "            if (i+1) % 100 == 0:\n",
    "                avg_loss = running_loss / 100\n",
    "                print(f\"Epoch [{epoch}/{epochs}], Step [{i+1}/{total_steps}], Loss: {avg_loss:.4f}\")\n",
    "                running_loss = 0.0\n",
    "    \n",
    "    # 测试\n",
    "    test_acc = evaluate_accuracy(model, test_loader)\n",
    "    print(f\"Test Accuracy: {test_acc:.2f}%\")\n",
    "\n",
    "# ------------------------------\n",
    "# 4. 主函数\n",
    "# ------------------------------\n",
    "if __name__ == \"__main__\":\n",
    "    train_test_model(epochs=3, batch_size=64, lr=1e-3)"
   ]
  },
  {
   "cell_type": "markdown",
   "id": "2127e66e-a646-4c17-9292-2eeb06860bf0",
   "metadata": {},
   "source": [
    "2. 不带池化层、划分验证集 —— cnn_no_pool_with_val.py"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 20,
   "id": "d2f1804d-b024-43e9-8748-4ea3f50fbfc9",
   "metadata": {},
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "Total steps per epoch: 860\n",
      "Epoch [1/3], Step [100/860], Loss: 0.4814\n",
      "Epoch [1/3], Step [200/860], Loss: 0.1527\n",
      "Epoch [1/3], Step [300/860], Loss: 0.1111\n",
      "Epoch [1/3], Step [400/860], Loss: 0.0819\n",
      "Epoch [1/3], Step [500/860], Loss: 0.0839\n",
      "Epoch [1/3], Step [600/860], Loss: 0.0761\n",
      "Epoch [1/3], Step [700/860], Loss: 0.0748\n",
      "Epoch [1/3], Step [800/860], Loss: 0.0608\n",
      "[Val] Accuracy: 97.80%\n",
      "Epoch [2/3], Step [100/860], Loss: 0.0365\n",
      "Epoch [2/3], Step [200/860], Loss: 0.0441\n",
      "Epoch [2/3], Step [300/860], Loss: 0.0379\n",
      "Epoch [2/3], Step [400/860], Loss: 0.0435\n",
      "Epoch [2/3], Step [500/860], Loss: 0.0391\n",
      "Epoch [2/3], Step [600/860], Loss: 0.0432\n",
      "Epoch [2/3], Step [700/860], Loss: 0.0356\n",
      "Epoch [2/3], Step [800/860], Loss: 0.0458\n",
      "[Val] Accuracy: 98.12%\n",
      "Epoch [3/3], Step [100/860], Loss: 0.0165\n",
      "Epoch [3/3], Step [200/860], Loss: 0.0186\n",
      "Epoch [3/3], Step [300/860], Loss: 0.0181\n",
      "Epoch [3/3], Step [400/860], Loss: 0.0260\n",
      "Epoch [3/3], Step [500/860], Loss: 0.0255\n",
      "Epoch [3/3], Step [600/860], Loss: 0.0211\n",
      "Epoch [3/3], Step [700/860], Loss: 0.0308\n",
      "Epoch [3/3], Step [800/860], Loss: 0.0251\n",
      "[Val] Accuracy: 98.48%\n",
      "Test Accuracy: 98.77%\n"
     ]
    }
   ],
   "source": [
    "\"\"\"\n",
    "cnn_no_pool_with_val.py\n",
    "\n",
    "在 MNIST 上训练一个不带池化层的 CNN，划分出验证集 (5k)。\n",
    "打印类似：\n",
    "Epoch [1/3], Step [100/938], Loss: 0.4625\n",
    "...\n",
    "[Val] Accuracy: 97.32%\n",
    "Test Accuracy: 98.83%\n",
    "\"\"\"\n",
    "\n",
    "import torch\n",
    "import torch.nn as nn\n",
    "import torch.optim as optim\n",
    "import torchvision\n",
    "import torchvision.transforms as transforms\n",
    "\n",
    "# ------------------------------\n",
    "# 1. 加载 MNIST + 划分验证集\n",
    "# ------------------------------\n",
    "def load_mnist_data(batch_size=64):\n",
    "    transform = transforms.Compose([\n",
    "        transforms.ToTensor(),\n",
    "        transforms.Normalize((0.1307,), (0.3081,))\n",
    "    ])\n",
    "    \n",
    "    # 整体训练集 60k -> 拆分 55k 训练 + 5k 验证\n",
    "    full_train_dataset = torchvision.datasets.MNIST(\n",
    "        root='./data', train=True, download=True, transform=transform\n",
    "    )\n",
    "    train_size = 55000\n",
    "    val_size   = 5000\n",
    "    train_dataset, val_dataset = torch.utils.data.random_split(\n",
    "        full_train_dataset, [train_size, val_size]\n",
    "    )\n",
    "    \n",
    "    train_loader = torch.utils.data.DataLoader(\n",
    "        train_dataset, batch_size=batch_size, shuffle=True, num_workers=2\n",
    "    )\n",
    "    val_loader = torch.utils.data.DataLoader(\n",
    "        val_dataset, batch_size=batch_size, shuffle=False, num_workers=2\n",
    "    )\n",
    "    \n",
    "    # 测试集 10k\n",
    "    test_dataset = torchvision.datasets.MNIST(\n",
    "        root='./data', train=False, download=True, transform=transform\n",
    "    )\n",
    "    test_loader = torch.utils.data.DataLoader(\n",
    "        test_dataset, batch_size=batch_size, shuffle=False, num_workers=2\n",
    "    )\n",
    "    \n",
    "    return train_loader, val_loader, test_loader\n",
    "\n",
    "\n",
    "# ------------------------------\n",
    "# 2. 不带池化层的 CNN (用 stride=2 下采样)\n",
    "# ------------------------------\n",
    "class CNNNoPool(nn.Module):\n",
    "    def __init__(self, num_classes=10):\n",
    "        super(CNNNoPool, self).__init__()\n",
    "        \n",
    "        # conv1: stride=1, 不改尺寸\n",
    "        self.conv1 = nn.Conv2d(in_channels=1, out_channels=32,\n",
    "                               kernel_size=3, stride=1, padding=1)\n",
    "        self.relu1 = nn.ReLU()\n",
    "        \n",
    "        # conv2: stride=2，相当于下采样 => 28->14\n",
    "        self.conv2 = nn.Conv2d(in_channels=32, out_channels=64,\n",
    "                               kernel_size=3, stride=2, padding=1)\n",
    "        self.relu2 = nn.ReLU()\n",
    "        \n",
    "        # 现在尺寸 14x14，通道64 => 64*14*14=12544\n",
    "        self.fc1 = nn.Linear(64*14*14, 128)\n",
    "        self.relu_fc1 = nn.ReLU()\n",
    "        self.fc2 = nn.Linear(128, num_classes)\n",
    "        \n",
    "    def forward(self, x):\n",
    "        x = self.conv1(x)\n",
    "        x = self.relu1(x)\n",
    "        \n",
    "        x = self.conv2(x)\n",
    "        x = self.relu2(x)\n",
    "        \n",
    "        x = x.view(x.size(0), -1)  # (N,12544)\n",
    "        x = self.fc1(x)\n",
    "        x = self.relu_fc1(x)\n",
    "        x = self.fc2(x)\n",
    "        return x\n",
    "\n",
    "# ------------------------------\n",
    "# 3. 训练、验证、测试流程\n",
    "# ------------------------------\n",
    "def evaluate_accuracy(model, data_loader):\n",
    "    model.eval()\n",
    "    correct = 0\n",
    "    total = 0\n",
    "    with torch.no_grad():\n",
    "        for images, labels in data_loader:\n",
    "            outputs = model(images)\n",
    "            _, predicted = torch.max(outputs, dim=1)\n",
    "            correct += (predicted == labels).sum().item()\n",
    "            total += labels.size(0)\n",
    "    return 100.0 * correct / total\n",
    "\n",
    "def train_test_model(epochs=3, batch_size=64, lr=1e-3):\n",
    "    train_loader, val_loader, test_loader = load_mnist_data(batch_size)\n",
    "    model = CNNNoPool(num_classes=10)\n",
    "    \n",
    "    criterion = nn.CrossEntropyLoss()\n",
    "    optimizer = optim.Adam(model.parameters(), lr=lr)\n",
    "    \n",
    "    total_steps = len(train_loader)\n",
    "    print(f\"Total steps per epoch: {total_steps}\")\n",
    "    \n",
    "    # 训练\n",
    "    for epoch in range(1, epochs+1):\n",
    "        model.train()\n",
    "        running_loss = 0.0\n",
    "        \n",
    "        for i, (images, labels) in enumerate(train_loader):\n",
    "            outputs = model(images)\n",
    "            loss = criterion(outputs, labels)\n",
    "            \n",
    "            optimizer.zero_grad()\n",
    "            loss.backward()\n",
    "            optimizer.step()\n",
    "            \n",
    "            running_loss += loss.item()\n",
    "            \n",
    "            if (i+1) % 100 == 0:\n",
    "                avg_loss = running_loss / 100\n",
    "                print(f\"Epoch [{epoch}/{epochs}], Step [{i+1}/{total_steps}], Loss: {avg_loss:.4f}\")\n",
    "                running_loss = 0.0\n",
    "        \n",
    "        # 每个 epoch 结束后，做一次验证集准确率\n",
    "        val_acc = evaluate_accuracy(model, val_loader)\n",
    "        print(f\"[Val] Accuracy: {val_acc:.2f}%\")\n",
    "    \n",
    "    # 测试\n",
    "    test_acc = evaluate_accuracy(model, test_loader)\n",
    "    print(f\"Test Accuracy: {test_acc:.2f}%\")\n",
    "\n",
    "# ------------------------------\n",
    "# 4. 主函数\n",
    "# ------------------------------\n",
    "if __name__ == \"__main__\":\n",
    "    train_test_model(epochs=3, batch_size=64, lr=1e-3)"
   ]
  },
  {
   "cell_type": "markdown",
   "id": "fed645b3-7309-44e7-90fa-4955182ed50c",
   "metadata": {},
   "source": [
    "3. 带池化层、划分验证集 —— cnn_with_pool_with_val.py"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 21,
   "id": "55acdcdf-0ce2-4231-a098-e0300573aaaf",
   "metadata": {},
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "Total steps per epoch: 860\n",
      "Epoch [1/3], Step [100/860], Loss: 0.4421\n",
      "Epoch [1/3], Step [200/860], Loss: 0.1316\n",
      "Epoch [1/3], Step [300/860], Loss: 0.1127\n",
      "Epoch [1/3], Step [400/860], Loss: 0.0921\n",
      "Epoch [1/3], Step [500/860], Loss: 0.0723\n",
      "Epoch [1/3], Step [600/860], Loss: 0.0720\n",
      "Epoch [1/3], Step [700/860], Loss: 0.0717\n",
      "Epoch [1/3], Step [800/860], Loss: 0.0597\n",
      "[Val] Accuracy: 98.60%\n",
      "Epoch [2/3], Step [100/860], Loss: 0.0386\n",
      "Epoch [2/3], Step [200/860], Loss: 0.0468\n",
      "Epoch [2/3], Step [300/860], Loss: 0.0385\n",
      "Epoch [2/3], Step [400/860], Loss: 0.0385\n",
      "Epoch [2/3], Step [500/860], Loss: 0.0420\n",
      "Epoch [2/3], Step [600/860], Loss: 0.0364\n",
      "Epoch [2/3], Step [700/860], Loss: 0.0426\n",
      "Epoch [2/3], Step [800/860], Loss: 0.0332\n",
      "[Val] Accuracy: 98.58%\n",
      "Epoch [3/3], Step [100/860], Loss: 0.0178\n",
      "Epoch [3/3], Step [200/860], Loss: 0.0196\n",
      "Epoch [3/3], Step [300/860], Loss: 0.0201\n",
      "Epoch [3/3], Step [400/860], Loss: 0.0217\n",
      "Epoch [3/3], Step [500/860], Loss: 0.0245\n",
      "Epoch [3/3], Step [600/860], Loss: 0.0344\n",
      "Epoch [3/3], Step [700/860], Loss: 0.0269\n",
      "Epoch [3/3], Step [800/860], Loss: 0.0225\n",
      "[Val] Accuracy: 98.94%\n",
      "Test Accuracy: 99.04%\n"
     ]
    }
   ],
   "source": [
    "\"\"\"\n",
    "cnn_with_pool_with_val.py\n",
    "\n",
    "在 MNIST 上训练一个带池化层的 CNN，并划分出验证集 (5k)。\n",
    "打印类似：\n",
    "Epoch [1/3], Step [100/938], Loss: 0.4625\n",
    "...\n",
    "[Val] Accuracy: 97.32%\n",
    "Test Accuracy: 98.83%\n",
    "\"\"\"\n",
    "\n",
    "import torch\n",
    "import torch.nn as nn\n",
    "import torch.optim as optim\n",
    "import torchvision\n",
    "import torchvision.transforms as transforms\n",
    "\n",
    "# ------------------------------\n",
    "# 1. 加载 MNIST + 划分验证集\n",
    "# ------------------------------\n",
    "def load_mnist_data(batch_size=64):\n",
    "    transform = transforms.Compose([\n",
    "        transforms.ToTensor(),\n",
    "        transforms.Normalize((0.1307,), (0.3081,))\n",
    "    ])\n",
    "    \n",
    "    full_train_dataset = torchvision.datasets.MNIST(\n",
    "        root='./data', train=True, download=True, transform=transform\n",
    "    )\n",
    "    train_size = 55000\n",
    "    val_size   = 5000\n",
    "    train_dataset, val_dataset = torch.utils.data.random_split(\n",
    "        full_train_dataset, [train_size, val_size]\n",
    "    )\n",
    "    \n",
    "    train_loader = torch.utils.data.DataLoader(\n",
    "        train_dataset, batch_size=batch_size, shuffle=True, num_workers=2\n",
    "    )\n",
    "    val_loader = torch.utils.data.DataLoader(\n",
    "        val_dataset, batch_size=batch_size, shuffle=False, num_workers=2\n",
    "    )\n",
    "    \n",
    "    test_dataset = torchvision.datasets.MNIST(\n",
    "        root='./data', train=False, download=True, transform=transform\n",
    "    )\n",
    "    test_loader = torch.utils.data.DataLoader(\n",
    "        test_dataset, batch_size=batch_size, shuffle=False, num_workers=2\n",
    "    )\n",
    "    \n",
    "    return train_loader, val_loader, test_loader\n",
    "\n",
    "# ------------------------------\n",
    "# 2. 带池化层的 CNN\n",
    "# ------------------------------\n",
    "class CNNWithPool(nn.Module):\n",
    "    def __init__(self, num_classes=10):\n",
    "        super(CNNWithPool, self).__init__()\n",
    "        \n",
    "        self.conv1 = nn.Conv2d(in_channels=1, out_channels=32,\n",
    "                               kernel_size=3, stride=1, padding=1)\n",
    "        self.relu1 = nn.ReLU()\n",
    "        \n",
    "        self.conv2 = nn.Conv2d(in_channels=32, out_channels=64,\n",
    "                               kernel_size=3, stride=1, padding=1)\n",
    "        self.relu2 = nn.ReLU()\n",
    "\n",
    "        self.pool = nn.MaxPool2d(kernel_size=2, stride=2)  # 28->14\n",
    "        \n",
    "        self.fc1 = nn.Linear(64 * 14 * 14, 128)\n",
    "        self.relu_fc1 = nn.ReLU()\n",
    "        self.fc2 = nn.Linear(128, num_classes)\n",
    "        \n",
    "    def forward(self, x):\n",
    "        x = self.conv1(x)\n",
    "        x = self.relu1(x)\n",
    "        x = self.conv2(x)\n",
    "        x = self.relu2(x)\n",
    "        \n",
    "        x = self.pool(x)\n",
    "        \n",
    "        x = x.view(x.size(0), -1)\n",
    "        x = self.fc1(x)\n",
    "        x = self.relu_fc1(x)\n",
    "        x = self.fc2(x)\n",
    "        return x\n",
    "\n",
    "# ------------------------------\n",
    "# 3. 训练、验证、测试流程\n",
    "# ------------------------------\n",
    "def evaluate_accuracy(model, data_loader):\n",
    "    model.eval()\n",
    "    correct = 0\n",
    "    total = 0\n",
    "    with torch.no_grad():\n",
    "        for images, labels in data_loader:\n",
    "            outputs = model(images)\n",
    "            _, predicted = torch.max(outputs, dim=1)\n",
    "            correct += (predicted == labels).sum().item()\n",
    "            total += labels.size(0)\n",
    "    return 100.0 * correct / total\n",
    "\n",
    "def train_test_model(epochs=3, batch_size=64, lr=1e-3):\n",
    "    train_loader, val_loader, test_loader = load_mnist_data(batch_size)\n",
    "    \n",
    "    model = CNNWithPool(num_classes=10)\n",
    "    criterion = nn.CrossEntropyLoss()\n",
    "    optimizer = optim.Adam(model.parameters(), lr=lr)\n",
    "    \n",
    "    total_steps = len(train_loader)\n",
    "    print(f\"Total steps per epoch: {total_steps}\")\n",
    "    \n",
    "    # 训练\n",
    "    for epoch in range(1, epochs+1):\n",
    "        model.train()\n",
    "        running_loss = 0.0\n",
    "        \n",
    "        for i, (images, labels) in enumerate(train_loader):\n",
    "            outputs = model(images)\n",
    "            loss = criterion(outputs, labels)\n",
    "            \n",
    "            optimizer.zero_grad()\n",
    "            loss.backward()\n",
    "            optimizer.step()\n",
    "            \n",
    "            running_loss += loss.item()\n",
    "            \n",
    "            # 每 100 steps 打印一次\n",
    "            if (i+1) % 100 == 0:\n",
    "                avg_loss = running_loss / 100\n",
    "                print(f\"Epoch [{epoch}/{epochs}], Step [{i+1}/{total_steps}], Loss: {avg_loss:.4f}\")\n",
    "                running_loss = 0.0\n",
    "        \n",
    "        # 验证集\n",
    "        val_acc = evaluate_accuracy(model, val_loader)\n",
    "        print(f\"[Val] Accuracy: {val_acc:.2f}%\")\n",
    "    \n",
    "    # 测试集\n",
    "    test_acc = evaluate_accuracy(model, test_loader)\n",
    "    print(f\"Test Accuracy: {test_acc:.2f}%\")\n",
    "\n",
    "# ------------------------------\n",
    "# 4. 主函数\n",
    "# ------------------------------\n",
    "if __name__ == \"__main__\":\n",
    "    train_test_model(epochs=3, batch_size=64, lr=1e-3)"
   ]
  },
  {
   "cell_type": "markdown",
   "id": "eb744713-ec62-434c-927d-643285b68184",
   "metadata": {},
   "source": [
    "「带池化层 + 划分验证集」的网络在 MNIST 上往往能达到更高的准确率\n",
    "\n",
    "1. 池化层帮助抑制过拟合并提取稳健特征\n",
    "* 池化层 (Pooling) 的引入，会降低特征图分辨率、减少可学习参数，从而减轻过拟合；\n",
    "* 它还带来一定程度的「平移不变性」，使网络对输入细微变动更具鲁棒性；\n",
    "* 对于 MNIST 这样的小图像，池化操作（如 MaxPool2d(kernel=2, stride=2)）通常能快速稳定地提取关键信息，提升模型泛化能力。\n",
    "\n",
    "2. 验证集指导超参数调优，进一步提升泛化\n",
    "* 验证集 (Validation Set) 可在训练过程中及时评估模型表现：\n",
    "* 监控是否出现过拟合；\n",
    "* 动态调整学习率、迭代次数、正则化系数等超参数；\n",
    "* 进行早停 (Early Stopping) 或其他策略时也能防止过拟合到训练集本身。\n",
    "* 没有验证集时，往往会把所有训练样本都用于训练，虽然能获得更多的训练量，但可能无法及时监测模型对新数据的泛化表现，导致超参数选取不够精细、难以及时纠正过拟合或欠拟合问题。\n",
    "* 有了验证集，可以更精细地找到合适的超参数设置，通常会使最终测试集准确率更高。\n",
    "\n",
    "3. 两者结合：减少过拟合 + 合理调参\n",
    "\n",
    "当池化层与验证集调参一起使用时：\n",
    "\n",
    "    1. 网络结构上，池化层为卷积神经网络提供了简洁有效的特征提取方式；\n",
    "    \n",
    "    2. 训练流程上，验证集可帮助更好地控制训练和调参，把模型「打磨」到最佳状态；\n",
    "    \n",
    "    3. 结果：在 MNIST 这类相对简单的数据集上，更容易达到 99% 或更高的准确率。\n",
    "\n",
    "在更大规模或更复杂的任务中，这两者（合理的网络结构 + 验证集调参）依旧是深度学习项目中常见且有效的组合策略。"
   ]
  },
  {
   "cell_type": "markdown",
   "id": "9363f406-42f3-4fb4-9342-4c1e0f171f78",
   "metadata": {},
   "source": [
    "cnn_with_pool_with_val_compare.py GPU VS CPU"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "id": "edd87ad1-e55d-45d8-8bb3-ddb9877dc559",
   "metadata": {},
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "\n",
      "Using device: mps, total steps per epoch = 860\n",
      "Epoch [1/3], Step [100/860], Loss: 0.4981\n",
      "Epoch [1/3], Step [200/860], Loss: 0.1493\n",
      "Epoch [1/3], Step [300/860], Loss: 0.1068\n",
      "Epoch [1/3], Step [400/860], Loss: 0.0846\n",
      "Epoch [1/3], Step [500/860], Loss: 0.0771\n",
      "Epoch [1/3], Step [600/860], Loss: 0.0673\n",
      "Epoch [1/3], Step [700/860], Loss: 0.0601\n",
      "Epoch [1/3], Step [800/860], Loss: 0.0625\n",
      "[Val] Accuracy: 98.36%\n",
      "Epoch [2/3], Step [100/860], Loss: 0.0375\n",
      "Epoch [2/3], Step [200/860], Loss: 0.0379\n",
      "Epoch [2/3], Step [300/860], Loss: 0.0389\n",
      "Epoch [2/3], Step [400/860], Loss: 0.0393\n",
      "Epoch [2/3], Step [500/860], Loss: 0.0392\n",
      "Epoch [2/3], Step [600/860], Loss: 0.0384\n",
      "Epoch [2/3], Step [700/860], Loss: 0.0418\n",
      "Epoch [2/3], Step [800/860], Loss: 0.0415\n",
      "[Val] Accuracy: 98.60%\n",
      "Epoch [3/3], Step [100/860], Loss: 0.0189\n",
      "Epoch [3/3], Step [200/860], Loss: 0.0228\n",
      "Epoch [3/3], Step [300/860], Loss: 0.0257\n",
      "Epoch [3/3], Step [400/860], Loss: 0.0213\n",
      "Epoch [3/3], Step [500/860], Loss: 0.0221\n",
      "Epoch [3/3], Step [600/860], Loss: 0.0239\n",
      "Epoch [3/3], Step [700/860], Loss: 0.0265\n",
      "Epoch [3/3], Step [800/860], Loss: 0.0262\n",
      "[Val] Accuracy: 98.62%\n",
      "Training completed in 19.24 seconds on device=mps.\n",
      "Test Accuracy on mps: 98.70%\n",
      "\n",
      "Using device: cpu, total steps per epoch = 860\n",
      "Epoch [1/3], Step [100/860], Loss: 0.4349\n",
      "Epoch [1/3], Step [200/860], Loss: 0.1432\n",
      "Epoch [1/3], Step [300/860], Loss: 0.0947\n",
      "Epoch [1/3], Step [400/860], Loss: 0.0872\n",
      "Epoch [1/3], Step [500/860], Loss: 0.0777\n",
      "Epoch [1/3], Step [600/860], Loss: 0.0828\n",
      "Epoch [1/3], Step [700/860], Loss: 0.0534\n",
      "Epoch [1/3], Step [800/860], Loss: 0.0565\n",
      "[Val] Accuracy: 98.36%\n",
      "Epoch [2/3], Step [100/860], Loss: 0.0336\n"
     ]
    }
   ],
   "source": [
    "\"\"\"\n",
    "cnn_with_pool_with_val_compare.py\n",
    "\n",
    "在 MNIST 上训练一个带池化层的 CNN，并划分出验证集 (5k)。\n",
    "比较在 Apple Silicon 环境下，使用 CPU vs. MPS (GPU) 的训练耗时和准确率。\n",
    "\n",
    "打印示例：\n",
    "Epoch [1/3], Step [100/938], Loss: 0.4625\n",
    "...\n",
    "[Val] Accuracy: 97.32%\n",
    "Test Accuracy: 98.83%\n",
    "\n",
    "并输出训练耗时：\n",
    "Training completed in XX.XX seconds on device=mps/cpu\n",
    "\"\"\"\n",
    "\n",
    "import time\n",
    "import torch\n",
    "import torch.nn as nn\n",
    "import torch.optim as optim\n",
    "import torchvision\n",
    "import torchvision.transforms as transforms\n",
    "\n",
    "# ------------------------------\n",
    "# 1. 加载 MNIST + 划分验证集\n",
    "# ------------------------------\n",
    "def load_mnist_data(batch_size=64):\n",
    "    transform = transforms.Compose([\n",
    "        transforms.ToTensor(),\n",
    "        transforms.Normalize((0.1307,), (0.3081,))\n",
    "    ])\n",
    "    \n",
    "    # 整体 60k -> 划分 55k(训练) + 5k(验证)\n",
    "    full_train_dataset = torchvision.datasets.MNIST(\n",
    "        root='./data', train=True, download=True, transform=transform\n",
    "    )\n",
    "    train_size = 55000\n",
    "    val_size   = 5000\n",
    "    train_dataset, val_dataset = torch.utils.data.random_split(\n",
    "        full_train_dataset, [train_size, val_size]\n",
    "    )\n",
    "    \n",
    "    train_loader = torch.utils.data.DataLoader(\n",
    "        train_dataset, batch_size=batch_size, shuffle=True, num_workers=2\n",
    "    )\n",
    "    val_loader = torch.utils.data.DataLoader(\n",
    "        val_dataset, batch_size=batch_size, shuffle=False, num_workers=2\n",
    "    )\n",
    "    \n",
    "    test_dataset = torchvision.datasets.MNIST(\n",
    "        root='./data', train=False, download=True, transform=transform\n",
    "    )\n",
    "    test_loader = torch.utils.data.DataLoader(\n",
    "        test_dataset, batch_size=batch_size, shuffle=False, num_workers=2\n",
    "    )\n",
    "    \n",
    "    return train_loader, val_loader, test_loader\n",
    "\n",
    "# ------------------------------\n",
    "# 2. 带池化层的 CNN\n",
    "# ------------------------------\n",
    "class CNNWithPool(nn.Module):\n",
    "    def __init__(self, num_classes=10):\n",
    "        super(CNNWithPool, self).__init__()\n",
    "        \n",
    "        self.conv1 = nn.Conv2d(in_channels=1, out_channels=32,\n",
    "                               kernel_size=3, stride=1, padding=1)\n",
    "        self.relu1 = nn.ReLU()\n",
    "        \n",
    "        self.conv2 = nn.Conv2d(in_channels=32, out_channels=64,\n",
    "                               kernel_size=3, stride=1, padding=1)\n",
    "        self.relu2 = nn.ReLU()\n",
    "\n",
    "        self.pool = nn.MaxPool2d(kernel_size=2, stride=2)  # 28->14\n",
    "        \n",
    "        self.fc1 = nn.Linear(64 * 14 * 14, 128)\n",
    "        self.relu_fc1 = nn.ReLU()\n",
    "        self.fc2 = nn.Linear(128, num_classes)\n",
    "        \n",
    "    def forward(self, x):\n",
    "        x = self.conv1(x)   # (N,32,28,28)\n",
    "        x = self.relu1(x)\n",
    "        x = self.conv2(x)   # (N,64,28,28)\n",
    "        x = self.relu2(x)\n",
    "        \n",
    "        x = self.pool(x)    # (N,64,14,14)\n",
    "        \n",
    "        x = x.view(x.size(0), -1)  # (N,64*14*14)\n",
    "        x = self.fc1(x)            # (N,128)\n",
    "        x = self.relu_fc1(x)\n",
    "        x = self.fc2(x)            # (N,10)\n",
    "        return x\n",
    "\n",
    "# ------------------------------\n",
    "# 3. 训练、验证、测试流程\n",
    "# ------------------------------\n",
    "def evaluate_accuracy(model, data_loader, device):\n",
    "    model.eval()\n",
    "    correct = 0\n",
    "    total = 0\n",
    "    with torch.no_grad():\n",
    "        for images, labels in data_loader:\n",
    "            images = images.to(device)\n",
    "            labels = labels.to(device)\n",
    "            \n",
    "            outputs = model(images)\n",
    "            _, predicted = torch.max(outputs, dim=1)\n",
    "            correct += (predicted == labels).sum().item()\n",
    "            total += labels.size(0)\n",
    "    return 100.0 * correct / total\n",
    "\n",
    "def train_test_model(device=\"cpu\", epochs=3, batch_size=64, lr=1e-3):\n",
    "    \"\"\"\n",
    "    在指定 device 上训练并测试:\n",
    "      - device: \"cpu\" or \"mps\" (Apple GPU)\n",
    "      - epochs, batch_size, lr: 超参数\n",
    "    \"\"\"\n",
    "    # 加载数据\n",
    "    train_loader, val_loader, test_loader = load_mnist_data(batch_size)\n",
    "    \n",
    "    # 初始化模型到指定设备 模型和数据默认是在 CPU 上。现在通过 .to(device) 显式地切换。\n",
    "    model = CNNWithPool(num_classes=10).to(device)\n",
    "    \n",
    "    criterion = nn.CrossEntropyLoss()\n",
    "    optimizer = optim.Adam(model.parameters(), lr=lr)\n",
    "    \n",
    "    total_steps = len(train_loader)\n",
    "    print(f\"\\nUsing device: {device}, total steps per epoch = {total_steps}\")\n",
    "    \n",
    "    # 训练\n",
    "    start_time = time.time()\n",
    "    for epoch in range(1, epochs+1):\n",
    "        model.train()\n",
    "        running_loss = 0.0\n",
    "        \n",
    "        for i, (images, labels) in enumerate(train_loader):\n",
    "            # 搬数据到 device\n",
    "            images = images.to(device)\n",
    "            labels = labels.to(device)\n",
    "            \n",
    "            outputs = model(images)\n",
    "            loss = criterion(outputs, labels)\n",
    "            \n",
    "            optimizer.zero_grad()\n",
    "            loss.backward()\n",
    "            optimizer.step()\n",
    "            \n",
    "            running_loss += loss.item()\n",
    "            \n",
    "            # 每 100 steps 打印一次\n",
    "            if (i+1) % 100 == 0:\n",
    "                avg_loss = running_loss / 100\n",
    "                print(f\"Epoch [{epoch}/{epochs}], Step [{i+1}/{total_steps}], Loss: {avg_loss:.4f}\")\n",
    "                running_loss = 0.0\n",
    "        \n",
    "        # 验证集准确率\n",
    "        val_acc = evaluate_accuracy(model, val_loader, device)\n",
    "        print(f\"[Val] Accuracy: {val_acc:.2f}%\")\n",
    "    end_time = time.time()\n",
    "    \n",
    "    # 计算训练耗时\n",
    "    elapsed_time = end_time - start_time\n",
    "    print(f\"Training completed in {elapsed_time:.2f} seconds on device={device}.\")\n",
    "\n",
    "    # 测试集准确率\n",
    "    test_acc = evaluate_accuracy(model, test_loader, device)\n",
    "    print(f\"Test Accuracy on {device}: {test_acc:.2f}%\")\n",
    "\n",
    "# ------------------------------\n",
    "# 4. 主函数：分别在 mps/cpu 上对比\n",
    "# ------------------------------\n",
    "if __name__ == \"__main__\":\n",
    "    # 判断 MPS 是否可用 (针对 macOS + M1/M2 芯片)\n",
    "    can_use_mps = torch.backends.mps.is_available()\n",
    "    \n",
    "    # 如果 MPS 可用，先跑 MPS 训练，再跑 CPU；否则只能跑 CPU\n",
    "    if can_use_mps:\n",
    "        train_test_model(device=\"mps\", epochs=3, batch_size=64, lr=1e-3)\n",
    "    \n",
    "    # CPU 训练 (可对比)\n",
    "    train_test_model(device=\"cpu\", epochs=3, batch_size=64, lr=1e-3)"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "id": "85dc055a-63b6-4927-9ef1-8102d5e4069a",
   "metadata": {},
   "outputs": [],
   "source": []
  }
 ],
 "metadata": {
  "kernelspec": {
   "display_name": "Python 3 (ipykernel)",
   "language": "python",
   "name": "python3"
  },
  "language_info": {
   "codemirror_mode": {
    "name": "ipython",
    "version": 3
   },
   "file_extension": ".py",
   "mimetype": "text/x-python",
   "name": "python",
   "nbconvert_exporter": "python",
   "pygments_lexer": "ipython3",
   "version": "3.12.7"
  }
 },
 "nbformat": 4,
 "nbformat_minor": 5
}