1、引言
在本篇文章中，我们将深入探讨并实现一些现代卷积神经网络（CNN）架构的变体。近年来，学界提出了众多新颖的网络架构。其中一些最具影响力，并且至今仍然具有重要地位的架构包括：GoogleNet/Inception架构（2014年ILSVRC竞赛的冠军），ResNet（2015年ILSVRC竞赛的冠军），以及DenseNet（2017年CVPR会议的最佳论文奖得主）。这些网络在刚提出时均代表了当时技术的最前沿，它们的核心理念已成为当前大多数顶尖架构的基石。因此，深入理解这些架构的细节，并掌握其实现方法，对于我们来说至关重要。
首先，我们来导入一些常用的库。
## 标准库
import os # 导入操作系统接口
import numpy as np # 导入科学计算库
import random # 导入随机数生成库
from PIL import Image # 从PIL库导入图像处理模块
from types import SimpleNamespace # 导入命名空间类型
## 绘图相关导入
import matplotlib.pyplot as plt # 导入matplotlib的绘图模块
%matplotlib inline # 使matplotlib绘图在Jupyter Notebook中内联显示
from IPython.display import set_matplotlib_formats # 导入设置matplotlib格式的函数
set_matplotlib_formats('svg', 'pdf') # 设置matplotlib输出格式为SVG和PDF，便于导出
import matplotlib # 导入matplotlib库
matplotlib.rcParams['lines.linewidth'] = 2.0 # 设置matplotlib线条宽度为2.0
import seaborn as sns # 导入seaborn库，用于数据可视化
sns.reset_orig() # 重置seaborn的默认设置
## PyTorch
import torch # 导入PyTorch库
import torch.nn as nn # 导入PyTorch的神经网络模块
import torch.utils.data as data # 导入PyTorch的数据工具模块
import torch.optim as optim # 导入PyTorch的优化器模块
# Torchvision
import torchvision # 导入Torchvision库，用于处理图像和视频
from torchvision.datasets import CIFAR10 # 从Torchvision的datasets模块导入CIFAR-10数据集
from torchvision import transforms # 从Torchvision导入图像转换模块
在本文中，我们将沿用之前博文中使用的set_seed函数，并利用DATASET_PATH和CHECKPOINT_PATH这两个路径变量。根据实际情况，你可能需要对这些路径进行相应的调整。这样做可以确保我们的代码在不同环境中都能正确地找到数据集和模型检查点。
# 定义数据集存储路径，例如CIFAR-10数据集的下载位置
DATASET_PATH = "../data"
# 定义预训练模型的保存路径
CHECKPOINT_PATH = "../saved_models/tutorial5"
# 设置随机种子的函数，确保实验结果的可复现性
def set_seed(seed):
random.seed(seed) # 设置Python内置随机数生成器的种子
np.random.seed(seed) # 设置NumPy随机数生成器的种子
torch.manual_seed(seed) # 设置PyTorch随机数生成器的种子
if torch.cuda.is_available(): # 如果CUDA可用
torch.cuda.manual_seed(seed) # 设置CUDA随机数生成器的种子
torch.cuda.manual_seed_all(seed) # 设置CUDA所有设备的随机数生成器种子
# 调用set_seed函数，传入种子值42
set_seed(42)
# 确保在GPU上的所有操作都是确定性的（如果使用GPU），以提高结果的可复现性
torch.backends.cudnn.deterministic = True # 设置为True，确保CuDNN的算法是确定性的
torch.backends.cudnn.benchmark = False # 设置为False，关闭CuDNN的自动调优功能
# 根据CUDA是否可用，选择使用GPU或CPU
device = torch.device("cuda:0") if torch.cuda.is_available() else torch.device("cpu")
在本文中，我们提供了预先训练好的模型以及TensorBoard日志文件（稍后会对这些内容进行详细说明）。你可以通过以下链接下载这些资源。这将帮助你更快速地开始实验，同时确保你能够直观地查看训练过程中的各种指标变化。
import urllib.request # 导入urllib库的request模块，用于处理网络请求
from urllib.error import HTTPError # 导入HTTPError，用于捕获HTTP请求错误
# 定义GitHub仓库中存储预训练模型的URL
base_url = "https://raw.githubusercontent.com/phlippe/saved_models/main/tutorial5/"
# 定义需要下载的文件列表
pretrained_files = ["GoogleNet.ckpt", "ResNet.ckpt", "ResNetPreAct.ckpt", "DenseNet.ckpt",
"tensorboards/GoogleNet/events.out.tfevents.googlenet",
"tensorboards/ResNet/events.out.tfevents.resnet",
"tensorboards/ResNetPreAct/events.out.tfevents.resnetpreact",
"tensorboards/DenseNet/events.out.tfevents.densenet"]
# 如果检查点路径不存在，则创建它
os.makedirs(CHECKPOINT_PATH, exist_ok=True)
# 对于列表中的每个文件，检查它是否已经存在。如果不存在，尝试下载它。
for file_name in pretrained_files:
file_path = os.path.join(CHECKPOINT_PATH, file_name) # 构建文件的完整路径
if "/" in file_name: # 如果文件名中包含"/"，则需要创建相应的目录结构
os.makedirs(file_path.rsplit("/", 1)[0], exist_ok=True) # 创建目录
if not os.path.isfile(file_path): # 检查文件是否已经存在
file_url = base_url + file_name # 构建文件的下载URL
print(f"正在下载 {file_url}...")
try:
urllib.request.urlretrieve(file_url, file_path) # 尝试下载文件
except HTTPError as e: # 捕获并处理HTTP请求错误
print("下载过程中出现问题。请尝试从Google Drive文件夹下载文件，或将完整的输出信息，包括以下错误，联系作者：\n", e)
在本文中，我们将对CIFAR-10数据集进行模型的训练与评估。这样，你便可以将本教程中得到的结果与你在首次作业中实现的模型结果进行对比。根据我们在上一个教程中学到的初始化知识，数据经过预处理使其均值归零是至关重要的。因此，我们首先需要计算CIFAR数据集的均值和标准差。
# 导入CIFAR-10数据集，设置数据集的根目录为DATASET_PATH，指定为训练集，并自动下载
train_dataset = CIFAR10(root=DATASET_PATH, train=True, download=True)
# 计算数据集的均值，将数据归一化到[0,1]区间后，沿所有图像通道计算平均值
DATA_MEANS = (train_dataset.data / 255.0).mean(axis=(0, 1, 2))
# 计算数据集的标准差，同样先归一化到[0,1]区间，然后沿所有图像通道计算标准差
DATA_STD = (train_dataset.data / 255.0).std(axis=(0, 1, 2))
# 打印数据集的均值和标准差
print("数据均值:", DATA_MEANS)
print("数据标准差:", DATA_STD)
在本文中，我们将使用这些统计数据来构建一个transforms.Normalize模块，以便对数据进行适当的标准化处理。此外，为了提高模型的泛化能力并减少过拟合的风险，我们在训练过程中引入了数据增强技术。具体来说，我们会应用两种数据增强方法。
首先，我们将对每张图像以50%的概率进行随机的水平翻转，这是通过transforms.RandomHorizontalFlip实现的。一般来说，图像的水平翻转不会影响其类别识别，因为物体的类别与图像的水平方向无关。然而，如果我们的目标是图像中的数字或字母识别，那么方向性就变得重要了。
第二种数据增强方法是transforms.RandomResizedCrop，它通过在一定范围内随机裁剪图像，可能改变图像的纵横比，然后再将其缩放回原始尺寸。这样，尽管图像的像素值发生了变化，但图像的内容和整体语义信息仍然保持一致。
接下来，我们会将训练数据集随机划分为训练子集和验证子集。验证子集将用于评估模型在训练过程中的性能，以便于我们实施早停策略（early stopping）。训练结束后，我们将在CIFAR的测试集上对模型进行评估，以测试其对未知数据的处理能力。
# 定义测试集的转换操作，包括转换为张量以及标准化
test_transform = transforms.Compose([
transforms.ToTensor(), # 将PIL图像或Numpy数组转换为`FloatTensor`，并将数值范围从[0, 255]缩放到[0.0, 1.0]
transforms.Normalize(DATA_MEANS, DATA_STD) # 标准化，使数据具有指定的均值和标准差
])
# 定义训练集的转换操作，包括数据增强和标准化
train_transform = transforms.Compose([
transforms.RandomHorizontalFlip(), # 以50%的概率随机水平翻转图像
transforms.RandomResizedCrop((32, 32), scale=(0.8, 1.0), ratio=(0.9, 1.1)), # 随机裁剪并缩放图像
transforms.ToTensor(), # 转换为张量
transforms.Normalize(DATA_MEANS, DATA_STD) # 标准化
])
# 加载训练数据集，并将其划分为训练集和验证集，注意验证集不使用数据增强
train_dataset = CIFAR10(root=DATASET_PATH, train=True, transform=train_transform, download=True)
val_dataset = CIFAR10(root=DATASET_PATH, train=True, transform=test_transform, download=True)
# 设置随机种子以确保结果的可复现性
set_seed(42)
# 将训练数据集随机划分为45000张图像用于训练，5000张图像用于验证
train_set, _ = torch.utils.data.random_split(train_dataset, [45000, 5000])
set_seed(42)
# 将验证数据集随机划分，这里使用相同的划分策略，但实际上验证集不需要划分
_, val_set = torch.utils.data.random_split(val_dataset, [45000, 5000])
# 加载测试集
test_set = CIFAR10(root=DATASET_PATH, train=False, transform=test_transform, download=True)
# 定义数据加载器，用于后续的训练、验证和测试
train_loader = data.DataLoader(train_set, batch_size=128, shuffle=True, drop_last=True, pin_memory=True, num_workers=4)
val_loader = data.DataLoader(val_set, batch_size=128, shuffle=False, drop_last=False, num_workers=4)
test_loader = data.DataLoader(test_set, batch_size=128, shuffle=False, drop_last=False, num_workers=4)
为了确保我们的标准化处理达到预期效果，我们可以通过打印单个数据批次的均值和标准差来进行检验。理想情况下，每个颜色通道的均值应该趋近于0，标准差趋近于1。这是一种常用的做法，用以确保数据经过标准化处理后，分布更加接近标准正态分布，从而有助于模型的训练和收敛。
# 获取训练数据加载器的第一个批次的数据
imgs, _ = next(iter(train_loader))
# 打印该批次数据的均值，dim参数指定了要计算均值的维度
print("批次均值", imgs.mean(dim=[0, 2, 3]))
# 打印该批次数据的标准差，dim参数指定了要计算标准差的维度
print("批次标准差", imgs.std(dim=[0, 2, 3]))
接下来，我们将对训练集中的部分图像进行可视化展示，并观察它们在应用了随机数据增强技术之后的效果。这有助于我们直观地理解数据增强对图像的影响，以及它是如何帮助改善模型的泛化能力的。
# 定义要显示的图像数量
NUM_IMAGES = 4
# 从训练数据集中获取指定数量的图像
images = [train_dataset[idx][0] for idx in range(NUM_IMAGES)]
# 将原始图像数据转换为PIL图像格式
orig_images = [Image.fromarray(np.transpose(train_dataset.data[idx], (1, 2, 0))) for idx in range(NUM_IMAGES)]
# 应用测试转换操作到原始图像上
orig_images = [test_transform(img) for img in orig_images]
# 使用torchvision的make_grid函数创建图像网格，将原始图像和增强后的图像堆叠起来
# nrow指定每行显示的图像数量，normalize设置为True表示将像素值归一化到[0,1]，pad_value设置为0.5表示填充值
img_grid = torchvision.utils.make_grid(torch.stack(images + orig_images, dim=0), nrow=4, normalize=True, pad_value=0.5)
# permute函数用于重新排列图像张量的维度，以适应matplotlib的显示要求
img_grid = img_grid.permute(1, 2, 0)
# 使用matplotlib创建图像展示
plt.figure(figsize=(8, 8)) # 设置图像展示窗口的大小
plt.title("CIFAR10数据增强示例") # 设置图像展示的标题
plt.imshow(img_grid) # 显示图像网格
plt.axis('off') # 不显示坐标轴
plt.show() # 展示图像
plt.close() # 展示完毕后关闭图像展示窗口
2.使用PytTorch Linghtning
在本文和后续的文章中，我们将利用一个名为PyTorch Lightning的库。PyTorch Lightning是一个框架，它极大地简化了在PyTorch中编写训练、评估和测试模型的代码。它还负责将日志信息记录到TensorBoard——这是一个用于机器学习实验的可视化工具，并且能够自动保存模型的检查点，而我们几乎不需要编写额外的代码。这对于我们来说极为便利，因为我们更愿意将精力集中在不同模型架构的实现上，而不是花费大量时间处理其他代码问题。本文使用的是PyTorch Lightning1.8版本，请读者自行到官网更新最新的版本。
现在，我们将在PyTorch Lightning中迈出探索的第一步，并在后续的文章中继续深入了解这个框架。首先，我们需要导入这个库。
# 尝试导入PyTorch Lightning库
try:
import pytorch_lightning as pl
except ModuleNotFoundError: # 如果模块未找到异常，例如Google Colab默认没有安装PyTorch Lightning
# 使用pip命令静默安装PyTorch Lightning，版本要求大于等于1.5
!pip install --quiet pytorch-lightning>=1.5
# 再次尝试导入PyTorch Lightning库
import pytorch_lightning as pl
PyTorch Lightning框架内建了大量实用的功能，包括一个用于设定随机种子的便捷方法：
# 设置随机种子以确保实验的可重复性
pl.seed_everything(42)
在未来的工作中，我们将无需再自行定义设置随机种子的函数。
在PyTorch Lightning框架中，我们通过pl.LightningModule（继承自PyTorch的torch.nn.Module）来构建我们的模型，它将代码逻辑划分为五个核心部分：
- 初始化 (
__init__)：在这里，我们初始化所有必要的参数和模型结构。 - 配置优化器 (
configure_optimizers)：在这部分，我们定义模型的优化器、学习率调整策略等。 - 训练步骤 (
training_step)：我们仅需指定单个数据批次的损失计算方法，而优化器的梯度清零、损失反向传播和参数更新等步骤，以及日志记录或保存操作，都由框架在后台自动处理。 - 验证步骤 (
validation_step)：与训练类似，我们定义每个验证步骤需要进行的操作。 - 测试步骤(
test_step)：这与验证步骤类似，只不过是应用于测试数据集。
通过这种方式，PyTorch Lightning并没有简化PyTorch的代码，而是对其进行了有序组织，并提供了一些常用的默认操作。如果你需要对训练、验证或测试流程进行特定的调整，框架提供了多种可重写的方法来满足个性化需求（具体细节请参考官方文档）。
接下来，我们可以观察一个使用PyTorch Lightning构建的用于训练卷积神经网络的模块示例。
class CIFARModule(pl.LightningModule):
def __init__(self, model_name, model_hparams, optimizer_name, optimizer_hparams):
"""
构造函数初始化一个用于CIFAR数据集的模型模块。
参数:
model_name - 要运行的模型/CNN名称，用于创建模型。
model_hparams - 模型的超参数字典。
optimizer_name - 使用的优化器名称，目前支持Adam和SGD。
optimizer_hparams - 优化器的超参数字典，包括学习率、权重衰减等。
"""
super().__init__() # 调用父类构造函数
self.save_hyperparameters() # 将超参数导出到YAML文件，并创建"self.hparams"命名空间
# 创建模型，使用给定的模型名称和超参数
self.model = create_model(model_name, model_hparams)
# 创建损失模块
self.loss_module = nn.CrossEntropyLoss()
# 用于在Tensorboard中可视化图的示例输入
self.example_input_array = torch.zeros((1, 3, 32, 32), dtype=torch.float32)
def forward(self, imgs):
# 当可视化图时运行的前向函数
return self.model(imgs)
def configure_optimizers(self):
# 根据选择的优化器名称创建优化器
if self.hparams.optimizer_name == "Adam":
# 使用AdamW，即带有正确实现权重衰减的Adam（详见提供的链接）
optimizer = optim.AdamW(self.parameters(), **self.hparams.optimizer_hparams)
elif self.hparams.optimizer_name == "SGD":
optimizer = optim.SGD(self.parameters(), **self.hparams.optimizer_hparams)
else:
# 如果提供了未知的优化器名称，断言失败
assert False, f"Unknown optimizer: \"{self.hparams.optimizer_name}\""
# 学习率调度器，在第100和150个epoch后将学习率降低0.1
scheduler = optim.lr_scheduler.MultiStepLR(optimizer, milestones=[100, 150], gamma=0.1)
return [optimizer], [scheduler] # 返回优化器和学习率调度器
def training_step(self, batch, batch_idx):
# 训练步骤，对每个批次的数据执行
imgs, labels = batch
preds = self.model(imgs) # 模型预测
loss = self.loss_module(preds, labels) # 计算损失
acc = (preds.argmax(dim=-1) == labels).float().mean() # 计算准确率
# 在Tensorboard中记录每个epoch的准确率（跨批次的加权平均值）
self.log('train_acc', acc, on_step=False, on_epoch=True)
self.log('train_loss', loss)
return loss # 返回损失张量以调用".backward"
def validation_step(self, batch, batch_idx):
# 验证步骤，对验证数据执行
imgs, labels = batch
preds = self.model(imgs).argmax(dim=-1)
acc = (labels == preds).float().mean()
# 默认每个epoch记录一次（跨批次的加权平均值）
self.log('val_acc', acc)
def test_step(self, batch, batch_idx):
# 测试步骤，对测试数据执行
imgs, labels = batch
preds = self.model(imgs).argmax(dim=-1)
acc = (labels == preds).float().mean()
# 默认每个epoch记录一次（跨批次的加权平均值），然后返回
self.log('test_acc', acc)
代码的组织结构清晰有序，这有助于他人理解你的代码逻辑。
PyTorch Lightning框架中一个关键的概念是回调（callbacks）。回调函数是一些自包含的函数，它们包含了Lightning Module中非核心的逻辑。这些回调函数通常在训练周期结束后被调用，但它们也可能会影响到训练循环的其他环节。例如，我们将采用以下两个预定义的回调：LearningRateMonitor（学习率监控器）和ModelCheckpoint（模型检查点）。学习率监控器会将当前的学习率信息添加到TensorBoard中，这有助于我们确认学习率调度器是否按预期工作。模型检查点回调则允许你定制检查点的保存策略，比如保留多少个检查点、何时进行保存、根据哪个指标来决定保存等。下面是这些回调的导入代码：
# 回调函数导入
from pytorch_lightning.callbacks import LearningRateMonitor, ModelCheckpoint
为了能够使用同一个Lightning模块来运行多种不同的模型，我们下面定义了一个函数，它将模型名称映射到相应的模型类。目前，model_dict 字典为空，但我们将在接下来的笔记本使用过程中用新模型来填充它。
model_dict = {}
def create_model(model_name, model_hparams):
if model_name in model_dict:
return model_dict[model_name](**model_hparams)
else:
assert False, f"未知的模型名称\"{model_name}\"。可用的模型有：{str(model_dict.keys())}"
类似地，为了在我们的模型中将激活函数作为超参数使用，我们下面定义了一个将名称映射到函数的字典：
act_fn_by_name = {
"tanh": nn.Tanh,
"relu": nn.ReLU,
"leakyrelu": nn.LeakyReLU,
"gelu": nn.GELU
通过这种方式，代码的复用性和灵活性得以增强，同时也方便了不同模型和配置之间的切换，保持了代码的整洁和易于维护。如果我们直接将类或对象作为参数传递给Lightning模块，就无法享受PyTorch Lightning提供的自动保存和加载超参数的特性。
在PyTorch Lightning框架中，除了Lightning模块外，另一个核心组件是Trainer（训练器）。训练器的职责是执行Lightning模块中定义的训练步骤，并确保整个训练流程的完整性。与Lightning模块一样，你可以对任何你不想自动执行的关键部分进行覆盖，但通常情况下，默认设置就是最佳实践。有关全部功能的详细介绍，请查看官方文档。我们下面使用到的最重要的几个函数包括：
-
trainer.fit：接收一个Lightning模块、一个训练数据集，以及一个（可选的）验证数据集作为输入。此函数负责在训练数据集上训练指定的模块，并且可以定期进行验证（默认是每个epoch一次，但这个频率可以调整）。 -
trainer.test：接收一个模型和我们希望进行测试的数据集作为输入。它将返回该数据集上的测试指标。
在进行训练和测试时，我们无需担心如将模型设置为评估模式（model.eval()）等细节，因为这些操作都会自动完成。以下是我们如何定义模型的训练函数的示例：
def train_model(model_name, save_name=None, **kwargs):
"""
参数:
model_name - 要运行的模型名称，用于在"model_dict"中查找对应的类。
save_name (可选) - 如果提供，这个名称将被用于创建检查点和日志目录。
"""
if save_name是None：
save_name = model_name
# 创建一个PyTorch Lightning训练器，并配置生成回调
trainer = pl.Trainer(default_root_dir=os.path.join(CHECKPOINT_PATH, save_name), # 模型保存位置
accelerator="gpu" if str(device).startswith("cuda") else "cpu", # 尽可能在GPU上运行
devices=1, # 使用的GPU/CPU数量（笔记本示例中1足够了）
max_epochs=180, # 训练的最大周期数，若未设置早停则使用此值
callbacks=[ModelCheckpoint(save_weights_only=True, mode="max", monitor="val_acc"), # 基于最大val_acc记录保存最佳检查点，仅保存权重而非优化器状态
LearningRateMonitor("epoch")], # 每个epoch记录学习率
enable_progress_bar=True) # 是否显示进度条
trainer.logger._log_graph