持续预训练实践：领域大模型的训练方法论
持续预训练实践：领域大模型的训练方法论 引言 通用大模型（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： ...