引言

大模型的增量训练（Continued Pretraining，CPT）是指在已有预训练模型基础上，使用新数据继续训练，以注入新知识或适应新领域。与从零训练相比，增量训练大幅降低计算成本，同时保留模型的通用能力。但增量训练面临独特的技术挑战：知识注入与遗忘的平衡、训练稳定性、数据配比优化等。本文分享大模型增量训练的工程实践。

增量训练的定位

与其他训练方式的关系

训练方式数据量训练成本知识注入深度适用场景
从零预训练万亿级token极高最深构建全新基础模型
增量训练（CPT）百亿级token高深知识更新/领域注入
SFT微调百万级中中指令跟随能力
LoRA微调万级低浅轻量适配

何时需要增量训练

  • 模型需要掌握大量新的领域知识（如从通用模型到医疗模型）
  • 领域语言模式与通用文本差异大（如法律条文、代码）
  • 需要更新时间敏感的知识（如新法规、新技术）
  • 现有模型在目标领域的幻觉率过高

不推荐增量训练的场景：仅需调整输出格式（用SFT）、仅需少量领域知识（用RAG）、计算资源不足。

数据准备

数据来源与处理

class IncrementalDataPipeline:
    def __init__(self, tokenizer, max_length=4096):
        self.tokenizer = tokenizer
        self.max_length = max_length
    
    def process(self, raw_data_sources):
        """完整数据处理流水线"""
        all_data = []
        
        for source in raw_data_sources:
            # 1. 格式统一
            text = self.normalize_format(source)
            
            # 2. 质量过滤
            if not self.quality_filter(text):
                continue
            
            # 3. 去重
            text = self.deduplicate(text)
            
            # 4. 分词
            tokens = self.tokenizer.encode(text)
            
            # 5. 长文档切分
            chunks = self.split_long_text(tokens)
            
            all_data.extend(chunks)
        
        # 6. 数据配比
        mixed_data = self.mix_data(all_data)
        
        return mixed_data
    
    def quality_filter(self, text):
        """质量过滤"""
        # 去除过短文本
        if len(text) < 100:
            return False
        # 去除重复率高的文本
        if self.repetition_ratio(text) > 0.3:
            return False
        # 去除乱码
        if self.garbled_ratio(text) > 0.1:
            return False
        return True
    
    def split_long_text(self, tokens):
        """长文本切分，保留上下文"""
        chunks = []
        overlap = 512  # 重叠token数
        
        for i in range(0, len(tokens), self.max_length - overlap):
            chunk = tokens[i:i + self.max_length]
            chunks.append(chunk)
            if i + self.max_length >= len(tokens):
                break
        
        return chunks
    
    def mix_data(self, domain_data, general_ratio=0.3):
        """领域数据与通用数据混合"""
        domain_size = len(domain_data)
        general_size = int(domain_size * general_ratio / (1 - general_ratio))
        
        general_data = self.load_general_data(general_size)
        
        mixed = domain_data + general_data
        random.shuffle(mixed)
        return mixed

数据配比策略

数据配比是增量训练最关键的超参数之一：

# 推荐配比方案
data_mix_strategies = {
    '轻度领域注入': {
        'domain': 0.3,
        'general': 0.7,
        'use_case': '在保持通用能力基础上增加领域知识'
    },
    '中度领域注入': {
        'domain': 0.5,
        'general': 0.5,
        'use_case': '平衡领域专业性和通用能力'
    },
    '深度领域注入': {
        'domain': 0.7,
        'general': 0.3,
        'use_case': '构建领域专用模型'
    }
}

通用数据的作用是"锚定"模型的通用能力，防止灾难性遗忘。推荐的通用数据来源：原模型的部分预训练数据、高质量通用语料（维基百科、书籍）、代码数据（保持推理能力）。

训练配置

基础训练配置

import torch
from transformers import (
    AutoModelForCausalLM,
    AutoTokenizer,
    TrainingArguments,
    Trainer,
    DataCollatorForLanguageModeling
)

def setup_incremental_training(model_path, train_data):
    # 加载模型和分词器
    model = AutoModelForCausalLM.from_pretrained(
        model_path,
        torch_dtype=torch.bfloat16,
        device_map="auto",
        attn_implementation="flash_attention_2"
    )
    tokenizer = AutoTokenizer.from_pretrained(model_path)
    
    # 训练参数
    training_args = TrainingArguments(
        output_dir="./incremental_output",
        overwrite_output_dir=True,
        
        # 训练轮数：增量训练通常1-3轮
        num_train_epochs=2,
        
        # 批量大小
        per_device_train_batch_size=4,
        gradient_accumulation_steps=8,  # 有效batch=32
        
        # 学习率：增量训练使用较小学习率
        learning_rate=5e-6,  # 比预训练低1个数量级
        lr_scheduler_type="cosine",
        warmup_ratio=0.05,
        
        # 精度
        bf16=True,
        
        # 梯度裁剪
        max_grad_norm=1.0,
        
        # 日志和保存
        logging_steps=50,
        save_strategy="steps",
        save_steps=1000,
        save_total_limit=3,
        
        # 深度加速
        gradient_checkpointing=True,
        flash_attention_2=True,
    )
    
    # 数据整理器
    data_collator = DataCollatorForLanguageModeling(
        tokenizer=tokenizer,
        mlm=False  # 因果语言模型
    )
    
    trainer = Trainer(
        model=model,
        args=training_args,
        train_dataset=train_data,
        data_collator=data_collator,
    )
    
    return trainer

学习率策略

增量训练的学习率选择是关键：

# 学习率选择建议
lr_recommendations = {
    '7B模型': 1e-5,      # 小模型可用较大学习率
    '13B模型': 5e-6,
    '70B模型': 1e-6,     # 大模型需要更小学习率
    '130B模型': 5e-7,
}

# 余弦衰减调度
def cosine_schedule_with_warmup(optimizer, num_warmup_steps, num_training_steps):
    """带预热的余弦衰减调度"""
    def lr_lambda(current_step):
        if current_step < num_warmup_steps:
            return float(current_step) / float(max(1, num_warmup_steps))
        progress = float(current_step - num_warmup_steps) / float(
            max(1, num_training_steps - num_warmup_steps))
        return 0.5 * (1.0 + torch.cos(torch.pi * progress))
    
    return torch.optim.lr_scheduler.LambdaLR(optimizer, lr_lambda)

训练稳定性保障

class StableTrainer(Trainer):
    """稳定的增量训练器"""
    
    def __init__(self, *args, loss_spike_threshold=3.0, **kwargs):
        super().__init__(*args, **kwargs)
        self.loss_history = []
        self.loss_spike_threshold = loss_spike_threshold
    
    def training_step(self, model, inputs):
        # 计算loss
        loss = super().training_step(model, inputs)
        
        # 检测loss异常
        self.loss_history.append(loss.item())
        if len(self.loss_history) > 100:
            recent_avg = np.mean(self.loss_history[-100:])
            if loss.item() > recent_avg * self.loss_spike_threshold:
                print(f"WARNING: Loss spike detected: {loss.item():.4f} vs avg {recent_avg:.4f}")
                # 可选：回滚到上一个检查点
        
        # 梯度异常检测
        for name, param in model.named_parameters():
            if param.grad is not None:
                if torch.isnan(param.grad).any():
                    print(f"ERROR: NaN gradient in {name}")
                    param.grad.data.zero_()  # 清零梯度
                if torch.isinf(param.grad).any():
                    print(f"ERROR: Inf gradient in {name}")
                    param.grad.data.zero_()
        
        return loss

分布式训练

DeepSpeed配置

// ds_config.json
{
    "bf16": {
        "enabled": true
    },
    "zero_optimization": {
        "stage": 2,
        "offload_optimizer": {
            "enabled": true,
            "device": "cpu"
        },
        "allgather_partitions": true,
        "allgather_bucket_size": 5e8,
        "reduce_scatter": true,
        "reduce_bucket_size": 5e8,
        "overlap_comm": true
    },
    "gradient_accumulation_steps": 8,
    "gradient_clipping": 1.0,
    "train_batch_size": "auto",
    "train_micro_batch_size_per_gpu": 4,
    "wall_clock_breakdown": false
}

多节点训练启动

# 多节点训练启动脚本
deepspeed \
    --num_gpus=8 \
    --num_nodes=4 \
    --hostfile=hostfile \
    train_incremental.py \
    --model_path /models/base_model \
    --data_path /data/domain_data \
    --output_dir /output/incremental_model \
    --deepspeed ds_config.json

效果评估

持续评估流程

class IncrementalEvaluator:
    def __init__(self, domain_benchmark, general_benchmark):
        self.domain_benchmark = domain_benchmark
        self.general_benchmark = general_benchmark
        self.history = []
    
    def evaluate_checkpoint(self, model, checkpoint_path):
        """评估检查点"""
        results = {
            'checkpoint': checkpoint_path,
            'domain_score': self.run_benchmark(model, self.domain_benchmark),
            'general_score': self.run_benchmark(model, self.general_benchmark),
            'timestamp': datetime.now()
        }
        
        results['domain_gain'] = results['domain_score'] - self.baseline_domain
        results['general_loss'] = self.baseline_general - results['general_score']
        results['net_benefit'] = results['domain_gain'] - results['general_loss']
        
        self.history.append(results)
        return results
    
    def find_optimal_checkpoint(self):
        """找到最佳检查点"""
        best = max(self.history, key=lambda x: x['net_benefit'])
        return best

评估维度

评估维度评估方法目标
领域知识领域QA基准显著提升
通用能力MMLU/HellaSwag等下降<5%
代码能力HumanEval下降<3%
数学推理GSM8K下降<3%
幻觉率事实核查基准显著降低
安全性安全评估集不退化

常见问题与解决

问题一：训练发散

症状：loss突然飙升或变为NaN 解决方案：降低学习率（1/2或1/5）、增大warmup步数、启用梯度裁剪、检查数据中是否有异常样本

问题二：过度遗忘

症状：通用基准分数大幅下降 解决方案：增加通用数据比例、减少训练轮数、降低学习率、使用EWC等正则化方法

问题三：知识注入不足

症状：领域基准提升不明显 解决方案：增加领域数据比例、适当增加训练轮数、检查领域数据质量、考虑分阶段训练

问题四：训练速度慢

解决方案：启用Flash Attention 2、使用梯度检查点、优化数据加载Pipeline、使用更高效的分布式策略

结语

大模型增量训练是在已有模型基础上注入领域知识和更新信息的有效手段。成功的关键在于：精心准备和配比数据、选择合适的学习率和训练轮数、建立全面的评估体系监控训练效果、确保训练稳定性。增量训练不是简单的"继续训练"，而是一项需要精细调控的工程任务。随着工具链的成熟和最佳实践的积累，增量训练正在成为大模型领域适配的标准手段之一。

加入讨论

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

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