持续学习折腾手记

作者: 醉月思发布日期: 2023-01-26 分类: 技术实践

如果只能用一句话说持续学习：先把失败复现出来。

问题现场

项目是做电商商品的多分类识别。一开始是 10 个大类别，模型在测试集上跑到 95% 准确率，拿着这个模型就上了线。

过了一个月，运营说新增了 3 个类别的商品，需要模型也能识别。我当时想这还不好办，把新类别的数据混进去重新训练不就完了。

结果试了一下，发现这个想法太天真了：

# 原始训练流程
import torch
import torch.nn as nn

model = MyClassifier(num_classes=10)
optimizer = torch.optim.Adam(model.parameters(), lr=0.001)

# 第一阶段训练：10 个类别
train_old_categories(model, train_data_10classes)

# 第二阶段训练：新加 3 个类别
model = MyClassifier(num_classes=13)  # 换个输出层
optimizer = torch.optim.Adam(model.parameters(), lr=0.001)
train_new_categories(model, train_data_13classes)

# 测试旧类别的表现
test(model, test_data_10classes)  # 准确率从 95% 跌到 40%

第一阶段的模型在旧类别上表现正常，第二阶段训练完后，新类别是学到了，但旧类别几乎全军覆没。

这不是什么罕见现象， catastrophic forgetting 在持续学习领域就是个老问题。神经网络那堆共享参数学完新任务之后，很难保持对旧任务仍然好用。

第一个尝试：全量重新训练

最直接的方案就是把所有数据都攒起来重新训练一遍。理论上可行，但实际问题很多：

数据量越来越大，每次都要重新过一遍全部历史数据
训练成本持续累积，三个月下来 GPU 账单就吃不消
新来的人要看旧数据，数据权限和隐私管理变得复杂
出现新的噪声和标注错误，会影响已经训练好的部分

# 全量重新训练
def retrain_from_scratch(all_data, num_classes):
    model = MyClassifier(num_classes=num_classes)
    optimizer = torch.optim.Adam(model.parameters(), lr=0.001)

    for epoch in range(num_epochs):
        for batch in all_data:
            # 每次都要过全部历史数据
            loss = model.loss_fn(batch)
            loss.backward()
            optimizer.step()
            optimizer.zero_grad()

    return model

# 成本：时间和算力都线性增长

如果是数据量可控的小项目，这个方案其实挺干脆的。但在实际业务里数据只会越来越多，用全量重训去对抗遗忘，基本就是用更大的成本换取同样的准确率。

第二个尝试：保留一部分旧数据重播

既然全部数据太多，那就留一部分出来。思路是训练新任务的时候，时不时喂一点旧任务的数据进去，让模型不至于把旧东西忘得太干净。

实现起来也不算复杂：

import random
from collections import deque

class ReplayBuffer:
    def __init__(self, buffer_size=1000):
        self.buffer = deque(maxlen=buffer_size)

    def add(self, data):
        self.buffer.append(data)

    def sample(self, batch_size):
        return random.sample(self.buffer, batch_size)

# 训练时混合新旧数据
def incremental_training_with_replay(model, new_data, replay_buffer, epochs=10):
    optimizer = torch.optim.Adam(model.parameters(), lr=0.001)

    for epoch in range(epochs):
        for new_batch in new_data:
            # 新数据
            optimizer.zero_grad()
            loss_new = model.loss_fn(new_batch)
            loss_new.backward()

            # 从重播缓冲区采样旧数据
            old_batch = replay_buffer.sample(len(new_batch))
            loss_old = model.loss_fn(old_batch)
            loss_old.backward()

            optimizer.step()

这个方案缓解了遗忘，但问题也不少：

缓冲区选多大太玄学，小了不管用，大了又回到存储成本的问题
采样的策略很关键，均匀采样、困难样本采样、类别均衡采样，效果差异很大
还是需要一部分旧数据，只不过少一点，但数据和隐私问题还在

我在项目里试过这个方案，缓冲区设成数据量的 10%，结果旧类别的准确率能维持在 75% 左右，比直接忘干净好，但离原来的 95% 差得还是远。

第三个尝试：Elastic Weight Consolidation

既然数据存起来麻烦，那就从模型参数本身下手。Elastic Weight Consolidation（EWC）的想法是：对那些对旧任务很重要的参数，在训练新任务时要少动；不重要的可以随便改。

判断参数重要性的方法是 Fisher 信息矩阵，就是在旧任务上算一个二阶导数：

import torch.nn.functional as F

class EWC:
    def __init__(self, model, old_data, lambda_ewc=5000):
        self.model = model
        self.lambda_ewc = lambda_ewc

        # 保存旧任务的最优参数
        self.old_params = {}
        for name, param in model.named_parameters():
            self.old_params[name] = param.clone().detach()

        # 计算 Fisher 信息矩阵
        self.fisher = self._compute_fisher(model, old_data)

    def _compute_fisher(self, model, data):
        fisher = {}
        for name, param in model.named_parameters():
            fisher[name] = torch.zeros_like(param)

        model.eval()
        for batch in data:
            outputs = model(batch)
            loss = F.cross_entropy(outputs, batch.labels)

            # 计算每个参数的梯度
            model.zero_grad()
            loss.backward()

            for name, param in model.named_parameters():
                if param.grad is not None:
                    fisher[name] += param.grad.pow(2)

        # 归一化
        num_batches = len(data)
        for name in fisher:
            fisher[name] /= num_batches

        return fisher

    def penalty(self, model):
        loss = 0
        for name, param in model.named_parameters():
            if name in self.fisher and name in self.old_params:
                loss += (self.fisher[name] * (param - self.old_params[name]).pow(2)).sum()
        return loss * self.lambda_ewc

# 使用 EWC 训练新任务
def incremental_training_with_ewc(model, new_data, ewc, epochs=10):
    optimizer = torch.optim.Adam(model.parameters(), lr=0.001)

    for epoch in range(epochs):
        for batch in new_data:
            optimizer.zero_grad()

            # 新任务的损失
            loss_new = model.loss_fn(batch)

            # EWC 正则化项
            loss_ewc = ewc.penalty(model)

            loss = loss_new + loss_ewc
            loss.backward()
            optimizer.step()

这个方案的好处是不需要存旧数据，只需要存模型参数和 Fisher 矩阵。存储量小很多，而且计算量也比重播缓冲区小。

但问题也不少：

Fisher 矩阵的计算本身就有成本，而且要在旧数据上过一遍
超参数 lambda_ewc 很难调，太小管不住遗忘，太大又学不到新东西
多任务累积的时候，不同任务的 Fisher 信息会冲突，模型参数会被锁死

我在项目里用这个方案试了一圈，发现 lambda_ewc 从 100 调到 10000，效果也是忽高忽低。最后算是找了个妥协值，旧类别的准确率能维持在 80% 左右，但新类别的学习速度明显慢了很多。

第四个尝试：直接用增量学习框架

试了一圈自己实现，还是决定看看有没有现成框架。目前几个比较成熟的方案：

Avalanche：一个完整的持续学习库，支持多种策略
L2P：Learning to Prompt，用 prompt 的方式做知识传递
DER++：结合重播和蒸馏的方案

Avalanche 用起来相对简单：

from avalanche.benchmarks import SplitCIFAR10
from avalanche.models import SimpleMLP
from avalanche.training.strategies import EWCStrategy
from avalanche.evaluation.metrics import accuracy_metrics
from avalanche.logging import InteractiveLogger

# 创建持续学习数据集
benchmark = SplitCIFAR10(n_experiences=5)
model = SimpleMLP(num_classes=10)

# 使用 EWC 策略
eval_plugin = InteractiveLogger()
strategy = EWCStrategy(
    model=model,
    optimizer=torch.optim.SGD(model.parameters(), lr=0.01),
    train_mb_size=32,
    train_epochs=5,
    eval_mb_size=100,
    device='cuda',
    evaluator=eval_plugin,
    ewc_lambda=5000
)

# 逐个任务训练
for experience in benchmark.train_stream:
    strategy.train(experience)
    strategy.eval(benchmark.test_stream)

Avalanche 的好处是封装了很多现成的策略，不用自己从零实现。缺点是文档还不够细致，很多超参数要自己试。

我最后在项目里用的是简化版的 EWC 加上一小部分重播，算是把效果和成本都控制在可接受范围。

实际落地的配置

折腾了一圈，最后落地的方案大概是这个样子：

class HybridIncrementalLearning:
    def __init__(self, model, buffer_size=500, lambda_ewc=3000):
        self.model = model
        self.replay_buffer = ReplayBuffer(buffer_size)
        self.ewc = None
        self.lambda_ewc = lambda_ewc

    def first_task(self, train_data):
        # 第一个任务正常训练
        optimizer = torch.optim.Adam(self.model.parameters(), lr=0.001)
        self._train(self.model, train_data, optimizer, epochs=20)

        # 保存到重播缓冲区
        for batch in train_data:
            self.replay_buffer.add(batch)

        # 初始化 EWC
        self.ewc = EWC(self.model, train_data, self.lambda_ewc)

    def subsequent_task(self, new_data):
        optimizer = torch.optim.Adam(self.model.parameters(), lr=0.001)

        for epoch in range(15):
            for new_batch in new_data:
                optimizer.zero_grad()

                # 新任务损失
                loss_new = self._compute_loss(new_batch)

                # EWC 正则化
                loss_ewc = self.ewc.penalty(self.model)

                # 重播损失
                old_batch = self.replay_buffer.sample(len(new_batch))
                loss_replay = self._compute_loss(old_batch)

                loss = loss_new + 0.5 * loss_ewc + 0.3 * loss_replay
                loss.backward()
                optimizer.step()

        # 更新缓冲区和 EWC
        for batch in new_data:
            self.replay_buffer.add(batch)
        self.ewc = EWC(self.model, new_data, self.lambda_ewc)

这个配置在项目里跑了几个月，效果还算稳定：

旧类别准确率维持在 85% 左右
新类别学习速度不会太慢
重播缓冲区只保留 5% 的旧数据，存储成本可控

当然也没到完美的程度，偶尔还是会出现某个类别突然忘干净的情况。遇到这种情况，就只能单独拿那个类别的数据重新训一轮。

还没有完全解决的坑

虽然方案算是落地了，但坑还是在几个地方：

类别不平衡问题：新类别的数据量通常比旧类别大，模型会倾向新类别。这个问题我试过加权损失、难例采样，效果都不太稳定。
长尾遗忘：一些本来就很难的类别，用任何方案都容易忘。最后只能定期监控，发现掉太多就单独补训。
超参数管理：每个阶段的 lambda_ewc、重播比例、学习率都要调，而且不同任务的参数不通用。这部分经验很难迁移到下一个项目。
评估成本：要准确知道遗忘程度，就必须在所有历史任务上都测一遍。任务多了之后，评估的时间比训练还长。

简单收个尾

持续学习这件事，理论上讲得通，实际落地就全是妥协。

全量重训最干净，但成本扛不住；EWC 算起来优雅，但参数调得头秃；重播缓冲区能缓解，但数据和隐私问题还在。最后能落地的方案，通常是几者的混合。

我现在对待这个问题的态度比一开始务实多了。不再追求把旧任务的表现完全保持住，而是设定一个可接受的阈值，只要不掉到这个阈值以下就行。

毕竟在实际业务里，模型是活的，数据是活的，需求也是活的。要想让一个模型永远不变地记住所有东西，这个思路本身就不太现实。

参考资料

Kirkpatrick et al., “Overcoming catastrophic forgetting in neural networks”, PNAS 2017
Rebuffi et al., “iCaRL: Incremental Classifier and Representation Learning”, CVPR 2017
Lomonaco & Malitini, “Avalanche: an End-to-End Library for Continual Learning”, CoRR 2021

版权声明: 本文首发于指尖魔法屋-持续学习折腾手记（https://blog.thinkmoon.cn/post/182-continual-learning-catastrophic-forgetting-incremental-learning/）转载或引用必须申明原指尖魔法屋来源及源地址！

AI学习率调度踩坑记录

按教程设置0.001，开始训练，盯着loss曲线等它下降，然后隔几轮降低学习率。

📁 技术实践

📅 2021-03-15

AI批量归一化：这次怎么落地的

AI批量归一化：这次怎么落地的一旦进项目，好看的架构图就没那么管用了。

📁 技术实践

📅 2021-05-06