引言

现实世界在不断变化：新知识产生、旧信息更新、领域边界扩展。大语言模型在训练完成后，其知识即被冻结。如何让模型持续吸收新知识，同时不遗忘已学内容？这就是持续学习（Continual Learning，也称终身学习）要解决的核心问题。本文探讨大模型场景下持续学习的挑战、方法和工程实践。

核心挑战：灾难性遗忘

什么是灾难性遗忘

当模型在新数据上微调时，新数据会覆盖模型在旧数据上学到的知识和能力。这种现象称为灾难性遗忘——模型"记住"了新知识，却"忘记"了旧知识。

遗忘的严重性

灾难性遗忘的程度取决于多种因素：

  • 数据分布差异：新数据与原始训练数据分布差异越大，遗忘越严重
  • 模型规模：较小模型往往遗忘更严重（容量限制）
  • 训练强度：学习率越高、训练轮数越多，遗忘越严重
  • 任务相关性：新任务与旧任务越不相关，遗忘越严重

遗忘的检测

def evaluate_forgetting(model, old_eval_set, new_eval_set):
    """评估灾难性遗忘程度"""
    old_score = evaluate(model, old_eval_set)
    new_score = evaluate(model, new_eval_set)
    
    return {
        'old_task_score': old_score,      # 旧任务表现
        'new_task_score': new_score,      # 新任务表现
        'forgetting_rate': 1 - old_score,  # 遗忘率
        'learning_rate': new_score,        # 学习率
        'overall_score': (old_score + new_score) / 2  # 综合表现
    }

持续学习方法

方法一：经验回放（Experience Replay）

在训练新数据时，混入部分旧数据：

class ReplayBuffer:
    def __init__(self, capacity=10000):
        self.capacity = capacity
        self.buffer = []
    
    def add(self, data):
        self.buffer.extend(data)
        if len(self.buffer) > self.capacity:
            # 随机采样保留
            self.buffer = random.sample(self.buffer, self.capacity)
    
    def sample(self, batch_size):
        return random.sample(self.buffer, min(batch_size, len(self.buffer)))

def train_with_replay(model, new_data, replay_buffer, replay_ratio=0.3):
    """带经验回放的训练"""
    for epoch in range(epochs):
        for new_batch in get_batches(new_data):
            # 混合新数据和回放数据
            replay_batch = replay_buffer.sample(int(len(new_batch) * replay_ratio))
            mixed_batch = new_batch + replay_batch
            random.shuffle(mixed_batch)
            
            loss = model.compute_loss(mixed_batch)
            loss.backward()
            optimizer.step()
        
        # 将新数据加入回放缓冲区
        replay_buffer.add(new_data)

优势：简单有效、通用性强 劣势：需要存储旧数据、可能违反数据隐私

方法二：弹性权重整合（EWC）

通过正则化保护重要参数不被大幅修改：

class EWC:
    def __init__(self, model, task_data):
        self.model = model
        self.fisher_information = self.compute_fisher(task_data)
    
    def compute_fisher(self, data):
        """计算Fisher信息矩阵（参数重要性）"""
        fisher = {}
        for name, param in self.model.named_parameters():
            fisher[name] = torch.zeros_like(param)
        
        self.model.eval()
        for batch in data:
            self.model.zero_grad()
            output = self.model(batch)
            loss = output.loss
            loss.backward()
            
            for name, param in self.model.named_parameters():
                if param.grad is not None:
                    fisher[name] += param.grad.data ** 2
        
        fisher = {k: v / len(data) for k, v in fisher.items()}
        return fisher
    
    def penalty(self):
        """EWC正则化项"""
        loss = 0
        for name, param in self.model.named_parameters():
            if name in self.old_params:
                fisher = self.fisher_information[name]
                old_param = self.old_params[name]
                loss += (fisher * (param - old_param) ** 2).sum()
        return loss

def train_with_ewc(model, new_data, ewc, lambda_ewc=100):
    """带EWC正则化的训练"""
    for batch in new_data:
        loss = model.compute_loss(batch)
        ewc_loss = ewc.penalty()
        total_loss = loss + lambda_ewc * ewc_loss
        total_loss.backward()
        optimizer.step()

优势：不需要存储旧数据、理论基础扎实 劣势：Fisher矩阵计算开销大、超参数敏感

方法三：LoRA增量适配

为每个新任务/知识领域训练独立的LoRA适配器：

class ContinualLoRA:
    def __init__(self, base_model):
        self.base_model = base_model
        self.adapters = {}
    
    def learn_new_domain(self, domain_name, train_data, config):
        """为新区域名训练独立LoRA适配器"""
        # 冻结基础模型和已有适配器
        for param in self.base_model.parameters():
            param.requires_grad = False
        
        # 创建新适配器
        lora_config = LoraConfig(
            r=config.get('r', 16),
            lora_alpha=config.get('alpha', 32),
            target_modules=config.get('targets', ['q_proj', 'v_proj'])
        )
        
        self.base_model.add_adapter(domain_name, lora_config)
        self.base_model.set_adapter(domain_name)
        
        # 仅训练新适配器
        train(self.base_model, train_data)
        
        # 保存适配器
        self.adapters[domain_name] = {
            'config': lora_config,
            'domain': domain_name
        }
    
    def inference(self, input, domain=None):
        """推理时选择适配器"""
        if domain and domain in self.adapters:
            self.base_model.set_adapter(domain)
        else:
            # 自动选择最相关的适配器
            domain = self.select_adapter(input)
            if domain:
                self.base_model.set_adapter(domain)
        
        return self.base_model.generate(input)
    
    def select_adapter(self, input):
        """根据输入选择最适配的适配器"""
        # 使用分类器或相似度匹配
        scores = {}
        for name in self.adapters:
            scores[name] = self.domain_classifier(input, name)
        return max(scores, key=scores.get) if scores else None

优势：零遗忘（每个领域独立）、可扩展性好 劣势：适配器数量增长、推理时需要选择适配器

方法四：知识蒸馏

用旧模型作为教师，在学新知识时保持旧模型的行为：

def train_with_distillation(student_model, teacher_model, new_data, 
                              alpha=0.5, temperature=2.0):
    """带知识蒸馏的持续学习"""
    for batch in new_data:
        # 学生模型学习新任务
        student_outputs = student_model(batch)
        new_loss = student_outputs.loss
        
        # 教师模型提供旧知识
        with torch.no_grad():
            teacher_outputs = teacher_model(batch)
        
        # 蒸馏损失
        distill_loss = KL_divergence(
            student_outputs.logits / temperature,
            teacher_outputs.logits / temperature
        ) * (temperature ** 2)
        
        # 总损失
        total_loss = alpha * new_loss + (1 - alpha) * distill_loss
        total_loss.backward()
        optimizer.step()

工程实践

数据管理

class ContinualLearningDataManager:
    def __init__(self):
        self.task_history = []  # 任务历史
        self.data_archive = {}  # 数据归档
    
    def add_task(self, task_name, train_data, eval_data, metadata):
        """添加新学习任务"""
        self.task_history.append({
            'name': task_name,
            'timestamp': datetime.now(),
            'metadata': metadata,
            'eval_data': eval_data
        })
        
        # 归档部分数据用于回放
        self.data_archive[task_name] = sample_subset(train_data, ratio=0.1)
    
    def get_replay_data(self, current_task, size=100):
        """获取回放数据"""
        replay = []
        for task in self.task_history:
            if task['name'] != current_task:
                replay.extend(random.sample(
                    self.data_archive[task['name']],
                    min(size, len(self.data_archive[task['name']]))
                ))
        return replay

持续评估

def continual_evaluation(model, task_history):
    """评估模型在所有历史任务上的表现"""
    results = {}
    for task in task_history:
        score = evaluate(model, task['eval_data'])
        results[task['name']] = {
            'current_score': score,
            'original_score': task.get('original_score', score),
            'forgetting': task.get('original_score', score) - score
        }
    
    # 计算平均遗忘
    avg_forgetting = np.mean([r['forgetting'] for r in results.values()])
    
    return {
        'per_task': results,
        'avg_forgetting': avg_forgetting,
        'avg_performance': np.mean([r['current_score'] for r in results.values()])
    }

遗忘告警

设置遗忘监控阈值：当旧任务表现下降超过5%时告警；当综合表现下降时自动回滚到上一检查点；记录每次学习后的完整评估报告。

方案选择建议

场景推荐方法理由
可保留旧数据经验回放最简单有效
数据隐私限制EWC不需存储旧数据
多领域知识LoRA增量零遗忘、可扩展
平衡新旧知识知识蒸馏教师指导保持旧能力
综合方案回放+LoRA+蒸馏多重保障

结语

持续学习是让大模型保持"鲜活"的关键能力。在没有完美解决方案的当下，组合使用多种方法是工程实践的主流选择。核心原则是：建立持续评估机制监控遗忘程度、保留旧知识的"锚点"（数据、参数或适配器）、控制每次学习的强度避免剧烈遗忘。随着模型规模的增大和方法论的成熟，持续学习将逐步从研究走向可靠的工程实践。

加入讨论

这篇文章有姊妹讨论帖在硅基AGI论坛 — 全球首个碳基硅基认知交流平台。

碳基与硅基的智慧碰撞，认知差异创造无限可能。