持续学习实战：解决灾难性遗忘的经验回放与PyTorch基线

先从结论说起：持续学习（Continual Learning）解决的不是“把单任务模型训得更准”，而是让模型在连续到达的任务流上持续更新，同时尽量不破坏已经学到的旧任务能力。如果一个业务场景根本没有数据流，模型一次训练上线后不再更新，那持续学习暂时用不上；但凡涉及“今天来了新类别、明天来了新领域、后天数据分布又变了”，持续学习就应该被纳入选型范围。

为什么要强调“in Transition”？因为持续学习这个方向这几年正在发生明显转向：评价场景从 Split MNIST 这类玩具基准，转向大语言模型和视觉-语言模型上的持续微调；学习方式从离线多任务串行训练，转向在线、少样本、无任务边界的数据流；研究对象也从纯粹的算法指标，延展到模型部署、数据预算和系统稳定性。换句话说，持续学习正从一个“论文指标方向”，变成一个需要考虑工程可行性的落地方向。

对大部分读者来说，这篇文章能回答四个问题：第一，持续学习到底解决什么问题、什么时候才该用它；第二，当前主流方法分几类、各自的代价是什么；第三，怎么用 PyTorch 搭一个最小可跑的持续学习基线并正确评估；第四，真正复现和落地时最容易踩哪些坑。文中代码以教学验证为主，直接粘到自己的实验目录里就可以改。

1. 核心内容速览

从这张表也能看出来，持续学习不是一个“模型文件”，而是一套训练协议和评估协议的组合。这也是为什么单纯下载一个开源模型无法解决问题，真正要关注的是任务流怎么定义、数据预算怎么分配、指标怎么算。

2. 适用场景与使用边界

2.1 什么场景真的需要持续学习

持续学习不是万能的，但下面几类场景天然符合它的假设：

第一类是数据流持续到达的业务系统。典型的例子是推荐系统。用户的兴趣分布会随时间漂移，新商品、新话题不断出现，模型如果每隔一段时间全量重训一次，训练成本很高，而且重训数据里旧样本的时效性已经变差。持续学习可以按天或按小时增量更新模型，减少全量重训频率。

第二类是类别持续扩展的识别系统。视觉质检、安防识别、医疗影像筛查这类业务中，新类别会陆续被标注出来。如果每次增加一个类别都重新训练整个模型，标注成本和计算成本都不可控。类增量设置（Class-Incremental）就是为这种场景设计的。

第三类是个性化设备端学习。手机、机器人、边缘设备上的模型无法把全部历史数据保存在本地，但可以保存一个小型重放缓冲，在设备端持续适配用户习惯。这种场景对隐私敏感，不能把数据上传到中心服务器，持续学习提供了一种“本地增量更新”的替代方案。

第四类是对齐大模型知识时效性。大语言模型的持续预训练、持续指令微调已经变成研究热点。每过一段时间就有新的文档、新的对话数据、新的工具调用格式需要注入模型。直接全量微调成本太高，只在新数据上继续微调又会出现灾难性遗忘，因此需要持续学习或基于 PEFT 的持续微调方案。

2.2 什么场景不适合用持续学习

如果业务本身是静态数据集训练，模型训练完直接上线，之后再无新数据，那持续学习没有任何增益，常规训练流程更简单、更稳定。

如果数据分布变化非常大，旧任务和新任务几乎没有共享结构，持续学习也很难同时保住“高可塑性”和“高稳定性”。比如前一个任务学猫狗图像，后一个任务直接变成雷达波形分类，再好的持续学习算法也保存不了太多旧能力。

如果旧数据因为合规要求必须删除，且任务又与旧数据强相关，那基于重放的持续学习方法就不合规。这时候要么选无数据重放的架构类或正则化方法，要么在算法设计上彻底避免存储原始样本。

2.3 使用边界与合规提醒

持续学习涉及数据保存、任务流划分、模型更新，因此有几点必须提前确认：

• 重放缓冲中保存的旧样本是否有版权和隐私限制。涉及人脸、声音、医疗、儿童数据时，需要先确认授权和保留期限。
• 增量更新的模型是否用于生产环境。发布前要做效果复核，尤其是旧任务精度不能低于可接受阈值。
• 不要在未获授权的数据上进行去标识化、伪造或识别类实验。
• 如果项目中使用开源数据集，如 CORe50、DomainNet 等，需要保留数据许可证信息并遵守引用要求。

3. 持续学习的方法分类与理论基础

3.1 为什么普通训练会遗忘

普通监督学习假设训练集是独立同分布采样，模型在同一份数据上来回迭代直到收敛。当数据变成流式输入后，模型每到一个新任务，梯度都在向新任务的目标方向更新，旧任务的决策边界就会被覆盖。这种现象叫灾难性遗忘。

遗忘的本质是参数共享冲突。神经网络的同一组权重同时承担旧任务和新任务的表示，当优化目标切换到新任务时，旧任务方向的梯度被新任务梯度取代。如果新任务数据量足够大，旧任务的局部最优解会被破坏得比较彻底。

3.2 三条主流技术路线

当前持续学习算法大致分三类，理解它们之后再选择会更容易。

基于重放（Replay）的方法。这类方法在遇到新任务时，从旧任务中保留一部分样本或生成样本，与新任务数据混合在一起训练。最简单的形式是经验回放（Experience Replay，简称 ER），直接维护一个固定大小的缓冲，每次训练时从缓冲中采样一批旧样本，和新样本组成同一个 batch。iCaRL 在回放基础上加入了类别均值特征和近邻分类，A-GEM 则用投影梯度方式保证旧任务损失不上升。重放类方法实现简单、效果稳定，是多数业务场景的首选基线。

基于正则化（Regularization）的方法。这类方法不保存旧样本，而是在损失函数中增加一个惩罚项，限制对旧任务重要的参数变化。EWC（Elastic Weight Consolidation）用 Fisher 信息矩阵估计每个参数对旧任务的重要程度，重要参数尽可能不动。SI（Synaptic Intelligence）则在训练过程中在线估算参数重要度。知识蒸馏类方法也属于这个方向，LwF（Learning without Forgetting）用旧模型输出来约束新模型的输出分布，避免旧任务特征被过度改写。正则化方法省内存，但通常稳定性不如重放类方法。

基于架构（Architecture）的方法。这类方法为不同任务分配不同子网络或掩码，从结构上隔离任务冲突。PackNet 用剪枝思路把网络划分成多个子网络，HAT 用 Hard Attention Mask 让不同任务各走不同的前向路径，动态扩展网络则在新任务到来时增加新的分支。架构类方法在任务边界已知时效果很好，但部署时子网络管理复杂，不适合任务数量无限扩展的场景。

三类方法不是互斥的。现代持续学习算法经常混用重放、蒸馏和参数隔离，比如先用 PEFT 隔离一部分参数，再配合回放缓冲稳定旧任务。

3.3 任务边界与在线学习

持续学习实验设计里有一个关键区别：任务边界是否已知。

• Task-Incremental：训练时知道当前属于哪个任务，推理时也知道任务 ID，模型可以使用多头输出头。这个问题相对简单。
• Domain-Incremental：训练时知道任务 ID，但推理时不知道，模型必须用同一个输出头处理所有领域。
• Class-Incremental：训练时知道任务 ID，推理时不知道，模型要在不断增长的类别集合上做分类。这是最难也最接近实际业务的设置。

如果数据流是逐样本到达而非逐任务到达，就进入 Online Continual Learning。在线设置下每个类别可能只出现几十个样本，模型见一次就必须学一次。这种场景更依赖重放缓冲和快速适应算法。

4. 实验环境准备与复现前置条件

这里给一套通用环境准备清单，适合本地模型研究与复现。如果你用的是学校或公司的 GPU 服务器，流程相同，只是路径不同。

4.1 基础软件依赖

• 操作系统：Linux 优先，Windows 也可以跑 PyTorch 小实验。
• Python：推荐 3.10 或以上版本。
• 深度学习框架：PyTorch 2.x，CPU 版本也能跑通 MNIST 级实验，但 CIFAR 以上建议用 GPU。
• 可选实验框架：Avalanche、Mammoth。
• 数据集存储：提前准备目录存放 MNIST、CIFAR-100 等数据集，后续拆分任务使用。

4.2 创建环境命令

conda create -n clab python=3.10 -y conda activate clab # 安装 CPU 或 GPU 版 PyTorch，GPU 版请到官网按 CUDA 版本选择命令 pip install torch torchvision # 如果使用 Avalanche，可以安装完整版 pip install avalanche-lib # 可视化与实验数据记录 pip install matplotlib tensorboard

安装完成后，用一小段代码确认环境可用：

import torch print(torch.__version__) print(torch.cuda.is_available())

如果torch.cuda.is_available()返回 False，不代表不能做持续学习实验，只代表所有训练会落到 CPU 上。Split MNIST 这类实验在 CPU 上几分钟就能跑完。

4.3 目录结构建议

推荐把实验代码、数据、结果分开管理，避免后续批量实验时目录混乱。

continual-learning-lab/ ├── data/ ├── models/ ├── methods/ │ ├── replay.py │ ├── ewc.py │ └── baseline.py ├── utils/ │ ├── metrics.py │ └── data_loader.py ├── configs/ │ └── exp1.yaml ├── outputs/ │ └── logs/ └── run_experiments.py

这个结构不是必须照搬，但建议至少分离数据、代码和输出。批量跑实验时，输出目录会很快塞满日志和模型文件。

5. 最小可跑的持续学习基线：经验回放

下面我从零写一个最小可跑的经验回放（ER）基线，只用 PyTorch 基础 API，不依赖 Avalanche。这样做的好处是代码完全可控，方便你改成 EWC、LwF 等其它算法。

5.1 准备任务流数据

先定义一个按任务切分数据集的函数。以 Split MNIST 为例，把 10 个数字按顺序切分成 5 个任务，每个任务包含 2 个类别。

import torch from torch.utils.data import DataLoader, TensorDataset from torchvision import datasets, transforms def split_mnist_by_task(root="./data", n_tasks=5, seed=0): transform = transforms.Compose([ transforms.ToTensor(), transforms.Normalize((0.1307,), (0.3081,)) ]) train_set = datasets.MNIST(root=root, train=True, download=True, transform=transform) test_set = datasets.MNIST(root=root, train=False, download=True, transform=transform) classes_per_task = 10 // n_tasks train_tasks, test_tasks = [], [] for t in range(n_tasks): cls = list(range(t * classes_per_task, (t + 1) * classes_per_task)) train_idx = [i for i, label in enumerate(train_set.targets) if label in cls] test_idx = [i for i, label in enumerate(test_set.targets) if label in cls] train_sub = TensorDataset(train_set.data[train_idx].float() / 255.0, train_set.targets[train_idx]) test_sub = TensorDataset(test_set.data[test_idx].float() / 255.0, test_set.targets[test_idx]) train_tasks.append(DataLoader(train_sub, batch_size=128, shuffle=True)) test_tasks.append(DataLoader(test_sub, batch_size=256)) return train_tasks, test_tasks

5.2 定义简单 MLP 模型

这里用两层 MLP，方便 CPU 快速验证。如果你想替换成 ResNet，只需要保证输入输出维度。

import torch.nn as nn class SimpleMLP(nn.Module): def __init__(self, input_size=784, hidden=256, num_classes=10): super().__init__() self.net = nn.Sequential( nn.Flatten(), nn.Linear(input_size, hidden), nn.ReLU(), nn.Linear(hidden, hidden), nn.ReLU(), nn.Linear(hidden, num_classes), ) def forward(self, x): return self.net(x)

5.3 实现重放缓冲区

重放缓冲区只保留最老或最近的部分样本，代码里用一个列表保存展平后的图像和标签。核心逻辑是：当缓冲超过buffer_size时，淘汰最老的样本。

import random class ReplayBuffer: def __init__(self, buffer_size=2000): self.buffer_size = buffer_size self.samples = [] def add(self, x, y): # x: [N, C, H, W] 或 [N, D] x = x.detach().cpu() y = y.detach().cpu() for i in range(x.size(0)): self.samples.append((x[i], y[i].item())) if len(self.samples) > self.buffer_size: self.samples = self.samples[-self.buffer_size:] def sample(self, batch_size): batch = random.sample(self.samples, batch_size) xs = torch.stack([s[0] for s in batch]) ys = torch.tensor([s[1] for s in batch], dtype=torch.long) return xs, ys

5.4 每个任务的训练循环

持续学习训练循环与普通训练的区别在于：每个任务内部可以正常迭代多轮，但任务切换后旧任务数据不能再全量获取，只能从重放缓冲区采样。

import torch.optim as optim import torch.nn.functional as F def train_one_task(model, optimizer, task_loader, replay_buffer, device, replay_batch_size=64): model.train() for x, y in task_loader: x, y = x.to(device), y.to(device) if len(replay_buffer.samples) > 0: rx, ry = replay_buffer.sample(min(replay_batch_size, len(replay_buffer.samples))) rx, ry = rx.to(device), ry.to(device) x = torch.cat([x, rx], dim=0) y = torch.cat([y, ry], dim=0) optimizer.zero_grad() logits = model(x) loss = F.cross_entropy(logits, y) loss.backward() optimizer.step()

这里有一个细节：如果新任务的类别和旧任务类别都放进同一个分类头训练，类别增量场景下可能会出现分类头偏差。最直观的处理方式是在每个任务训练完之后，用重放缓冲区重新校准分类头，或者使用最近类均值分类器。这个后续在最佳实践里再展开。

5.5 评估函数

评估时对每个任务单独算准确率，并记录到矩阵中。矩阵的第 i 行第 j 列表示训练完任务 i 之后，在任务 j 测试集上的准确率。

def evaluate_all(model, test_tasks, device): model.eval() accs = [] with torch.no_grad(): for test_loader in test_tasks: correct = 0 total = 0 for x, y in test_loader: x, y = x.to(device), y.to(device) logits = model(x) pred = logits.argmax(dim=1) correct += (pred == y).sum().item() total += y.size(0) accs.append(correct / total) return accs

5.6 主循环

这里串起五个任务，每个任务训练 5 个 epoch。训练完每个任务后立刻评估并记录结果。

def main(): device = "cuda" if torch.cuda.is_available() else "cpu" train_tasks, test_tasks = split_mnist_by_task(root="./data") model = SimpleMLP().to(device) replay_buffer = ReplayBuffer(buffer_size=2000) acc_matrix = [] optimizer = optim.SGD(model.parameters(), lr=0.01, momentum=0.9) for t, task_loader in enumerate(train_tasks): for epoch in range(5): train_one_task(model, optimizer, task_loader, replay_buffer, device) # 训练结束后把当前任务数据加入缓冲 for x, y in task_loader: replay_buffer.add(x, y) accs = evaluate_all(model, test_tasks, device) acc_matrix.append(accs) print(f"Task {t + 1} done: test accs = {[round(a, 4) for a in accs]}") return torch.tensor(acc_matrix) if __name__ == "__main__": acc_matrix = main()

这个基线已经能把“只在新任务上训练”的遗忘现象明显压制住。你可以跑两遍：一遍把 replay 部分去掉，一遍保留 replay，对比任务 2 结束之后旧任务准确率的差异。

6. 评估协议：准确率、遗忘率与迁移指标

持续学习的评估不是只看“最后一个模型在所有任务上的平均准确率”，还要看训练过程中的稳定性。先定义一个指标矩阵，然后从矩阵推导三个常用指标。

假设一共有 N 个任务，acc_matrix[i][j]表示训练完任务 i 之后，在任务 j 测试集上的准确率。

6.1 平均准确率（ACC）

平均准确率有两种常见算法。第一种是取对角线准确率的平均，衡量每个任务刚学完时的表现：

diagonal_acc = torch.diag(acc_matrix).mean().item()

第二种是取训练完所有任务后，模型在所有任务上的最终准确率，也叫最终平均准确率：

final_acc = acc_matrix[-1, :].mean().item()

实际论文中，ACC 通常指后者，因为它反映了模型在任务流结束时保留了多少能力。

6.2 后向迁移/遗忘（BWT）

BWT 衡量模型学完后续任务后，对之前任务准确率的影响。如果 BWT 为负，说明发生了遗忘；如果为正，说明新任务帮助提升了旧任务表现，这种情况称为正迁移。

def calculate_bwt(acc_matrix): n = acc_matrix.shape[0] bwt = 0.0 for t in range(n - 1): bwt += (acc_matrix[-1, t] - acc_matrix[t, t]) return bwt / (n - 1)

6.3 前向迁移（FWT）

FWT 衡量模型在尚未见过的任务上，是否因为之前学习积累了更好的初始化。通常用训练任务 i 时，在任务 j（j > i）测试集上的准确率减去随机初始化模型在该任务上的准确率基线。

def calculate_fwt(acc_matrix, random_baseline): n = acc_matrix.shape[0] fwt = 0.0 count = 0 for i in range(n): for j in range(i + 1, n): fwt += acc_matrix[i, j] - random_baseline[j] count += 1 return fwt / count

6.4 评估协议注意事项

评估时最容易被忽视的是任务顺序和随机种子。持续学习结果对任务顺序极其敏感，同一个数据集，交换任务顺序后结果可能差异很大。因此论文实验通常要跑多个种子，报告均值和标准差。

另一个问题是每个 epoch 结束后的评估频率。在线持续学习中，数据是可流式的，如果只在任务边界评估，看不到模型在新任务早期阶段的波动。建议在任务内部固定间隔做一次快速评估，观察稳定性。

7. 训练资源占用与性能观察

持续学习训练过程的资源占用和普通深度模型训练类似，但多了一个重放缓冲的内存开销。这里讨论怎么观察显存、内存变化，以及哪些因素会放大成本。

7.1 显存占用观察方法

使用 GPU 训练时，可以用命令行实时观察显存：

# 每隔 1 秒刷新一次 GPU 占用 watch -n 1 nvidia-smi

在代码里也可以打印当前已显存：

if torch.cuda.is_available(): allocated = torch.cuda.memory_allocated() / 1024**2 print(f"allocated memory: {allocated:.1f} MB")

显存占用主要来自三个部分：模型参数和梯度、当前 batch 的中间激活、重放样本的临时张量。重放缓冲本身如果保存的是原始小图，占用的主要是内存而非显存；只有把重放样本送到 GPU 时才会临时占用显存。

7.2 影响性能的关键因素

• 重放缓冲区大小。缓冲区越大，每个 batch 中旧样本占比越高，显存和内存开销越大，但对旧任务的保持效果通常越好。需要根据数据规模调整。
• batch 大小。新任务样本和重放样本合在一个 batch 时，总 batch 增大，显存占用上升。如果显存紧张，可以降低 replay batch size。
• 模型宽度和输入分辨率。MLP 在 CPU 上就能跑，ResNet 或 ViT 上显存需求明显上升。输入分辨率从 224 降到 128，显存和计算量下降很多。
• 任务数。任务数增加不会显著增加单次训练显存，但会增加评估耗时和重放缓冲的总样本量。

7.3 降低资源占用的建议

如果实验环境显存有限，优先把输入归一化放到 CPU 端完成，再用pin_memory预加载。重放样本可以直接以 npy 格式压缩保存到磁盘，训练时随机读取一个 mini-batch，而不是把所有重放样本一次性常驻内存。

# DataLoader 启用 pin_memory，可以减少 GPU 拷贝等待 DataLoader(..., num_workers=2, pin_memory=True)

如果使用大模型做持续微调，建议优先采用 LoRA 等 PEFT 方案，只训练低秩适配器。这样旧任务参数基本不变，同时显存占用可以大幅下降。

8. 常见问题与排查方法

持续学习实验跑起来之后，会遇到几类高频问题。这里整理成排查清单。

排障时最重要的一件事是：先固定一个简单基线。比如先跑纯 Naive（没有回放）的持续学习，确认代码逻辑正确，再逐步加回放、加蒸馏、加正则化。不要一上来就往复杂算法上堆。

9. 最佳实践与落地建议

9.1 实验设计层面

第一，每个方法都要控制相同的存储预算。持续学习论文里比较重放和正则化方法时，重放类方法存储旧样本会占用额外空间。公平比较应该在“同等额外存储开销”下进行，否则结论不公正。

第二，多个随机种子取平均。持续学习结果方差大，单次实验可能得出完全相反的结论。至少跑 3 个种子，报告均值和标准差。

第三，记录每个任务结束后的模型快照。这样回溯遗忘曲线时可以直接加载对应模型，不需要重新训练。

9.2 算法选择层面

如果你的业务允许存储少量旧样本，优先考虑基于重放的方法。实现简单、调参门槛低、效果稳定。

如果业务不允许保留原始样本，选择正则化类方法，但要对旧任务的精度保持设置一个可接受阈值。EWC 对超参数敏感，需要单独调 Fisher 采样数量和正则化权重。

如果任务边界清晰、任务数量有限，架构类方法能提供最强的旧任务保护。但部署时要注意子网络索引管理，任务数量过多时不能无限制扩展。

如果业务数据是流式逐样本到达，不要使用离线分任务的评估方式，要改为在线评估协议，并按时间窗口记录数据到达顺序。

9.3 工程化建议

• 把任务流定义成配置文件，包含任务顺序、数据路径、重放缓冲大小、训练轮数，保证实验可复现。
• 批量实验时使用独立的日志文件，记录每次运行的任务顺序、随机种子、超参数和评估结果。
• 涉及人脸、声音、版权数据时，必须确认授权。重放缓冲区保存的旧样本同样要合规。
• 生产环境发布前，用一套固定“任务流回归集”验证旧任务精度是否低于阈值。如果低于阈值，回滚到上一版本模型或触发一次全量重训。
• 使用大模型做增量微调时，优先尝试 LoRA 或前缀微调，并配合小型重放缓冲，成本可控，效果也容易评估。

9.4 从算法验证到业务部署

持续学习落地不是把训练脚本跑通就行。实际生产环境里，任务流来自真实数据管道，可能会有标签延迟、数据缺失、类别采样不均等问题。建议先做一个小范围影子测试：用真实数据流的一个月切片，重建历史任务流，对比“全量重训基线”和“持续学习增量更新基线”，重点看旧任务精度保持、新任务适应速度、训练耗时和存储占用。只有对比通过，再考虑替换生产训练链路。

10. 总结与下一步

持续学习最有价值的点，是把“模型上线后就不能再动”或者“每次更新都要全量重训”的二元选择，变成了一种平滑过渡方案。你可以用小批量、低代价的增量更新，在尽量不破坏旧能力的前提下，让模型跟随新任务流持续演进。

这篇文章里最值得先验证的是那个最小经验回放基线。它能直观展示灾难性遗忘现象，也能让你在 10 分钟内在 CPU 上跑通一轮完整任务流评估。最容易踩的坑集中在类增量评估和任务顺序敏感度上，尤其是分类头偏置问题，初学者很容易忽略。

后续想继续深入，可以从这几个方向展开：先给上面的 ER 基线加上知识蒸馏，复现一个 LwF；再把 EWC 的 Fisher 正则项加进去，对比两类方法在相同存储预算下的表现；然后试着把模型换成 ResNet，把数据集从 MNIST 换成 CIFAR-100，观察类增量场景下的精度变化；如果条件允许，再尝试在开源大语言模型上用 PEFT 做持续指令微调，验证不同 LoRA 配置对旧指令集的保持效果。

持续学习不是银弹，但它提供了一套值得研究的“过渡技术”：从静态训练过渡到增量更新，从离线评估过渡到在线验证，从单模型过渡到持续演进的模型状态机。建议有数据流和增量更新需求的团队，先收藏这篇文章，把基线代码跑通，再决定要不要把持续学习引入自己的业务链路。