DAY 41 简单CNN

1. 数据增强
2. 卷积神经网络定义的写法
3. batch归一化：调整一个批次的分布，常用与图像数据
4. 特征图：只有卷积操作输出的才叫特征图
5. 调度器：直接修改基础学习率

卷积操作常见流程如下：

1. 输入 → 卷积层 → Batch归一化层（可选） → 池化层 → 激活函数 → 下一层

2.Flatten -> Dense (with Dropout，可选) -> Dense (Output)

作业：尝试手动修改下不同的调度器和CNN的结构，观察训练的差异。

（一）数据增强

在图像数据预处理环节，为提升数据多样性，可采用数据增强（数据增广）策略。该策略通常不改变单次训练的样本总数，而是通过对现有图像进行多样化变换，使每次训练输入的样本呈现更丰富的形态差异，从而有效扩展模型训练的样本空间多样性。

常见的修改策略包括以下几类：

1. 几何变换：如旋转、缩放、平移、剪裁、裁剪、翻转

2. 像素变换：如修改颜色、亮度、对比度、饱和度、色相、高斯模糊（模拟对焦失败）、增加噪声、马赛克

3. 语义增强（暂时不用）：mixup，对图像进行结构性改造、cutout随机遮挡等

此外，在数据极少的场景长，常常用生成模型来扩充数据集，如GAN、VAE等。

注意：数据增强一般是不改变每个批次的数据量，是对原始数据修改后替换原始数据。其中该数据集事先知道其均值和标准差，如果不知道，需要提前计算。

（二）cnn模型

卷积的本质：通过卷积核在输入通道上的滑动乘积，提取跨通道的空间特征。只需要定义几个参数即可

1. 卷积核大小：卷积核的大小，如3x3、5x5、7x7等。

2. 输入通道数：输入图片的通道数，如1（单通道图片）、3（RGB图片）、4（RGBA图片）等。

3. 输出通道数：卷积核的个数，即输出的通道数。如本模型中通过 32→64→128 逐步增加特征复杂度

4. 步长（stride）：卷积核的滑动步长，默认为1。

（三）batch归一化

批归一化是一种通过对每个小批量数据进行标准化来加速深度网络训练的技术。

批归一化的主要步骤:

1. 计算批次的均值 μ = mean(x)

2. 计算批次的方差 σ² = var(x)

3. 标准化: x̂ = (x - μ) / √(σ² + ε) (ε是小的常数，防止除以0)

4. 缩放和平移: y = γx̂ + β (γ和β是可学习的参数)

深度学习的归一化有2类：

1. Batch Normalization：一般用于图像数据，因为图像数据通常是批量处理，有相对固定的 Batch Size ，能利用 Batch 内数据计算稳定的统计量（均值、方差 ）来做归一化。

2. Layer Normalization：一般用于文本数据，本数据的序列长度往往不同，像不同句子长短不一，很难像图像那样固定 Batch Size 。如果用 Batch 归一化，不同批次的统计量波动大，效果不好。层归一化是对单个样本的所有隐藏单元进行归一化，不依赖批次。

（四）特征图和调度器

特征图是卷积神经网络中卷积层的输出结果，也称为激活图(activation map)。

关键点:

• 每个卷积核会产生一个特征图

• 特征图的深度(通道数)等于该层卷积核的数量

• 特征图的空间尺寸由输入尺寸、卷积核大小、步长和填充决定

• 随着网络深度增加，特征图会逐渐变小(空间上)，但会变深(通道数增加)

• 浅层特征图通常捕捉低级特征(边缘、纹理)，深层特征图捕捉高级语义特征

特征图就代表着在之前特征提取器上提取到的特征，可以通过 Grad-CAM方法来查看模型在识别图像时，特征图所对应的权重是多少。

调度器通过直接修改优化器中的基础学习率，实现了学习率的动态调整，有助于模型更好地收敛。

ReduceLROnPlateau调度器适用于当监测的指标（如验证损失）停滞时降低学习率。是大多数任务的首选调度器，尤其适合验证集波动较大的情况。

常见的优化器有 adam、SGD、RMSprop 等，而除此之外学习率调度器有 lr_scheduler.StepLR、lr_scheduler.ExponentialLR、lr_scheduler.CosineAnnealingLR 等。

import torch
import torch.nn as nn
import torch.optim as optim
from torchvision import datasets, transforms
from torch.utils.data import DataLoader
import matplotlib.pyplot as plt
import numpy as np
 
# 设置中文字体支持
plt.rcParams["font.family"] = ["SimHei"]
plt.rcParams['axes.unicode_minus'] = False  # 解决负号显示问题
 
# 检查GPU是否可用
device = torch.device("cuda" if torch.cuda.is_available() else "cpu")
print(f"使用设备: {device}")
 
# 1. 数据预处理
# 训练集：使用多种数据增强方法提高模型泛化能力
train_transform = transforms.Compose([
    # 随机裁剪图像，从原图中随机截取32x32大小的区域
    transforms.RandomCrop(32, padding=4),
    # 随机水平翻转图像（概率0.5）
    transforms.RandomHorizontalFlip(),
    # 随机颜色抖动：亮度、对比度、饱和度和色调随机变化
    transforms.ColorJitter(brightness=0.2, contrast=0.2, saturation=0.2, hue=0.1),
    # 随机旋转图像（最大角度15度）
    transforms.RandomRotation(15),
    # 将PIL图像或numpy数组转换为张量
    transforms.ToTensor(),
    # 标准化处理：每个通道的均值和标准差，使数据分布更合理
    transforms.Normalize((0.4914, 0.4822, 0.4465), (0.2023, 0.1994, 0.2010))
])
 
# 测试集：仅进行必要的标准化，保持数据原始特性，标准化不损失数据信息，可还原
test_transform = transforms.Compose([
    transforms.ToTensor(),
    transforms.Normalize((0.4914, 0.4822, 0.4465), (0.2023, 0.1994, 0.2010))
])
 
# 2. 加载CIFAR-10数据集
train_dataset = datasets.CIFAR10(
    root='./data',
    train=True,
    download=True,
    transform=train_transform  # 使用增强后的预处理
)
 
test_dataset = datasets.CIFAR10(
    root='./data',
    train=False,
    transform=test_transform  # 测试集不使用增强
)
 
# 3. 创建数据加载器
batch_size = 64
train_loader = DataLoader(train_dataset, batch_size=batch_size, shuffle=True)
test_loader = DataLoader(test_dataset, batch_size=batch_size, shuffle=False)

# 4. 定义CNN模型的定义（替代原MLP）
class CNN(nn.Module):
    def __init__(self):
        super(CNN, self).__init__()
        
        # 第一个卷积块保持不变
        self.conv1 = nn.Conv2d(3, 32, 3, padding=1)
        self.bn1 = nn.BatchNorm2d(32)
        self.relu1 = nn.ReLU()
        self.pool1 = nn.MaxPool2d(2, 2)
        
        # 第二个卷积块保持不变
        self.conv2 = nn.Conv2d(32, 64, 3, padding=1)
        self.bn2 = nn.BatchNorm2d(64)
        self.relu2 = nn.ReLU()
        self.pool2 = nn.MaxPool2d(2)
        
        # 第三个卷积块保持不变
        self.conv3 = nn.Conv2d(64, 128, 3, padding=1)
        self.bn3 = nn.BatchNorm2d(128)
        self.relu3 = nn.ReLU()
        self.pool3 = nn.MaxPool2d(2)
        
        # 调整全连接层结构
        self.fc1 = nn.Linear(128 * 4 * 4, 512)
        self.bn_fc1 = nn.BatchNorm1d(512)  # 新增全连接层的BatchNorm
        self.dropout = nn.Dropout(p=0.3)   # 降低Dropout率从0.5到0.3
        self.fc2 = nn.Linear(512, 10)
 
    def forward(self, x):
        # 卷积块处理保持不变
        x = self.pool1(self.relu1(self.bn1(self.conv1(x))))
        x = self.pool2(self.relu2(self.bn2(self.conv2(x))))
        x = self.pool3(self.relu3(self.bn3(self.conv3(x))))
        
        # 展平与全连接层调整
        x = x.view(-1, 128 * 4 * 4)
        x = self.fc1(x)
        x = self.bn_fc1(x)  # 新增全连接层的BatchNorm
        x = self.relu3(x)
        x = self.dropout(x)
        x = self.fc2(x)
        
        return x
 
 
 
# 初始化模型
model = CNN()
model = model.to(device)  # 将模型移至GPU（如果可用）
 
criterion = nn.CrossEntropyLoss()
optimizer = optim.Adam(model.parameters(), lr=0.001)
 
# 替换为CosineAnnealingLR调度器
scheduler = optim.lr_scheduler.CosineAnnealingLR(
    optimizer,        # 指定要控制的优化器
    T_max=10,        # 余弦周期长度(epoch数)
    eta_min=1e-5     # 最小学习率下限
)

# 5. 训练模型（记录每个 iteration 的损失）
def train(model, train_loader, test_loader, criterion, optimizer, scheduler, device, epochs):
    model.train()  # 设置为训练模式
    
    # 记录每个 iteration 的损失
    all_iter_losses = []  # 存储所有 batch 的损失
    iter_indices = []     # 存储 iteration 序号
    
    # 记录每个 epoch 的准确率和损失
    train_acc_history = []
    test_acc_history = []
    train_loss_history = []
    test_loss_history = []
    
    for epoch in range(epochs):
        running_loss = 0.0
        correct = 0
        total = 0
        
        for batch_idx, (data, target) in enumerate(train_loader):
            data, target = data.to(device), target.to(device)  # 移至GPU
            
            optimizer.zero_grad()  # 梯度清零
            output = model(data)  # 前向传播
            loss = criterion(output, target)  # 计算损失
            loss.backward()  # 反向传播
            optimizer.step()  # 更新参数
            
            # 记录当前 iteration 的损失
            iter_loss = loss.item()
            all_iter_losses.append(iter_loss)
            iter_indices.append(epoch * len(train_loader) + batch_idx + 1)
            
            # 统计准确率和损失
            running_loss += iter_loss
            _, predicted = output.max(1)
            total += target.size(0)
            correct += predicted.eq(target).sum().item()
            
            # 每100个批次打印一次训练信息
            if (batch_idx + 1) % 100 == 0:
                print(f'Epoch: {epoch+1}/{epochs} | Batch: {batch_idx+1}/{len(train_loader)} '
                      f'| 单Batch损失: {iter_loss:.4f} | 累计平均损失: {running_loss/(batch_idx+1):.4f}')
        
        # 计算当前epoch的平均训练损失和准确率
        epoch_train_loss = running_loss / len(train_loader)
        epoch_train_acc = 100. * correct / total
        train_acc_history.append(epoch_train_acc)
        train_loss_history.append(epoch_train_loss)
        
        # 测试阶段
        model.eval()  # 设置为评估模式
        test_loss = 0
        correct_test = 0
        total_test = 0
        
        with torch.no_grad():
            for data, target in test_loader:
                data, target = data.to(device), target.to(device)
                output = model(data)
                test_loss += criterion(output, target).item()
                _, predicted = output.max(1)
                total_test += target.size(0)
                correct_test += predicted.eq(target).sum().item()
        
        epoch_test_loss = test_loss / len(test_loader)
        epoch_test_acc = 100. * correct_test / total_test
        test_acc_history.append(epoch_test_acc)
        test_loss_history.append(epoch_test_loss)
        
        # 更新学习率调度器
        scheduler.step(epoch_test_loss)
        
        print(f'Epoch {epoch+1}/{epochs} 完成 | 训练准确率: {epoch_train_acc:.2f}% | 测试准确率: {epoch_test_acc:.2f}%')
    
    # 绘制所有 iteration 的损失曲线
    plot_iter_losses(all_iter_losses, iter_indices)
    
    # 绘制每个 epoch 的准确率和损失曲线
    plot_epoch_metrics(train_acc_history, test_acc_history, train_loss_history, test_loss_history)
    
    return epoch_test_acc  # 返回最终测试准确率
 
# 6. 绘制每个 iteration 的损失曲线
def plot_iter_losses(losses, indices):
    plt.figure(figsize=(10, 4))
    plt.plot(indices, losses, 'b-', alpha=0.7, label='Iteration Loss')
    plt.xlabel('Iteration（Batch序号）')
    plt.ylabel('损失值')
    plt.title('每个 Iteration 的训练损失')
    plt.legend()
    plt.grid(True)
    plt.tight_layout()
    plt.show()
 
# 7. 绘制每个 epoch 的准确率和损失曲线
def plot_epoch_metrics(train_acc, test_acc, train_loss, test_loss):
    epochs = range(1, len(train_acc) + 1)
    
    plt.figure(figsize=(12, 4))
    
    # 绘制准确率曲线
    plt.subplot(1, 2, 1)
    plt.plot(epochs, train_acc, 'b-', label='训练准确率')
    plt.plot(epochs, test_acc, 'r-', label='测试准确率')
    plt.xlabel('Epoch')
    plt.ylabel('准确率 (%)')
    plt.title('训练和测试准确率')
    plt.legend()
    plt.grid(True)
    
    # 绘制损失曲线
    plt.subplot(1, 2, 2)
    plt.plot(epochs, train_loss, 'b-', label='训练损失')
    plt.plot(epochs, test_loss, 'r-', label='测试损失')
    plt.xlabel('Epoch')
    plt.ylabel('损失值')
    plt.title('训练和测试损失')
    plt.legend()
    plt.grid(True)
    
    plt.tight_layout()
    plt.show()
 
# 8. 执行训练和测试
epochs = 20  # 增加训练轮次以获得更好效果
print("开始使用CNN训练模型...")
final_accuracy = train(model, train_loader, test_loader, criterion, optimizer, scheduler, device, epochs)
print(f"训练完成！最终测试准确率: {final_accuracy:.2f}%")
 
# 保存模型
torch.save(model.state_dict(), 'cifar10_cnn_model.pth')
print("模型已保存为: cifar10_cnn_model.pth")