简单CNN
知识回顾
- 数据增强
- 卷积神经网络定义的写法
- batch归一化：调整一个批次的分布，常用与图像数据
- 特征图：只有卷积操作输出的才叫特征图
- 调度器：直接修改基础学习率
卷积操作常见流程如下：
1. 输入 → 卷积层 → Batch归一化层（可选） → 池化层 → 激活函数 → 下一层
- Flatten -> Dense (with Dropout，可选) -> Dense (Output)
简单 CNN
1. 数据增强
数据增强就像是在餐厅的食材上应用各种处理方法，以增加菜品的多样性和丰富性。例如，可以通过旋转、翻转、裁剪等方法来扩充食材的种类和形态，使模型在训练过程中看到更多的“菜品”变化，从而提高模型的泛化能力。
2. 卷积神经网络定义的写法
卷积神经网络（CNN）的定义就像是餐厅的高级菜谱，其中包含了多种复杂的烹饪步骤和技巧。CNN 通过卷积层、池化层、激活函数等组件来提取图像的特征，从而实现对图像的分类或识别。
3. Batch 归一化
Batch 归一化就像是在厨房中对食材进行标准化处理，确保每一批次的食材在烹饪前都有相同的均值和方差。这有助于加快模型的训练速度，并提高模型的稳定性。
4. 特征图
特征图就像是厨师在烹饪过程中创建的中间产物，这些中间产物包含了食材经过初步处理后的信息。在 CNN 中，特征图是由卷积层生成的，它们捕捉了图像的不同特征，如边缘、纹理等。
5. 调度器
调度器就像是餐厅的经理，根据餐厅的运营情况（如顾客流量、食材供应等）动态调整厨师的工作强度和节奏。在训练模型时，调度器会根据训练进度动态调整学习率，以加快收敛速度并提高模型性能。
6. 卷积操作常见流程
卷积操作的常见流程可以比喻为餐厅制作一道复杂菜品的步骤：
-
输入：就像厨师接到顾客的订单，开始准备食材。
-
卷积层：厨师对食材进行初步处理，如切割、搅拌等，提取出食材的基本特征。
-
Batch 归一化层（可选）：厨师对处理后的食材进行标准化，确保每一批次的食材质量一致。
-
池化层：厨师对食材进行进一步处理，如浓缩、提纯等，减少数据量并突出重要特征。
-
激活函数：厨师根据菜谱添加特定的调料，使菜品具有独特的风味，激活函数为模型引入非线性。
-
下一层：处理后的食材传递给下一位厨师，进行下一步烹饪。
7. Flatten -> Dense (with Dropout，可选) -> Dense (Output)
-
Flatten：将多维的特征图展平成一维向量，类似于将复杂菜品的所有成分混合在一起，形成一个统一的混合物。
-
Dense (with Dropout)：对混合物进行进一步加工，添加 Dropout 就像是在烹饪过程中随机丢弃一些成分，以防止过度依赖某些特定成分（防止过拟合）。
-
Dense (Output)：最终输出菜品，类似于将所有成分组合成一个完整的菜品，呈现给顾客。
作业：尝试手动修改下不同的调度器和CNN的结构，观察训练的差异。
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__() # 继承父类初始化
# ---------------------- 第一个卷积块 ----------------------
# 卷积层1：输入3通道（RGB），输出32个特征图，卷积核3x3，边缘填充1像素
self.conv1 = nn.Conv2d(
in_channels=3, # 输入通道数（图像的RGB通道）
out_channels=32, # 输出通道数（生成32个新特征图）
kernel_size=3, # 卷积核尺寸（3x3像素）
padding=1 # 边缘填充1像素，保持输出尺寸与输入相同
)
# 批量归一化层：对32个输出通道进行归一化，加速训练
self.bn1 = nn.BatchNorm2d(num_features=32)
# ReLU激活函数：引入非线性，公式：max(0, x)
self.relu1 = nn.ReLU()
# 最大池化层：窗口2x2，步长2，特征图尺寸减半（32x32→16x16）
self.pool1 = nn.MaxPool2d(kernel_size=2, stride=2) # stride默认等于kernel_size
# ---------------------- 第二个卷积块 ----------------------
# 卷积层2：输入32通道（来自conv1的输出），输出64通道
self.conv2 = nn.Conv2d(
in_channels=32, # 输入通道数（前一层的输出通道数）
out_channels=64, # 输出通道数（特征图数量翻倍）
kernel_size=3, # 卷积核尺寸不变
padding=1 # 保持尺寸：16x16→16x16（卷积后）→8x8（池化后）
)
self.bn2 = nn.BatchNorm2d(num_features=64)
self.relu2 = nn.ReLU()
self.pool2 = nn.MaxPool2d(kernel_size=2) # 尺寸减半：16x16→8x8
# ---------------------- 第三个卷积块 ----------------------
# 卷积层3：输入64通道，输出128通道
self.conv3 = nn.Conv2d(
in_channels=64, # 输入通道数（前一层的输出通道数）
out_channels=128, # 输出通道数（特征图数量再次翻倍）
kernel_size=3,
padding=1 # 保持尺寸：8x8→8x8（卷积后）→4x4（池化后）
)
self.bn3 = nn.BatchNorm2d(num_features=128)
self.relu3 = nn.ReLU() # 复用激活函数对象（节省内存）
self.pool3 = nn.MaxPool2d(kernel_size=2) # 尺寸减半：8x8→4x4
# ---------------------- 全连接层（分类器） ----------------------
# 计算展平后的特征维度：128通道 × 4x4尺寸 = 128×16=2048维
self.fc1 = nn.Linear(
in_features=128 * 4 * 4, # 输入维度（卷积层输出的特征数）
out_features=512 # 输出维度（隐藏层神经元数）
)
# Dropout层：训练时随机丢弃50%神经元，防止过拟合
self.dropout = nn.Dropout(p=0.5)
# 输出层：将512维特征映射到10个类别（CIFAR-10的类别数）
self.fc2 = nn.Linear(in_features=512, out_features=10)
def forward(self, x):
# 输入尺寸：[batch_size, 3, 32, 32]（batch_size=批量大小，3=通道数，32x32=图像尺寸）
# ---------- 卷积块1处理 ----------
x = self.conv1(x) # 卷积后尺寸：[batch_size, 32, 32, 32]（padding=1保持尺寸）
x = self.bn1(x) # 批量归一化，不改变尺寸
x = self.relu1(x) # 激活函数，不改变尺寸
x = self.pool1(x) # 池化后尺寸：[batch_size, 32, 16, 16]（32→16是因为池化窗口2x2）
# ---------- 卷积块2处理 ----------
x = self.conv2(x) # 卷积后尺寸：[batch_size, 64, 16, 16]（padding=1保持尺寸）
x = self.bn2(x)
x = self.relu2(x)
x = self.pool2(x) # 池化后尺寸：[batch_size, 64, 8, 8]
# ---------- 卷积块3处理 ----------
x = self.conv3(x) # 卷积后尺寸：[batch_size, 128, 8, 8]（padding=1保持尺寸）
x = self.bn3(x)
x = self.relu3(x)
x = self.pool3(x) # 池化后尺寸：[batch_size, 128, 4, 4]
# ---------- 展平与全连接层 ----------
# 将多维特征图展平为一维向量：[batch_size, 128*4*4] = [batch_size, 2048]
x = x.view(-1, 128 * 4 * 4) # -1自动计算批量维度，保持批量大小不变
x = self.fc1(x) # 全连接层：2048→512，尺寸变为[batch_size, 512]
x = self.relu3(x) # 激活函数（复用relu3，与卷积块3共用）
x = self.dropout(x) # Dropout随机丢弃神经元，不改变尺寸
x = self.fc2(x) # 全连接层：512→10，尺寸变为[batch_size, 10]（未激活，直接输出logits）
return x # 输出未经过Softmax的logits，适用于交叉熵损失函数
# 初始化模型
model = CNN()
#model = model.to(device) # 将模型移至GPU（如果可用）
# 5. 定义损失函数和优化器
criterion = nn.CrossEntropyLoss() # 交叉熵损失函数，适用于多分类任务
optimizer = optim.Adam(model.parameters(), lr=0.001) # Adam优化器，学习率0.001
# 引入学习率调度器，在训练过程中动态调整学习率--训练初期使用较大的 LR 快速降低损失，训练后期使用较小的 LR 更精细地逼近全局最优解。
# 在每个 epoch 结束后，需要手动调用调度器来更新学习率，可以在训练过程中调用 scheduler.step()
scheduler = optim.lr_scheduler.ReduceLROnPlateau(
optimizer, # 指定要控制的优化器（这里是Adam）
mode='min', # 监测的指标是"最小化"（如损失函数）
patience=3, # 如果连续3个epoch指标没有改善，才降低LR
factor=0.5 # 降低LR的比例（新LR = 旧LR × 0.5）
)
# scheduler = optim.lr_scheduler.StepLR(optimizer, step_size=5, gamma=0.1)
# # 每5个epoch，LR = LR × 0.1
# scheduler = optim.lr_scheduler.MultiStepLR(optimizer, milestones=[10, 20, 30], gamma=0.5)
# # 当epoch=10、20、30时，LR = LR × 0.5
# scheduler = optim.lr_scheduler.CosineAnnealingLR(optimizer, T_max=10, eta_min=0.0001)
# # LR在[0.0001, LR_initial]之间按余弦曲线变化，周期为2×T_max
# 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")1. 数据预处理和增强
-
训练集数据增强：使用了随机裁剪、随机水平翻转、随机颜色抖动和随机旋转等操作来扩充训练数据的多样性，帮助模型更好地泛化。
-
测试集预处理：仅进行了标准化处理，以保持数据的一致性。
2. 数据加载
-
使用
DataLoader将 CIFAR-10 数据集加载为批次数据，方便模型训练和测试。
3. CNN 模型定义
-
卷积块：包括卷积层、批量归一化层、ReLU 激活函数和最大池化层，用于提取图像特征。
-
全连接层：将卷积层提取的特征展平后，通过全连接层进行分类，最后输出分类结果。
4. 模型训练
-
损失函数：使用交叉熵损失函数，适用于多分类任务。
-
优化器：使用 Adam 优化器，学习率为 0.001。
-
学习率调度器：使用
ReduceLROnPlateau调度器，根据验证损失动态调整学习率。
5. 训练和测试函数
-
训练函数：记录每个 iteration 的损失，并在每个 epoch 结束后计算训练和测试的准确率及损失。
-
测试函数：在测试集上评估模型的性能。
6. 结果可视化
-
绘制损失曲线：展示每个 iteration 的训练损失变化。
-
绘制准确率和损失曲线：展示每个 epoch 的训练和测试准确率及损失变化。
7. 执行训练和测试
-
设置训练轮次为 20，开始训练模型，并在训练结束后输出最终的测试准确率。
这段代码通过 CNN 模型在 CIFAR-10 数据集上实现了图像分类任务，包含了数据预处理、模型定义、训练、测试和结果可视化的完整流程。通过数据增强和学习率调度等技巧，提高了模型的泛化能力和训练效果。
@浙大疏锦行
