别再只用MNIST了!Permuted/Split MNIST数据集实战:用PyTorch搭建你的第一个连续学习模型
用PyTorch实战连续学习:Permuted与Split MNIST数据集全解析
当你在Kaggle上看到第20个MNIST分类项目时,是否想过这个经典数据集还能玩出什么新花样?今天我们要打破常规,用PyTorch实现连续学习中的两个关键变体——Permuted MNIST和Split MNIST,让你亲身体验模型如何在不同任务间保持知识不遗忘。
1. 环境准备与数据加载
工欲善其事,必先利其器。我们先搭建好实验环境:
conda create -n cl python=3.8 conda activate cl pip install torch torchvision matplotlibPermuted MNIST的核心在于对图像像素进行随机重排。下面这段代码展示了如何生成10个不同排列顺序的任务:
import torch from torchvision import datasets, transforms def get_permuted_mnist(num_tasks=10): tasks = [] base_dataset = datasets.MNIST('./data', train=True, download=True) for i in range(num_tasks): if i == 0: # 第一个任务使用原始MNIST permutation = torch.arange(784) else: permutation = torch.randperm(784) transform = transforms.Compose([ transforms.ToTensor(), transforms.Lambda(lambda x: x.view(-1)[permutation].view(1, 28, 28)) ]) task_dataset = datasets.MNIST('./data', train=True, transform=transform) tasks.append(task_dataset) return tasksSplit MNIST的处理则完全不同,它将数字类别拆分到不同任务中:
def get_split_mnist(): tasks = [] base_transform = transforms.Compose([ transforms.ToTensor() ]) # 5个任务,每个任务包含两个数字类别 class_pairs = [(0,1), (2,3), (4,5), (6,7), (8,9)] for pair in class_pairs: class_mask = (dataset.targets == pair[0]) | (dataset.targets == pair[1]) task_data = dataset.data[class_mask] task_targets = dataset.targets[class_mask] # 将标签重新映射为0和1 task_targets = (task_targets == pair[1]).long() tasks.append((task_data, task_targets)) return tasks2. 连续学习模型架构设计
连续学习模型需要解决的核心问题是"灾难性遗忘"。我们实现两个经典方法:EWC(弹性权重固化)和LwF(学习不遗忘)。
2.1 EWC实现关键代码
EWC通过约束重要参数的更新来保护已有知识:
class EWC_Model(nn.Module): def __init__(self): super().__init__() self.fc1 = nn.Linear(784, 100) self.fc2 = nn.Linear(100, 10) def forward(self, x): x = x.view(-1, 784) x = F.relu(self.fc1(x)) return self.fc2(x) def compute_fisher(model, dataset, samples=100): fisher = {} for n, p in model.named_parameters(): fisher[n] = torch.zeros_like(p.data) for _ in range(samples): data, target = dataset[random.randint(0, len(dataset)-1)] output = model(data.unsqueeze(0)) loss = F.cross_entropy(output, torch.tensor([target])) model.zero_grad() loss.backward() for n, p in model.named_parameters(): fisher[n] += p.grad.data ** 2 / samples return fisher2.2 LwF实现要点
LwF使用知识蒸馏技术保留旧任务知识:
def lwf_loss(current_output, old_output, target, T=2, alpha=0.5): # 当前任务的交叉熵损失 ce_loss = F.cross_entropy(current_output, target) # 蒸馏损失 soft_target = F.softmax(old_output/T, dim=1) soft_current = F.log_softmax(current_output/T, dim=1) distill_loss = F.kl_div(soft_current, soft_target, reduction='batchmean') * (T**2) return alpha * ce_loss + (1-alpha) * distill_loss3. 训练流程与评估策略
连续学习的训练需要特别设计评估环节,以检测模型在所有已学任务上的表现。
3.1 训练循环示例
def train_task(model, task_data, optimizer, ewc=None, fisher=None, lambda_=1000): model.train() for epoch in range(10): # 每个任务训练10个epoch for data, target in DataLoader(task_data, batch_size=32): optimizer.zero_grad() output = model(data) loss = F.cross_entropy(output, target) # 如果是EWC,添加正则项 if ewc and fisher: for n, p in model.named_parameters(): loss += (lambda_/2) * torch.sum(fisher[n] * (p - ewc[n])**2) loss.backward() optimizer.step()3.2 评估指标设计
我们使用两个关键指标评估模型性能:
| 指标名称 | 计算公式 | 意义说明 |
|---|---|---|
| 准确率 | 正确预测数/总样本数 | 单任务表现 |
| 遗忘率 | (初始准确率-当前准确率) | 知识保留程度 |
| 正向迁移 | 新任务表现-随机初始化表现 | 旧知识对新任务的帮助 |
评估所有已学任务的代码片段:
def evaluate_all_tasks(model, tasks_so_far): results = {} model.eval() with torch.no_grad(): for i, task in enumerate(tasks_so_far): correct = 0 for data, target in DataLoader(task, batch_size=64): output = model(data) pred = output.argmax(dim=1) correct += (pred == target).sum().item() acc = correct / len(task) results[f"Task_{i}"] = acc return results4. 结果可视化与分析
可视化是理解模型行为的关键。我们使用Matplotlib绘制两个关键图表:
4.1 准确率矩阵
def plot_acc_matrix(acc_history): plt.figure(figsize=(10,8)) tasks = len(acc_history) acc_matrix = np.zeros((tasks, tasks)) for eval_task in range(tasks): for train_task in range(eval_task+1): acc_matrix[eval_task, train_task] = acc_history[train_task][eval_task] plt.imshow(acc_matrix, cmap='Blues', vmin=0, vmax=1) plt.colorbar() plt.xlabel("Evaluation Task") plt.ylabel("Training Task") plt.title("Accuracy Matrix")4.2 遗忘曲线
def plot_forgetting(acc_history): plt.figure(figsize=(10,6)) for task in range(len(acc_history)-1): initial_acc = acc_history[task][task] final_acc = acc_history[-1][task] plt.plot([task, len(acc_history)-1], [initial_acc, final_acc], label=f"Task {task}") plt.xlabel("Current Task") plt.ylabel("Accuracy") plt.title("Forgetting Curve") plt.legend()5. 实战技巧与常见问题
在真实项目中,我们积累了一些宝贵经验:
- 学习率调整:每个新任务开始时适当降低学习率(通常为初始值的1/5)
- 批次平衡:确保每个批次包含来自当前任务和之前任务的数据
- 正则化强度:EWC中的λ参数需要根据任务相似度调整
注意:Permuted MNIST最好使用MLP而非CNN,因为像素重排会破坏CNN依赖的空间局部性
常见错误及解决方案:
梯度爆炸:
- 现象:训练时loss突然变为NaN
- 解决:添加梯度裁剪
torch.nn.utils.clip_grad_norm_(model.parameters(), 1.0)
内存不足:
- 现象:训练多个任务后显存耗尽
- 解决:及时清理旧任务的缓存
torch.cuda.empty_cache()
评估模式错误:
- 现象:测试准确率远低于预期
- 解决:确保评估时调用
model.eval()并禁用梯度计算
6. 扩展与进阶方向
掌握了基础实现后,你可以尝试以下进阶方案:
- 混合策略:结合EWC和LwF的优点
- 动态架构:为每个任务添加专用子网络
- 元学习:使用MAML等元学习算法优化初始参数
# 混合策略示例 class Hybrid_Model(nn.Module): def __init__(self): super().__init__() self.shared = nn.Sequential( nn.Linear(784, 256), nn.ReLU() ) self.task_specific = nn.ModuleList([ nn.Linear(256, 10) for _ in range(num_tasks) ]) def forward(self, x, task_id): x = x.view(-1, 784) x = self.shared(x) return self.task_specific[task_id](x)在真实业务场景中,连续学习技术可以应用于:
- 用户行为预测(随时间变化的偏好)
- 金融风控(不断演变的欺诈模式)
- 医疗诊断(新增疾病类别的识别)
最后分享一个实用技巧:在Permuted MNIST实验中,记录下每个任务使用的排列顺序,这样可以在后续分析时重现完全相同的实验条件。这看似简单,却能让你的实验结果更具可重复性和说服力。
