持续预训练

什么是持续预训练 持续预训练（Continual Pre-Training, CPT）是在已有的预训练模型基础上，用领域数据继续训练，让模型"学会"领域知识。与 SFT 不同，CPT 的目标不是学习特定任务格式，而是注入知识本身。 预训练 (PT) 持续预训练 (CPT) 监督微调 (SFT) 通用语料 ──→ 基础模型 ──→ 领域模型 ──→ 任务模型 (万亿token) (百亿token) (万级样本) 1. 数据工程 数据来源与采集 class CPTDataCollector: def __init__(self, domain: str): self.domain = domain def collect(self) -> dict: sources = { "medical": self._collect_medical, "legal": self._collect_legal, "finance": self._collect_finance, "code": self._collect_code, } return sources.get(self.domain, self._collect_general)() def _collect_medical(self): return { "academic_papers": "PubMed Central 全文", "textbooks": "医学教材电子版", "clinical_guidelines": "各国临床指南", "drug_labels": "FDA/NMPA 药品说明书", "medical_encyclopedia": "医学百科全书", "estimated_tokens": "约 50B tokens" } def _collect_legal(self): return { "laws_regulations": "法律法规数据库", "court_cases": "裁判文书网", "legal_commentary": "法学期刊", "legal_textbooks": "法学教材", "contracts": "合同模板库", "estimated_tokens": "约 30B tokens" } 数据配比：最关键的超参数 数据配比决定了模型在领域知识和通用能力之间的平衡。 class DataMixer: def __init__(self): # 2026 年推荐配比（基于实验） self.ratios = { "domain_heavy": { # 重度领域适配 "domain": 0.7, "general": 0.2, "code": 0.05, "math": 0.05 }, "domain_balanced": { # 平衡适配（推荐） "domain": 0.5, "general": 0.35, "code": 0.1, "math": 0.05 }, "domain_light": { # 轻度适配 "domain": 0.3, "general": 0.55, "code": 0.1, "math": 0.05 } } def mix(self, domain_data: list, general_data: list, strategy: str = "domain_balanced"): ratio = self.ratios[strategy] total_tokens = 10_000_000_000 # 10B tokens domain_tokens = int(total_tokens * ratio["domain"]) general_tokens = int(total_tokens * ratio["general"]) mixed = [] mixed.extend(self._sample(domain_data, domain_tokens)) mixed.extend(self._sample(general_data, general_tokens)) # 打乱 random.shuffle(mixed) return mixed 配比实验结果 配比策略 领域任务 通用任务 代码 数学 灾难性遗忘 100% 领域 88% 42% 35% 38% 严重 ❌ 70% 领域 85% 68% 52% 55% 中等 ⚠️ 50% 领域 82% 75% 68% 65% 轻微 ✅ 30% 领域 76% 82% 72% 70% 无 ✅ 0% (baseline) 65% 85% 75% 72% 无 ✅ 结论：50% 领域数据是最佳平衡点。 ...

持续预训练实践：领域大模型的训练方法论 引言 通用大模型（GPT-4、LLaMA-3）在医疗、法律、金融等垂直领域常出现专业术语理解偏差、领域知识幻觉、行业惯例缺失等问题。全量预训练成本过高，而 SFT（指令微调）又不足以注入深层领域知识。 持续预训练（Continual Pre-Training, CPT）是两者之间的平衡点：在通用基座上用领域语料继续预训练，让模型深层理解领域知识，同时保留通用能力。 1. CPT 在训练体系中的定位 预训练（PT）→ 持续预训练（CPT）→ 指令微调（SFT）→ 偏好对齐（DPO/RLHF） ↑ 本文重点 阶段 数据量 数据形式 学习目标 参数更新 PT 万亿 tokens 纯文本 Next token prediction 全量 CPT 百亿~千亿 tokens 纯文本（领域） Next token prediction 全量或部分 SFT 百万~千万条 指令-响应对 指令跟随 全量或 PEFT DPO 数万~数十万条 偏好对 偏好优化 全量或 PEFT 1.1 CPT vs SFT：本质区别 维度 CPT SFT 数据 无标注领域文本 人工标注指令对 目标 学习领域知识分布 学习指令跟随格式 深度 改变模型内部知识表征 调整输出行为 数据量 10B-100B tokens 1M-10M 条 灾难性遗忘 严重（需缓解） 较轻 2. 领域数据构建 2.1 数据来源与采集 from typing import List, Dict import json import os class DomainDataBuilder: """领域预训练数据构建器""" def __init__(self, domain: str): self.domain = domain # "medical", "legal", "financial", etc. def collect_sources(self) -> Dict[str, List[str]]: """定义数据来源""" sources = { "medical": { "textbooks": ["医学教科书 PDF", "诊疗指南"], "papers": ["PubMed 论文", "中文医学期刊"], "records": ["脱敏病历", "医学百科"], "guidelines": ["临床路径", "用药指南"], "qa": ["医学问答", "丁香园问答"], }, "legal": { "textbooks": ["法学教材", "法律释义"], "papers": ["法学期刊论文"], "records": ["裁判文书", "法律案例"], "guidelines": ["司法解释", "部门规章"], "qa": ["法律咨询问答"], }, "financial": { "textbooks": ["金融学教材", "CFA教材"], "papers": ["金融研究论文"], "reports": ["年报", "券商研报"], "guidelines": ["会计准则", "监管文件"], "qa": ["金融问答", "投资者教育"], }, } return sources.get(self.domain, {}) def estimate_tokens(self, file_sizes_mb: List[float]) -> int: """估算 token 数（中文约 1.5 字/token，英文约 0.75 词/token）""" total_chars = sum(size * 1024 * 1024 * 3 for size in file_sizes_mb) # 假设中文3字节 return int(total_chars / 1.5) # 粗略估算 2.2 数据清洗管道 import re from typing import Optional class DataCleaner: """预训练数据清洗""" @staticmethod def clean_text(text: str) -> Optional[str]: """单文档清洗""" # 1. 去除 HTML 标签 text = re.sub(r'<[^>]+>', '', text) # 2. 去除 URL text = re.sub(r'https?://\S+', '', text) # 3. 去除多余空白 text = re.sub(r'\s+', ' ', text).strip() # 4. 去除乱码（非可打印字符） text = re.sub(r'[\x00-\x08\x0b\x0c\x0e-\x1f\x7f-\x9f]', '', text) # 5. 去重检测（文档级） if len(text) < 100: # 太短，丢弃 return None # 6. 质量过滤：字母/汉字占比 alpha_ratio = len(re.findall(r'[a-zA-Z\u4e00-\u9fff]', text)) / len(text) if alpha_ratio < 0.5: # 非文字内容太多 return None return text @staticmethod def deduplicate(texts: List[str], threshold: float = 0.8) -> List[str]: """MinHash 去重""" from datasketch import MinHash, MinHashLSH lsh = MinHashLSH(threshold=threshold, num_perm=128) unique_texts = [] for i, text in enumerate(texts): m = MinHash(num_perm=128) # 按 5-gram 分词 words = [text[j:j+5] for j in range(0, len(text)-4, 3)] for word in words: m.update(word.encode('utf-8')) # 检查是否重复 result = lsh.query(m) if not result: lsh.insert(str(i), m) unique_texts.append(text) return unique_texts 2.3 数据混合比例 def build_mixed_dataset( domain_data: List[str], # 领域文本 general_data: List[str], # 通用文本（如预训练数据子集） domain_ratio: float = 0.7, # 领域数据占比 ) -> List[str]: """ 混合领域数据和通用数据，防止灾难性遗忘 """ import random n_domain = int(len(domain_data) * domain_ratio / (1 - domain_ratio)) n_domain = min(n_domain, len(domain_data)) sampled_domain = random.sample(domain_data, n_domain) sampled_general = random.sample(general_data, len(domain_data) - n_domain + len(domain_data)) # 交错排列（避免连续大量同类型数据） mixed = [] domain_iter = iter(sampled_domain) general_iter = iter(sampled_general) while True: try: mixed.append(next(domain_iter)) except StopIteration: pass try: mixed.append(next(general_iter)) except StopIteration: pass if not (domain_iter or general_iter): break return mixed # 推荐混合比例 ratio_guide = { "medical": {"domain": 0.6, "general": 0.3, "code": 0.1}, # 医疗 "legal": {"domain": 0.65, "general": 0.25, "code": 0.10}, # 法律 "financial": {"domain": 0.55, "general": 0.35, "code": 0.10}, # 金融 "code": {"domain": 0.7, "general": 0.2, "code": 0.10}, # 代码 "general_domain": {"domain": 0.5, "general": 0.4, "code": 0.1}, # 通用领域 } 3. 训练策略 3.1 学习率调度 CPT 的学习率通常远低于预训练阶段，但高于 SFT： ...