迁移学习与模型微调

# 渐进解冻实现示例 import torch import torch.nn as nn from torchvision import models def gradual_unfreeze(model, num_stages=4): # 首先冻结所有层 for param in model.parameters(): param.requires_grad = False # 替换分类头（始终可训练） num_features = model.fc.in_features model.fc = nn.Linear(num_features, 10) return model def set_gradual_stage(model, stage, total_layers): layers = list(model.named_children()) backbone_layers = [n for n, _ in layers if n != 'fc'] if stage == 0: for name, param in model.named_parameters(): if 'fc' in name: param.requires_grad = True elif stage == 1: unfreeze_count = len(backbone_layers) // 4 for name, param in model.named_parameters(): if 'fc' in name: param.requires_grad = True for name, child in list(model.named_children())[-unfreeze_count:]: for param in child.parameters(): param.requires_grad = True elif stage == 2: unfreeze_count = len(backbone_layers) // 2 for name, param in model.named_parameters(): if 'fc' in name: param.requires_grad = True for name, child in list(model.named_children())[-unfreeze_count:]: for param in child.parameters(): param.requires_grad = True elif stage == 3: for param in model.parameters(): param.requires_grad = True resnet = models.resnet50(weights=models.ResNet50_Weights.IMAGENET1K_V1) model = gradual_unfreeze(resnet) stages = [0, 0, 0, 1, 1, 2, 2, 3] for epoch, stage in enumerate(stages): set_gradual_stage(model, stage, 50)

# PyTorch中实现区别学习率 import torch import torch.nn as nn from torchvision import models def get_discriminative_optimizer(model, base_lr=1e-4): layers = list(model.named_children()) backbone_layers = [n for n, _ in layers if n != 'fc'] mid = len(backbone_layers) // 2 group1_names = backbone_layers[:mid] group1_params = [] for name, child in model.named_children(): if name in group1_names: group1_params.extend(list(child.parameters())) group2_names = backbone_layers[mid:] group2_params = [] for name, child in model.named_children(): if name in group2_names: group2_params.extend(list(child.parameters())) group3_params = list(model.fc.parameters()) optimizer = torch.optim.AdamW([ {'params': group1_params, 'lr': base_lr * 0.1}, {'params': group2_params, 'lr': base_lr * 0.5}, {'params': group3_params, 'lr': base_lr * 2.0}, ], weight_decay=0.01) return optimizer model = models.resnet50(weights=models.ResNet50_Weights.IMAGENET1K_V1) model.fc = nn.Linear(2048, 10) optimizer = get_discriminative_optimizer(model) for epoch in range(10): for inputs, labels in dataloader: outputs = model(inputs) loss = nn.CrossEntropyLoss()(outputs, labels) loss.backward() optimizer.step() optimizer.zero_grad()

# 冻结主干网络的完整流程 import torch import torch.nn as nn import torchvision.transforms as transforms from torchvision import models from torch.utils.data import DataLoader, Dataset # 1. 加载预训练模型 model = models.resnet50(weights=models.ResNet50_Weights.IMAGENET1K_V1) # 2. 冻结所有层（主干） for param in model.parameters(): param.requires_grad = False # 3. 替换分类头（新分类头默认可训练） num_features = model.fc.in_features model.fc = nn.Sequential( nn.Dropout(0.5), nn.Linear(num_features, 256), nn.ReLU(inplace=True), nn.Dropout(0.3), nn.Linear(256, 10) ) # 4. 验证冻结状态 trainable_params = sum(p.numel() for p in model.parameters() if p.requires_grad) total_params = sum(p.numel() for p in model.parameters()) print(f'可训练参数: {trainable_params:,} / {total_params:,} ({100 * trainable_params / total_params:.2f}%)')

# 不同架构的分类头替换方法 # ResNet系列 resnet = models.resnet50(weights=models.ResNet50_Weights.IMAGENET1K_V1) resnet.fc = nn.Linear(2048, num_classes) # VGG系列 vgg = models.vgg16(weights=models.VGG16_Weights.IMAGENET1K_V1) vgg.classifier[6] = nn.Linear(4096, num_classes) # EfficientNet系列 effnet = models.efficientnet_b0(weights=models.EfficientNet_B0_Weights.IMAGENET1K_V1) effnet.classifier[1] = nn.Linear(1280, num_classes) # ViT (Vision Transformer) vit = models.vit_b_16(weights=models.ViT_B_16_Weights.IMAGENET1K_V1) vit.heads.head = nn.Linear(768, num_classes)

# 线性探测实现 from sklearn.linear_model import LogisticRegression from sklearn.preprocessing import StandardScaler import numpy as np import torch def extract_features(model, dataloader, device='cuda'): model.eval() features_list = [] labels_list = [] with torch.no_grad(): for inputs, labels in dataloader: inputs = inputs.to(device) features = model.avgpool(model.layer4( model.layer3(model.layer2( model.layer1(model.maxpool( model.relu(model.bn1(model.conv1(inputs))))))))) features = torch.flatten(features, 1) features_list.append(features.cpu().numpy()) labels_list.append(labels.numpy()) return np.concatenate(features_list), np.concatenate(labels_list) train_features, train_labels = extract_features(model, train_loader) test_features, test_labels = extract_features(model, test_loader) scaler = StandardScaler() train_features_scaled = scaler.fit_transform(train_features) test_features_scaled = scaler.transform(test_features) clf = LogisticRegression(max_iter=1000, multi_class='multinomial') clf.fit(train_features_scaled, train_labels) accuracy = clf.score(test_features_scaled, test_labels) print(f'线性探测准确率: {accuracy:.4f}')

# Batch Normalization适配与域自适应 def adapt_batchnorm(model, target_loader, device='cuda'): model.train() with torch.no_grad(): for inputs, _ in target_loader: inputs = inputs.to(device) _ = model(inputs) model.eval() return model def pseudo_label_finetune(model, unlabeled_loader, threshold=0.9): model.eval() pseudo_data = [] with torch.no_grad(): for inputs in unlabeled_loader: outputs = model(inputs) probs = torch.softmax(outputs, dim=1) max_probs, pseudo_labels = torch.max(probs, dim=1) mask = max_probs > threshold if mask.sum() > 0: pseudo_data.extend([(inputs[i], pseudo_labels[i]) for i in range(len(inputs)) if mask[i]]) if len(pseudo_data) == 0: print('没有高置信度的伪标签样本') return model.train() criterion = nn.CrossEntropyLoss() optimizer = torch.optim.Adam(model.fc.parameters(), lr=1e-4) for inputs, labels in DataLoader(pseudo_data, batch_size=32, shuffle=True): optimizer.zero_grad() outputs = model(inputs) loss = criterion(outputs, labels) loss.backward() optimizer.step() print(f'使用 {len(pseudo_data)} 个伪标签样本完成微调')

# PyTorch迁移学习完整实战代码 import torch import torch.nn as nn import torch.optim as optim import torchvision import torchvision.transforms as transforms from torchvision import models from torch.utils.data import DataLoader, random_split import numpy as np from tqdm import tqdm # ========== 1. 配置超参数 ========== class Config: data_dir = './flower_data' num_classes = 102 batch_size = 32 num_epochs = 30 initial_lr = 1e-4 device = torch.device('cuda' if torch.cuda.is_available() else 'cpu') num_workers = 4 random_seed = 42 config = Config() # ========== 2. 数据预处理 ========== train_transform = transforms.Compose([ transforms.RandomResizedCrop(224), transforms.RandomHorizontalFlip(p=0.5), transforms.RandomRotation(15), transforms.ColorJitter(brightness=0.2, contrast=0.2), transforms.ToTensor(), transforms.Normalize(mean=[0.485, 0.456, 0.406], std=[0.229, 0.224, 0.225]) ]) val_transform = transforms.Compose([ transforms.Resize(256), transforms.CenterCrop(224), transforms.ToTensor(), transforms.Normalize(mean=[0.485, 0.456, 0.406], std=[0.229, 0.224, 0.225]) ]) full_dataset = torchvision.datasets.Flowers102( root=config.data_dir, split='train', transform=train_transform, download=True ) train_size = int(0.8 * len(full_dataset)) val_size = len(full_dataset) - train_size train_dataset, val_dataset = random_split(full_dataset, [train_size, val_size]) val_dataset.dataset.transform = val_transform train_loader = DataLoader(train_dataset, batch_size=config.batch_size, shuffle=True, num_workers=config.num_workers) val_loader = DataLoader(val_dataset, batch_size=config.batch_size, shuffle=False, num_workers=config.num_workers) # ========== 3. 加载预训练模型 ========== def create_model(strategy='top'): model = models.resnet50(weights=models.ResNet50_Weights.IMAGENET1K_V1) num_features = model.fc.in_features model.fc = nn.Sequential( nn.Dropout(0.3), nn.Linear(num_features, 512), nn.ReLU(inplace=True), nn.BatchNorm1d(512), nn.Dropout(0.3), nn.Linear(512, config.num_classes) ) if strategy == 'feature_extraction': for param in model.parameters(): param.requires_grad = False for param in model.fc.parameters(): param.requires_grad = True elif strategy == 'top': for param in model.parameters(): param.requires_grad = False for param in model.layer4.parameters(): param.requires_grad = True for param in model.fc.parameters(): param.requires_grad = True elif strategy == 'full': for param in model.parameters(): param.requires_grad = True return model # ========== 4. 训练与评估函数 ========== def train_one_epoch(model, loader, criterion, optimizer, device): model.train() running_loss = 0.0 correct = 0 total = 0 for inputs, labels in tqdm(loader, desc='Training'): inputs, labels = inputs.to(device), labels.to(device) optimizer.zero_grad() outputs = model(inputs) loss = criterion(outputs, labels) loss.backward() optimizer.step() running_loss += loss.item() _, predicted = outputs.max(1) total += labels.size(0) correct += predicted.eq(labels).sum().item() return running_loss / len(loader), 100.0 * correct / total def evaluate(model, loader, criterion, device): model.eval() running_loss = 0.0 correct = 0 total = 0 with torch.no_grad(): for inputs, labels in tqdm(loader, desc='Evaluating'): inputs, labels = inputs.to(device), labels.to(device) outputs = model(inputs) loss = criterion(outputs, labels) running_loss += loss.item() _, predicted = outputs.max(1) total += labels.size(0) correct += predicted.eq(labels).sum().item() return running_loss / len(loader), 100.0 * correct / total # ========== 5. 主训练流程 ========== def main(): device = config.device print(f'使用设备: {device}') model = create_model(strategy='top') model = model.to(device) trainable = sum(p.numel() for p in model.parameters() if p.requires_grad) total = sum(p.numel() for p in model.parameters()) print(f'可训练参数: {trainable:,} / {total:,} ({100*trainable/total:.2f}%)') criterion = nn.CrossEntropyLoss(label_smoothing=0.1) optimizer = optim.AdamW([ {'params': model.layer4.parameters(), 'lr': config.initial_lr * 0.5}, {'params': model.fc.parameters(), 'lr': config.initial_lr} ], weight_decay=0.01) scheduler = optim.lr_scheduler.CosineAnnealingLR(optimizer, T_max=config.num_epochs) best_acc = 0.0 for epoch in range(config.num_epochs): print(f'\nEpoch {epoch+1}/{config.num_epochs}') train_loss, train_acc = train_one_epoch(model, train_loader, criterion, optimizer, device) val_loss, val_acc = evaluate(model, val_loader, criterion, device) scheduler.step() print(f'Train Loss: {train_loss:.4f}, Train Acc: {train_acc:.2f}%') print(f'Val Loss: {val_loss:.4f}, Val Acc: {val_acc:.2f}%') if val_acc > best_acc: best_acc = val_acc torch.save(model.state_dict(), 'best_model.pth') print(f'保存最佳模型, Acc: {best_acc:.2f}%') print(f'\n训练完成! 最佳验证准确率: {best_acc:.2f}%') if __name__ == '__main__': main()