Python训练营打卡 Day44

预训练模型

知识点回顾：

1. 预训练的概念
2. 常见的分类预训练模型
3. 图像预训练模型的发展史
4. 预训练的策略
5. 预训练代码实战：resnet18

预训练模型

1. 预训练的概念：

   • 就像一家连锁餐厅，其厨师团队已经在总店经过了严格的培训（在大规模数据集上预训练），拥有了丰富的烹饪经验（模型权重）。当我们开设新店（新任务）时，可以直接聘请这些有经验的厨师，他们能够快速适应新店的菜单（新任务），减少培训时间和资源。

2. 常见的分类预训练模型：

   • ResNet：就像餐厅的招牌菜系，经过多年的改良和顾客反馈，成为餐厅的镇店之宝。它的残差连接（后面会详细解释）就像厨师的独家秘诀，让菜品的味道层次更加丰富。

   • VGG：类似于餐厅的经典菜系，结构简单但深入人心，适合各种顾客的口味。

   • Inception：相当于餐厅的创新菜系，通过多种烹饪方法的结合，提供独特的用餐体验。

   • MobileNet：像是餐厅的外卖服务，虽然资源有限，但依然能提供高质量的菜品，适合快速消费场景。

3. 图像预训练模型的发展史：

   • AlexNet：就像是一家在烹饪界引起轰动的新餐厅，首次展示了深度学习在图像识别上的强大能力。

   • VGGNet：这家餐厅不断深化其菜系的深度，通过增加菜品的复杂度（网络深度）来提升顾客满意度（模型性能）。

   • ResNet：引入了一种新的烹饪理念（残差连接），解决了深层网络训练中的梯度消失问题，让餐厅能够提供更复杂的菜品。

   • Inception：这家餐厅通过多尺度的烹饪方法（多尺度特征提取），不断创新菜品，提升顾客体验。

   • EfficientNet：系统地研究了如何在保持菜品质量的同时，提高厨房的工作效率（模型效率和性能）。

4. 预训练的策略：

   • 直接迁移：就像直接采用总店的招牌菜谱，在新店直接上菜，无需任何调整。

   • 微调（Fine-tuning）：相当于在总店菜谱的基础上，根据新店顾客的口味进行微调，让菜品更受欢迎。

   • 特征提取：使用总店厨房已经处理好的食材（预训练模型的特征），由新店的厨师制作新的菜品（训练新的分类器）。

   • 冻结部分层：在微调过程中，保持某些基础的烹饪步骤（冻结预训练模型的部分层）不变，只调整最后的摆盘（训练顶部的分类层）。

5. 预训练代码实战：ResNet18：

   • 就像是在新店中引入了ResNet18这道招牌菜的制作流程，并根据新店的顾客口味（CIFAR-10数据集）对最后的调味（分类层）进行了调整，使其更适合新店的菜单（新的分类任务）。

作业

1. 尝试在CIFAR-10对比如下其他的预训练模型，观察差异：

   • 就像在新店尝试不同的招牌菜，比较它们的受欢迎程度（模型性能）、制作时间（训练时间）和厨房资源占用（显存占用），找出最适合新店的菜品（模型）。

2. 尝试通过Ctrl进入ResNet的内部，观察残差究竟是什么：

   • 就像是走进厨房，仔细观察ResNet这道菜的制作过程，尤其是它的残差连接（一种将原材料直接加入半成品中的技巧），理解它如何保持食材的新鲜度（信息流动）和提升

1. import torch
   import torch.nn as nn
   import torch.optim as optim
   from torchvision import datasets, transforms, models
   from torch.utils.data import DataLoader
   import matplotlib.pyplot as plt
   import os# 设置中文字体支持
   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([transforms.RandomCrop(32, padding=4),transforms.RandomHorizontalFlip(),transforms.ColorJitter(brightness=0.2, contrast=0.2, saturation=0.2, hue=0.1),transforms.RandomRotation(15),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. 定义ResNet50模型
   def create_resnet50(pretrained=True, num_classes=10):model = models.resnet50(pretrained=pretrained)# 修改最后一层全连接层in_features = model.fc.in_featuresmodel.fc = nn.Linear(in_features, num_classes)return model.to(device)# 5. 冻结/解冻模型层的函数
   def freeze_model(model, freeze=True):"""冻结或解冻模型的卷积层参数"""# 冻结/解冻除fc层外的所有参数for name, param in model.named_parameters():if 'fc' not in name:param.requires_grad = not freeze# 打印冻结状态frozen_params = sum(p.numel() for p in model.parameters() if not p.requires_grad)total_params = sum(p.numel() for p in model.parameters())if freeze:print(f"已冻结模型卷积层参数 ({frozen_params}/{total_params} 参数)")else:print(f"已解冻模型所有参数 ({total_params}/{total_params} 参数可训练)")return model# 6. 训练函数（支持阶段式训练和早停）
   def train_with_freeze_schedule(model, train_loader, test_loader, criterion, optimizer, scheduler, device, epochs, freeze_epochs=5, early_stop_patience=5):"""前freeze_epochs轮冻结卷积层，之后解冻所有层进行训练当测试集准确率连续early_stop_patience轮未提升时触发早停"""train_loss_history = []test_loss_history = []train_acc_history = []test_acc_history = []all_iter_losses = []iter_indices = []# 早停相关变量best_accuracy = 0.0best_epoch = 0early_stop_counter = 0should_stop_early = False# 初始冻结卷积层if freeze_epochs > 0:model = freeze_model(model, freeze=True)for epoch in range(epochs):# 早停检查if should_stop_early:print(f"触发早停: 在第 {best_epoch} 轮后测试准确率没有提升")break# 解冻控制：在指定轮次后解冻所有层if epoch == freeze_epochs:model = freeze_model(model, freeze=False)# 解冻后调整优化器（可选）optimizer.param_groups[0]['lr'] = 1e-4  # 降低学习率防止过拟合model.train()  # 设置为训练模式running_loss = 0.0correct_train = 0total_train = 0for batch_idx, (data, target) in enumerate(train_loader):data, target = data.to(device), target.to(device)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_train += target.size(0)correct_train += 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}")# 计算 epoch 级指标epoch_train_loss = running_loss / len(train_loader)epoch_train_acc = 100. * correct_train / total_train# 测试阶段model.eval()correct_test = 0total_test = 0test_loss = 0.0with 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# 记录历史数据train_loss_history.append(epoch_train_loss)test_loss_history.append(epoch_test_loss)train_acc_history.append(epoch_train_acc)test_acc_history.append(epoch_test_acc)# 更新学习率调度器if scheduler is not None:scheduler.step(epoch_test_loss)# 早停逻辑if epoch_test_acc > best_accuracy:best_accuracy = epoch_test_accbest_epoch = epoch + 1early_stop_counter = 0print(f"保存最佳模型: 第 {best_epoch} 轮, 准确率 {best_accuracy:.2f}%")else:early_stop_counter += 1print(f"早停计数器: {early_stop_counter}/{early_stop_patience}")if early_stop_counter >= early_stop_patience:should_stop_early = True# 打印 epoch 结果print(f"Epoch {epoch+1} 完成 | 训练损失: {epoch_train_loss:.4f} "f"| 训练准确率: {epoch_train_acc:.2f}% | 测试准确率: {epoch_test_acc:.2f}%")# 绘制损失和准确率曲线plot_iter_losses(all_iter_losses, iter_indices)plot_epoch_metrics(train_acc_history, test_acc_history, train_loss_history, test_loss_history)return best_accuracy, best_epoch  # 返回最佳测试准确率和对应的轮次# 7. 绘制Iteration损失曲线
   def plot_iter_losses(losses, indices):plt.figure(figsize=(10, 4))plt.plot(indices, losses, 'b-', alpha=0.7)plt.xlabel('Iteration（Batch序号）')plt.ylabel('损失值')plt.title('训练过程中的Iteration损失变化')plt.grid(True)plt.show()# 8. 绘制Epoch级指标曲线
   def plot_epoch_metrics(train_acc, test_acc, train_loss, test_loss):epochs = range(1, len(train_acc) + 1)plt.figure(figsize=(12, 5))# 准确率曲线plt.subplot(1, 2, 1)plt.plot(epochs, train_acc, 'b-', label='训练准确率')plt.plot(epochs, test_acc, 'r-', label='测试准确率')plt.axvline(x=5, color='g', linestyle='--', label='解冻卷积层')plt.xlabel('Epoch')plt.ylabel('准确率 (%)')plt.title('准确率随Epoch变化')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.axvline(x=5, color='g', linestyle='--', label='解冻卷积层')plt.xlabel('Epoch')plt.ylabel('损失值')plt.title('损失值随Epoch变化')plt.legend()plt.grid(True)plt.tight_layout()plt.show()# 主函数：训练模型
   def main():# 参数设置epochs = 40  # 总训练轮次freeze_epochs = 5  # 冻结卷积层的轮次learning_rate = 1e-3  # 初始学习率weight_decay = 1e-4  # 权重衰减early_stop_patience = 3  # 早停耐心值（测试准确率未改善的轮数）# 创建ResNet50模型（加载预训练权重）model = create_resnet50(pretrained=True, num_classes=10)# 定义优化器和损失函数optimizer = optim.Adam(model.parameters(), lr=learning_rate, weight_decay=weight_decay)criterion = nn.CrossEntropyLoss()# 定义学习率调度器scheduler = optim.lr_scheduler.ReduceLROnPlateau(optimizer, mode='min', factor=0.5, patience=2)# 开始训练（前5轮冻结卷积层，之后解冻）final_accuracy, best_epoch = train_with_freeze_schedule(model=model,train_loader=train_loader,test_loader=test_loader,criterion=criterion,optimizer=optimizer,scheduler=scheduler,device=device,epochs=epochs,freeze_epochs=freeze_epochs,early_stop_patience=early_stop_patience)print(f"训练完成！最佳测试准确率: {final_accuracy:.2f}% (第 {best_epoch} 轮)")# 保存最佳模型torch.save(model.state_dict(), 'resnet50_cifar10_finetuned.pth')print("模型已保存至: resnet50_cifar10_finetuned.pth")if __name__ == "__main__":main()