深度学习之卷积神经网络

import torch
from torch import nn
from d2l import torch as d2l

# 手写二维池化函数，用于演示池化层的核心计算过程
# X: 输入的二维张量（可理解为单通道特征图）
# pool_size: 池化窗口大小，格式为(窗口高, 窗口宽)
# mode: 池化模式，'max'表示最大池化，'avg'表示平均池化
def pool2d(X, pool_size, mode='max'):
    # 从池化窗口尺寸中拆出高和宽
    p_h, p_w = pool_size

    # 计算输出特征图大小：
    # 不使用填充(padding)、步幅默认为1时，输出高 = 输入高 - 窗口高 + 1，宽同理
    Y = torch.zeros((X.shape[0] - p_h + 1, X.shape[1] - p_w + 1))

    # 遍历输出张量的每一个位置
    # 输出位置(i, j)对应输入中的一个局部窗口 X[i:i+p_h, j:j+p_w]
    for i in range(Y.shape[0]):
        for j in range(Y.shape[1]):
            # 最大池化：取该窗口中的最大值
            if mode == 'max':
                Y[i, j] = X[i:i + p_h, j:j + p_w].max()
            # 平均池化：取该窗口中的平均值
            elif mode == 'avg':
                Y[i, j] = X[i:i + p_h, j:j + p_w].mean()

    # 返回池化结果
    return Y

# 构造一个3x3输入张量，元素按行递增，便于直观看池化结果
X = torch.tensor([[0.0, 1.0, 2.0], [3.0, 4.0, 5.0], [6.0, 7.0, 8.0]])

print("输入张量 X:")
print(X)

# 使用2x2窗口做最大池化，并打印输出
# 该示例会得到2x2结果，每个元素是对应2x2局部区域的最大值
print("最大池化结果:")
print(pool2d(X, (2, 2)))

# 使用2x2窗口做平均池化，并打印输出
# 该示例会得到2x2结果，每个元素是对应2x2局部区域的平均值
print("平均池化结果:")
print(pool2d(X, (2, 2), mode='avg'))

# 下面演示 PyTorch 内置池化层 nn.MaxPool2d 的用法
# 构造 4x4 输入，并补齐 batch 与 channel 维度，形状为 (N, C, H, W) = (1, 1, 4, 4)
X = torch.arange(16, dtype=torch.float32).reshape(1, 1, 4, 4)
print("输入张量 X:")
print(X)

# 定义最大池化层：
# kernel_size=3：窗口大小 3x3
# padding=1：输入四周补 1 圈 0
# stride=2：窗口每次移动 2 个像素
pool2d = nn.MaxPool2d(3, padding=1, stride=2)

# 为了直观展示 padding=1 的效果，这里先手动打印补零后的输入张量
X_padded = nn.functional.pad(X, pad=(1, 1, 1, 1), mode='constant', value=0)
print("padding=1 后的张量 X_padded:")
print(X_padded)

# 执行池化，得到输出 Y
Y = pool2d(X)
print("使用 nn.MaxPool2d 进行池化后的结果 Y:")
print(Y)

# -*- coding: utf-8 -*-
import torch
from torch import nn
from d2l import torch as d2l
import matplotlib.pyplot as plt

# 使用 nn.Sequential 搭建经典 LeNet 风格网络：
# 输入是 Fashion-MNIST 的灰度图，形状为 (N, 1, 28, 28)
# 网络结构：卷积 -> 激活 -> 池化 -> 卷积 -> 激活 -> 池化 -> 展平 -> 全连接分类
net = nn.Sequential(
    # 第一层卷积：输入通道 1，输出通道 6，卷积核 5x5，padding=2 保持空间尺寸不变（28x28 -> 28x28）
    nn.Conv2d(1, 6, kernel_size=5, padding=2), nn.Sigmoid(),
    # 平均池化：窗口 2x2，步幅 2，尺寸减半（28x28 -> 14x14）
    nn.AvgPool2d(kernel_size=2, stride=2),
    # 第二层卷积：输入通道 6，输出通道 16，卷积核 5x5，不加 padding（14x14 -> 10x10）
    nn.Conv2d(6, 16, kernel_size=5), nn.Sigmoid(),
    # 第二次平均池化：尺寸再次减半（10x10 -> 5x5）
    nn.AvgPool2d(kernel_size=2, stride=2),
    # 展平为二维张量，供全连接层使用：(N, 16, 5, 5) -> (N, 16*5*5)
    nn.Flatten(),
    # 三层全连接：分类头，最终输出 10 类 logits
    nn.Linear(16 * 5 * 5, 120), nn.Sigmoid(),
    nn.Linear(120, 84), nn.Sigmoid(),
    nn.Linear(84, 10))

# 用随机输入做一次前向传播，仅打印每一层输入与输出的形状
X_demo = torch.rand(size=(1, 1, 28, 28), dtype=torch.float32)
for idx, layer in enumerate(net):
    X_in = X_demo
    X_out = layer(X_in)
    print(f'layer {idx + 1}: {layer.__class__.__name__}')
    print('input shape:\t', X_in.shape)
    print('output shape:\t', X_out.shape)
    print('-' * 80)
    X_demo = X_out

# 准备数据集迭代器
# batch_size 越大，吞吐通常越高，但会占用更多显存
batch_size = 256
train_iter, test_iter = d2l.load_data_fashion_mnist(batch_size = batch_size)

def evaluate_accuracy_gpu(net, data_iter, device=None): #@save
    """使用GPU计算模型在数据集上的精度"""
    if isinstance(net, nn.Module):
        net.eval()  # 设置为评估模式
        if not device:
            device = next(iter(net.parameters())).device
    # 正确预测的数量，总预测的数量
    metric = d2l.Accumulator(2)
    with torch.no_grad():
        for X, y in data_iter:
            if isinstance(X, list):
                # BERT微调所需的（之后将介绍）
                X = [x.to(device) for x in X]
            else:
                X = X.to(device)
            y = y.to(device)
            metric.add(d2l.accuracy(net(X), y), y.numel())
    return metric[0] / metric[1]

#@save
def train_ch6(net, train_iter, test_iter, num_epochs, lr, device):
    """用GPU训练模型(在第六章定义)"""
    # 参数初始化：对线性层与卷积层采用 Xavier 均匀分布初始化
    # 该初始化有助于在训练初期保持较稳定的信号方差，缓解梯度消失/爆炸
    def init_weights(m):
        if type(m) == nn.Linear or type(m) == nn.Conv2d:
            nn.init.xavier_uniform_(m.weight)
    net.apply(init_weights)

    # 将模型搬到指定设备（GPU 或 CPU）
    print('training on', device)
    net.to(device)

    # 优化器与损失函数：
    # - SGD：随机梯度下降
    # - CrossEntropyLoss：多分类交叉熵（输入应为未归一化 logits）
    optimizer = torch.optim.SGD(net.parameters(), lr=lr)
    loss = nn.CrossEntropyLoss()

    # timer 用于统计训练吞吐，num_batches 表示每个 epoch 的批次数
    timer, num_batches = d2l.Timer(), len(train_iter)
    train_loss_history, train_acc_history, test_acc_history = [], [], []
    for epoch in range(num_epochs):
        # 训练损失之和，训练准确率之和，样本数
        metric = d2l.Accumulator(3)
        net.train()
        for i, (X, y) in enumerate(train_iter):
            timer.start()
            # 梯度清零（PyTorch 默认梯度累加）
            optimizer.zero_grad()

            # 将当前批次数据放到同一设备上
            X, y = X.to(device), y.to(device)

            # 前向传播 -> 计算损失 -> 反向传播 -> 参数更新
            y_hat = net(X)
            l = loss(y_hat, y)
            l.backward()
            optimizer.step()

            # 在不记录梯度的模式下累积统计量，降低额外开销
            with torch.no_grad():
                # l * batch_size：把“平均损失”恢复成“总损失”，便于后续按样本数求平均
                metric.add(l * X.shape[0], d2l.accuracy(y_hat, y), X.shape[0])
            timer.stop()
            train_l = metric[0] / metric[2]
            train_acc = metric[1] / metric[2]

        # 每个 epoch 结束后在测试集上评估一次
        test_acc = evaluate_accuracy_gpu(net, test_iter)
        train_loss_history.append(float(train_l))
        train_acc_history.append(float(train_acc))
        test_acc_history.append(float(test_acc))

    # 训练结束后输出最终指标与吞吐信息
    print(f'loss {train_l:.3f}, train acc {train_acc:.3f}, '
          f'test acc {test_acc:.3f}')
    print(f'{metric[2] * num_epochs / timer.sum():.1f} examples/sec '
          f'on {str(device)}')
    epochs = list(range(1, num_epochs + 1))
    plt.figure(figsize=(8, 5))
    plt.plot(epochs, train_loss_history, label='train loss')
    plt.plot(epochs, train_acc_history, label='train acc')
    plt.plot(epochs, test_acc_history, label='test acc')
    plt.xlabel('epoch')
    plt.title('Training Curve')
    plt.legend()
    plt.grid(True, linestyle='--', alpha=0.4)
    plt.tight_layout()
    curve_path = '/home/lxg/code/AI/code/20260331/training_curve.png'
    plt.savefig(curve_path, dpi=150)
    plt.close()
    print('training curve saved to:', curve_path)
    for name, param in net.named_parameters():
        print(f'{name}: shape={tuple(param.shape)}')
        print(param.detach().cpu().flatten()[:10])
    
# 训练超参数：学习率与训练轮数
lr, num_epochs = 0.9, 10
# 启动训练，d2l.try_gpu() 会优先选择可用 GPU，否则退回 CPU
train_ch6(net, train_iter, test_iter, num_epochs, lr, d2l.try_gpu())

# 使用 nn.Sequential 搭建经典 LeNet 风格网络：
# 输入是 Fashion-MNIST 的灰度图，形状为 (N, 1, 28, 28)
# 网络结构：卷积 -> 激活 -> 池化 -> 卷积 -> 激活 -> 池化 -> 展平 -> 全连接分类
net = nn.Sequential(
    # 第一层卷积：输入通道 1，输出通道 6，卷积核 5x5，padding=2 保持空间尺寸不变（28x28 -> 28x28）
    nn.Conv2d(1, 6, kernel_size=5, padding=2), nn.Sigmoid(),
    # 平均池化：窗口 2x2，步幅 2，尺寸减半（28x28 -> 14x14）
    nn.AvgPool2d(kernel_size=2, stride=2),
    # 第二层卷积：输入通道 6，输出通道 16，卷积核 5x5，不加 padding（14x14 -> 10x10）
    nn.Conv2d(6, 16, kernel_size=5), nn.Sigmoid(),
    # 第二次平均池化：尺寸再次减半（10x10 -> 5x5）
    nn.AvgPool2d(kernel_size=2, stride=2),
    # 展平为二维张量，供全连接层使用：(N, 16, 5, 5) -> (N, 16*5*5)
    nn.Flatten(),
    # 三层全连接：分类头，最终输出 10 类 logits
    nn.Linear(16 * 5 * 5, 120), nn.Sigmoid(),
    nn.Linear(120, 84), nn.Sigmoid(),
    nn.Linear(84, 10))