SFT

大模型的能力由预训练决定，但大模型的行为由对齐决定。一个能力强大但不对齐的模型，就像一个天才但不守规矩的员工——潜力越大，风险越大。本文将从工程实践角度，完整梳理从数据标注到RLHF的对齐流程。 一、对齐的全景图 大模型对齐不是单一技术，而是一个多阶段流水线： 预训练模型（Base Model） ↓ 有监督微调（SFT） — 学习"怎么回答" ↓ 奖励模型训练（RM） — 学习"什么是好回答" ↓ 强化学习优化（RLHF/DPO） — 优化"回答得更好" ↓ 安全对齐（Safety） — 确保"不回答不该回答的" 每个阶段都有不同的数据需求、训练方法和评估标准。 二、数据标注：对齐的基石 2.1 SFT数据标注 SFT数据是"指令-回复"对，教模型如何回答问题。看似简单，但质量差异巨大。 常见标注问题： 标注者倾向于写"教科书式"回答，缺乏多样性 不同标注者风格不一致，导致模型输出不稳定 安全相关的边界case标注不一致 最佳实践： 标注规范结构: 1. 任务定义: 清晰定义每种指令类型的预期回答格式 2. 质量标准: - 准确性: 信息正确，无幻觉 - 完整性: 覆盖问题要点 - 简洁性: 不冗余 - 安全性: 不含有害内容 3. 风格指南: 自然口语化，不要过于"AI腔" 4. 边界case: - 事实性问题 → 给出准确答案+来源 - 观点性问题 → 给出多角度分析 - 创意性问题 → 发挥创意但不越界 2.2 偏好数据标注（RLHF/DPO用） 偏好数据是"同一问题的两个回复，标注哪个更好"。这是对齐的核心数据。 ...

大模型微调的数据工程全流程 在大模型微调中，数据质量的重要性远超算法选择。同样的微调方法，高质量数据可以带来20-30%的效果提升，而低质量数据不仅无益，还可能损害模型的通用能力。本文系统梳理微调数据工程的全流程。 数据采集 来源规划 微调数据应该来自多个渠道，避免单一来源的偏见： 真实交互数据：从生产环境中收集用户与模型的交互数据，筛选高质量的对话。这是最有价值的数据来源，因为它反映了真实的使用模式。需要注意的是用户隐私保护和数据脱敏。 合成数据：使用更强大的模型生成训练数据。合成数据的优势是可以针对特定能力定向生成，劣势是可能继承教师模型的偏见。我们建议合成数据不超过总数据量的40%。 人工标注：领域专家编写的高质量样本。成本最高但质量最好，适用于关键能力的数据补充。 公开数据集：学术数据集和开源数据。需要注意许可协议和数据质量参差不齐的问题。 采样策略 数据采样的目标是确保训练集覆盖目标能力的各个方面。我们采用"能力矩阵"方法——定义需要微调的能力维度和难度等级，确保每个格子有足够的数据样本。 数据清洗 去重 数据去重看似简单，实则有不少坑。精确去重（完全相同的样本）只能去除最明显的冗余，大量近似重复的样本仍会浪费训练资源并导致过拟合。 我们采用了MinHash+LSH进行模糊去重，将Jaccard相似度超过0.7的样本对标记为近似重复，保留质量更高的那条。在一个10万条数据集上，模糊去重额外移除了约8%的样本。 质量过滤 低质量数据是微调效果的杀手。我们建立了多级质量过滤管道： 规则过滤：移除包含乱码、编码错误、过短或过长、重复模式严重的样本。这一步快速且成本低，可以过滤约10-15%的明显低质量数据。 模型过滤：使用一个训练好的质量评分模型对每条数据打分，低于阈值的被过滤。评分模型可以是一个在人类标注的质量数据上训练的小型分类器，也可以使用LLM-as-Judge。 人工抽检：随机抽取5-10%的样本进行人工检查，评估自动过滤的准确性，并校准过滤阈值。 格式规范化 微调数据的格式必须与模型的训练格式一致。常见问题包括：特殊token的不一致、对话角色的混淆、多轮对话中上下文的截断方式不一致等。 我们定义了严格的格式schema，使用JSON Schema验证每条数据的结构完整性。不符合schema的样本被标记并修复或丢弃。 数据增强 指令改写 同一条指令可以有多种表达方式。“帮我总结这篇文章"和"请概括以下内容的要点"在语义上等价但表达不同。通过指令改写增加表达多样性，可以提高模型的泛化能力。 我们使用LLM自动改写指令，每条原始指令生成3-5个改写版本。关键是要保持语义不变——我们用一个语义相似度模型过滤掉语义偏移过大的改写。 难度分层 同一任务的不同难度级别对模型学习很重要。如果数据全是简单任务，模型无法处理复杂场景；如果全是困难任务，模型难以学到基本模式。 我们为每个任务类型准备了三个难度级别的数据：简单（单步推理）、中等（2-3步推理）、困难（多步推理+边缘情况）。比例约为4:4:2。 数据混合策略 微调数据不仅要覆盖目标任务，还需要包含一定比例的通用数据，以防止灾难性遗忘——模型在学会新能力的同时忘记原有能力。 我们的数据混合方案： 60% 目标任务数据 25% 通用对话数据（从预训练数据中采样） 10% 安全对齐数据 5% 代码和推理数据 这个比例不是固定的，需要根据具体任务和模型表现进行调整。关键是监控模型在通用benchmark上的表现，确保微调没有显著损害通用能力。 质量评估 自动评估 在微调前，我们使用以下指标评估数据质量： 多样性指标：计算数据集的语义多样性分布，确保覆盖面足够 难度分布：统计不同难度级别的样本比例 毒性检测：使用分类器检测并移除有毒内容 偏见检测：检查数据是否存在性别、种族等方面的系统性偏见 消融实验 最终的数据质量需要通过微调实验来验证。我们采用消融实验——分别用完整数据集和去除某一部分后的数据集进行微调，比较效果差异。这帮助我们理解哪些数据对模型能力贡献最大。 结语 数据工程是微调中最耗时但也最值得投入的环节。在我们的实践中，数据工程占整个微调项目时间的60-70%，但也是回报最高的投入。好的数据可以让简单的微调方法超越复杂方法在差数据上的表现。数据为王，在微调领域依然是真理。 本文同步发布于 硅基AGI论坛

大模型微调的数据工程全流程 在大模型微调中，数据质量的重要性远超算法选择。同样的微调方法，高质量数据可以带来20-30%的效果提升，而低质量数据不仅无益，还可能损害模型的通用能力。本文系统梳理微调数据工程的全流程。 数据采集 来源规划 微调数据应该来自多个渠道，避免单一来源的偏见： 真实交互数据：从生产环境中收集用户与模型的交互数据，筛选高质量的对话。这是最有价值的数据来源，因为它反映了真实的使用模式。需要注意的是用户隐私保护和数据脱敏。 合成数据：使用更强大的模型生成训练数据。合成数据的优势是可以针对特定能力定向生成，劣势是可能继承教师模型的偏见。我们建议合成数据不超过总数据量的40%。 人工标注：领域专家编写的高质量样本。成本最高但质量最好，适用于关键能力的数据补充。 公开数据集：学术数据集和开源数据。需要注意许可协议和数据质量参差不齐的问题。 采样策略 数据采样的目标是确保训练集覆盖目标能力的各个方面。我们采用"能力矩阵"方法——定义需要微调的能力维度和难度等级，确保每个格子有足够的数据样本。 数据清洗 去重 数据去重看似简单，实则有不少坑。精确去重（完全相同的样本）只能去除最明显的冗余，大量近似重复的样本仍会浪费训练资源并导致过拟合。 我们采用了MinHash+LSH进行模糊去重，将Jaccard相似度超过0.7的样本对标记为近似重复，保留质量更高的那条。在一个10万条数据集上，模糊去重额外移除了约8%的样本。 质量过滤 低质量数据是微调效果的杀手。我们建立了多级质量过滤管道： 规则过滤：移除包含乱码、编码错误、过短或过长、重复模式严重的样本。这一步快速且成本低，可以过滤约10-15%的明显低质量数据。 模型过滤：使用一个训练好的质量评分模型对每条数据打分，低于阈值的被过滤。评分模型可以是一个在人类标注的质量数据上训练的小型分类器，也可以使用LLM-as-Judge。 人工抽检：随机抽取5-10%的样本进行人工检查，评估自动过滤的准确性，并校准过滤阈值。 格式规范化 微调数据的格式必须与模型的训练格式一致。常见问题包括：特殊token的不一致、对话角色的混淆、多轮对话中上下文的截断方式不一致等。 我们定义了严格的格式schema，使用JSON Schema验证每条数据的结构完整性。不符合schema的样本被标记并修复或丢弃。 数据增强 指令改写 同一条指令可以有多种表达方式。“帮我总结这篇文章"和"请概括以下内容的要点"在语义上等价但表达不同。通过指令改写增加表达多样性，可以提高模型的泛化能力。 我们使用LLM自动改写指令，每条原始指令生成3-5个改写版本。关键是要保持语义不变——我们用一个语义相似度模型过滤掉语义偏移过大的改写。 难度分层 同一任务的不同难度级别对模型学习很重要。如果数据全是简单任务，模型无法处理复杂场景；如果全是困难任务，模型难以学到基本模式。 我们为每个任务类型准备了三个难度级别的数据：简单（单步推理）、中等（2-3步推理）、困难（多步推理+边缘情况）。比例约为4:4:2。 数据混合策略 微调数据不仅要覆盖目标任务，还需要包含一定比例的通用数据，以防止灾难性遗忘——模型在学会新能力的同时忘记原有能力。 我们的数据混合方案： 60% 目标任务数据 25% 通用对话数据（从预训练数据中采样） 10% 安全对齐数据 5% 代码和推理数据 这个比例不是固定的，需要根据具体任务和模型表现进行调整。关键是监控模型在通用benchmark上的表现，确保微调没有显著损害通用能力。 质量评估 自动评估 在微调前，我们使用以下指标评估数据质量： 多样性指标：计算数据集的语义多样性分布，确保覆盖面足够 难度分布：统计不同难度级别的样本比例 毒性检测：使用分类器检测并移除有毒内容 偏见检测：检查数据是否存在性别、种族等方面的系统性偏见 消融实验 最终的数据质量需要通过微调实验来验证。我们采用消融实验——分别用完整数据集和去除某一部分后的数据集进行微调，比较效果差异。这帮助我们理解哪些数据对模型能力贡献最大。 结语 数据工程是微调中最耗时但也最值得投入的环节。在我们的实践中，数据工程占整个微调项目时间的60-70%，但也是回报最高的投入。好的数据可以让简单的微调方法超越复杂方法在差数据上的表现。数据为王，在微调领域依然是真理。 本文同步发布于 硅基AGI论坛

引言 指令微调（Instruction Tuning / SFT）是将基础模型变成对话助手的关键步骤。2026年的经验表明：微调效果90%取决于数据质量，10%取决于训练方法。 一、数据格式 { "messages": [ {"role": "system", "content": "你是一个专业的编程助手。"}, {"role": "user", "content": "解释什么是递归"}, {"role": "assistant", "content": "递归是一种编程技术..."} ] } 二、数据构建策略 2.1 种子数据+扩展 class InstructionDataBuilder: async def build_from_seeds(self, seed_instructions, expansion_rate=10): """从种子指令扩展""" expanded = [] for seed in seed_instructions: # 1. 改写指令 rewrites = await self.rewrite_instruction(seed, n=expansion_rate//2) # 2. 生成变体 variants = await self.generate_variants(seed, n=expansion_rate//2) expanded.extend(rewrites + variants) return expanded async def rewrite_instruction(self, instruction, n=5): """改写指令""" prompt = f""" 将以下指令改写为{n}个不同表述，保持意思相同: 原始: {instruction} """ result = await self.llm.call(prompt) return result["rewrites"] 2.2 Self-Instruct class SelfInstruct: async def generate(self, seed_tasks, num_tasks=1000): """Self-Instruct生成""" tasks = list(seed_tasks) while len(tasks) < num_tasks: # 1. 随机选择种子任务作为示例 examples = random.sample(tasks, min(3, len(tasks))) # 2. 生成新指令 new_instruction = await self.llm.generate( f"基于以下示例生成一个新的指令:\n{examples}" ) # 3. 过滤低质量 if self.is_quality(new_instruction): # 4. 生成回答 response = await self.llm.generate(new_instruction) tasks.append({ "instruction": new_instruction, "response": response }) return tasks 2.3 Evol-Instruct class EvolInstruct: """逐步进化指令复杂度""" async def evolve(self, instruction): """进化指令""" strategies = [ "增加约束条件", "增加推理步骤", "增加领域深度", "增加多步骤要求", "增加边界条件处理" ] strategy = random.choice(strategies) prompt = f""" 指令: {instruction} 请通过以下方式增加这个指令的复杂度: {strategy} """ return await self.llm.call(prompt) 三、数据质量 3.1 质量过滤 class QualityFilter: def filter(self, dataset): filtered = [] for sample in dataset: # 1. 长度检查 if len(sample["response"]) < 10: continue # 2. 重复检查 if self.is_duplicate(sample, filtered): continue # 3. 格式检查 if not self.validate_format(sample): continue # 4. 内容质量 if not self.check_content_quality(sample): continue filtered.append(sample) return filtered def check_content_quality(self, sample): """内容质量检查""" response = sample["response"] # 不应该是"我不知道"之类的无效回答 if response.strip() in ["我不知道", "无法回答", "I don't know"]: return False # 不应该是重复内容 if len(set(response.split())) / len(response.split()) < 0.3: return False return True 3.2 去重 class Deduplicator: def deduplicate(self, dataset): """多级去重""" # 1. 精确去重 seen = set() deduped = [] for sample in dataset: key = hash(sample["instruction"]) if key not in seen: seen.add(key) deduped.append(sample) # 2. 模糊去重（MinHash） from datasketch import MinHash minhashes = [] for sample in deduped: mh = MinHash(num_perm=128) for word in sample["instruction"].split(): mh.update(word.encode()) minhashes.append(mh) # 移除相似度>0.8的 final = [] for i, sample in enumerate(deduped): is_dup = False for j in range(len(final)): if minhashes[i].jaccard(minhashes[final[j]["index"]]) > 0.8: is_dup = True break if not is_dup: final.append({"index": i, "sample": sample}) return [f["sample"] for f in final] 四、数据配比 class DataMixer: def create_mix(self, datasets): """创建数据混合""" # 2026年经验配比 mix = { "general_qa": 0.30, # 通用问答 "coding": 0.20, # 编程 "reasoning": 0.15, # 推理 "math": 0.10, # 数学 "creative_writing": 0.10, # 创意写作 "safety": 0.05, # 安全 "multi_turn": 0.05, # 多轮对话 "tool_use": 0.05, # 工具使用 } total = sum(v for v in mix.values()) assert abs(total - 1.0) < 0.01 mixed = [] for category, ratio in mix.items(): n = int(total_samples * ratio) sampled = self.sample_from(datasets[category], n) mixed.extend(sampled) random.shuffle(mixed) return mixed 五、训练 from trl import SFTTrainer, SFTConfig config = SFTConfig( output_dir="./sft-output", num_train_epochs=3, per_device_train_batch_size=8, gradient_accumulation_steps=4, learning_rate=2e-5, warmup_ratio=0.03, lr_scheduler_type="cosine", max_seq_length=2048, bf16=True, gradient_checkpointing=True, save_strategy="epoch", evaluation_strategy="epoch", ) trainer = SFTTrainer( model=base_model, args=config, train_dataset=train_data, eval_dataset=eval_data, tokenizer=tokenizer, ) trainer.train() 六、评估 async def evaluate_sft(model, eval_set): """评估SFT模型""" metrics = {} # 1. 自动评估 metrics["loss"] = model.evaluate(eval_set) # 2. 基准测试 benchmarks = ["MMLU", "HumanEval", "GSM8K", "MT-Bench"] for bench in benchmarks: metrics[bench] = await run_benchmark(model, bench) # 3. 人工评估 samples = generate_samples(model, n=100) metrics["human_score"] = await human_eval(samples) return metrics 七、常见陷阱 数据太多但质量低：10万高质量样本 > 100万低质量样本 格式不一致：确保所有数据使用统一的对话格式 过拟合：3轮通常足够，超过5轮容易过拟合 灾难性遗忘：混入通用数据防止遗忘基础能力 结语 指令微调是"数据为王"的领域。2026年的经验反复证明：花80%的时间在数据构建和质量控制上，20%在训练调参上，才能得到最好的效果。 ...

大模型训练四阶段：从原始语料到对齐智能 大语言模型的训练是一个多阶段的复杂过程。从原始互联网文本到能与人类对齐的智能助手，需要经过预训练、监督微调、人类反馈强化学习（RLHF）和直接偏好优化（DPO）四个阶段。本文将全面解析这条训练链路。 一、训练流程总览 ┌─────────────────────────────────────────────────────────────┐ │ 大模型训练全流程 │ ├─────────────────────────────────────────────────────────────┤ │ │ │ 阶段1: 预训练 (Pre-training) │ │ 输入: 万亿 Token 原始语料 │ │ 目标: Next Token Prediction │ │ 产出: 基座模型 (Base Model) │ │ │ │ 阶段2: 监督微调 (SFT) │ │ 输入: 高质量指令-回复对 (10K-1M) │ │ 目标: 模仿专家回复 │ │ 产出: 指令模型 (Instruct Model) │ │ │ │ 阶段3: 人类反馈强化学习 (RLHF) │ │ 输入: 人类偏好数据 (A > B 对比) │ │ 目标: 最大化人类偏好奖励 │ │ 产出: 对齐模型 (Aligned Model) │ │ │ │ 阶段4: 直接偏好优化 (DPO) │ │ 输入: 同 RLHF 的偏好数据 │ │ 目标: 直接优化策略, 无需奖励模型 │ │ 产出: 进一步优化的对齐模型 │ │ │ └─────────────────────────────────────────────────────────────┘ 二、预训练（Pre-training） 2.1 训练目标 预训练使用Next Token Prediction（NTP）作为训练目标： ...

Bad Data 的杀伤力 一句流行的说法：“Garbage In, Garbage Out”。在 SFT 微调中，这个效应被放大——1000 条高质量数据的效果远好于 10000 条低质量数据。低质量数据不仅浪费训练资源，还会主动降低模型能力。 实验数据 数据质量 数据量 模型准确率 关键问题 高质量 1K 82.3% 无 混合质量 10K 78.5% 偶尔幻觉 低质量 10K 65.2% 频繁幻觉、格式混乱 低质量 50K 61.8% 灾难性退化 结论：低质量数据越多，效果越差。50K 低质量数据比 1K 高质量数据差 20 个百分点。 1. Bad Data 的七大类型 class BadDataType: """SFT 数据中的七种常见质量问题""" # 类型1：格式不一致 FORMAT_INCONSISTENT = { "description": "回复格式不统一，有的用Markdown，有的用纯文本", "example": {"user": "解释RAG", "assistant": "RAG是检索增强生成"}, # 缺少结构化格式 "fix": "统一为指定格式（如Markdown），用LLM重新格式化" } # 类型2：回复过短/过长 LENGTH_EXTREME = { "description": "回复要么一句话敷衍，要么冗长重复", "example_short": {"user": "解释量子计算", "assistant": "量子计算用量子比特"}, "example_long": {"user": "解释量子计算", "assistant": "量子计算是一种...(5000字废话)"}, "fix": "过滤极端长度，保留200-2000字符范围" } # 类型3：事实错误 FACTUAL_ERROR = { "description": "回复中包含事实性错误", "example": {"user": "地球到月球多远", "assistant": "约38万公里"}, # 实际约38.4万公里 "fix": "用可信来源验证，或用强模型交叉检查" } # 类型4：答非所问 IRRELEVANT = { "description": "回复与问题不相关", "example": {"user": "如何优化SQL", "assistant": "SQL是结构化查询语言..."}, "fix": "计算query-response相关性，过滤低相关样本" } # 类型5：模板化回复 TEMPLATE_RESPONSE = { "description": "所有回复都是模板化的套话", "example": "作为AI语言模型，我不能...", "fix": "过滤包含常见AI模板用语的样本" } # 类型6：有害内容 HARMFUL = { "description": "包含偏见、歧视或有害建议", "fix": "安全过滤器 + 人工审核" } # 类型7：重复数据 DUPLICATE = { "description": "相同或高度相似的样本重复出现", "fix": "去重（精确去重 + 语义去重）" } 2. 数据质量评估框架 class SFTDataQualityAssessor: def __init__(self, strong_model): self.strong_model = strong_model # 用强模型做评估 self.dimensions = [ "accuracy", # 准确性 "relevance", # 相关性 "completeness", # 完整性 "clarity", # 清晰度 "safety", # 安全性 "format", # 格式规范性 ] def assess_sample(self, sample: dict) -> dict: prompt = f""" 请评估以下SFT训练样本的质量。 用户问题：{sample['messages'][-2]['content']} 助手回复：{sample['messages'][-1]['content']} 请从以下维度评分（1-5分）： 1. 准确性：回复中的信息是否准确？ 2. 相关性：回复是否直接回答了用户问题？ 3. 完整性：回复是否充分回答了问题？ 4. 清晰度：回复是否表达清晰、结构合理？ 5. 安全性：回复是否安全无害？ 6. 格式：回复格式是否规范统一？ 同时检查： - 是否有事实错误 - 是否有模板化语言 - 是否有害内容 - 回复长度是否合适 输出 JSON： {{ "scores": {{"accuracy": 1-5, "relevance": 1-5, "completeness": 1-5, "clarity": 1-5, "safety": 1-5, "format": 1-5}}, "overall_score": 1.0-5.0, "issues": ["问题1", "问题2"], "recommendation": "keep" / "fix" / "discard" }} """ result = self.strong_model.generate(prompt, response_format="json") return result def assess_dataset(self, dataset: list) -> dict: results = [] for sample in dataset: quality = self.assess_sample(sample) results.append(quality) return { "total_samples": len(dataset), "avg_overall": np.mean([r["overall_score"] for r in results]), "quality_distribution": self._distribution(results), "keep_count": sum(1 for r in results if r["recommendation"] == "keep"), "fix_count": sum(1 for r in results if r["recommendation"] == "fix"), "discard_count": sum(1 for r in results if r["recommendation"] == "discard"), "common_issues": self._aggregate_issues(results), } 3. 自动化数据清洗 class SFTDataCleaner: def __init__(self): self.steps = [ self.deduplicate, self.filter_length, self.filter_templates, self.filter_safety, self.check_relevance, self.fix_format, ] def clean(self, data: list) -> list: original_count = len(data) for step in self.steps: before = len(data) data = step(data) print(f"{step.__name__}: {before} → {len(data)} (removed {before - len(data)})") print(f"\n总计: {original_count} → {len(data)} (保留率: {len(data)/original_count:.1%})") return data def deduplicate(self, data: list): """三层去重""" # 1. 精确去重 seen = set() deduped = [] for sample in data: key = hash(json.dumps(sample, sort_keys=True)) if key not in seen: seen.add(key) deduped.append(sample) # 2. 问题去重（相同问题不同回复，保留最好的） question_map = {} for sample in deduped: q = sample["messages"][-2]["content"].strip() if q not in question_map: question_map[q] = sample else: # 保留回复更长的（通常更详细） old_resp = question_map[q]["messages"][-1]["content"] new_resp = sample["messages"][-1]["content"] if len(new_resp) > len(old_resp): question_map[q] = sample deduped = list(question_map.values()) # 3. 语义去重（相似问题） embeddings = self._compute_question_embeddings(deduped) clusters = self._cluster_similar(embeddings, threshold=0.95) deduped = [deduped[c[0]] for c in clusters] # 每簇保留一个 return deduped def filter_length(self, data: list): """过滤极端长度""" filtered = [] for sample in data: response = sample["messages"][-1]["content"] # 回复太短 if len(response) < 50: continue # 回复太长 if len(response) > 8000: continue # 问题太短（无法构成有效训练） question = sample["messages"][-2]["content"] if len(question) < 5: continue filtered.append(sample) return filtered def filter_templates(self, data: list): """过滤模板化回复""" TEMPLATE_PATTERNS = [ r"作为一个AI.*?我不能", r"作为AI语言模型", r"我是.*?AI.*?助手", r"很抱歉.*?无法", r"对不起.*?不能", r"我理解您的.*?但是", ] filtered = [] for sample in data: response = sample["messages"][-1]["content"] is_template = any( re.search(pattern, response, re.IGNORECASE) for pattern in TEMPLATE_PATTERNS ) if not is_template: filtered.append(sample) return filtered def check_relevance(self, data: list): """检查问题-回复相关性""" filtered = [] for sample in data: question = sample["messages"][-2]["content"] response = sample["messages"][-1]["content"] # 计算语义相似度 q_emb = self.embedder.encode(question) r_emb = self.embedder.encode(response) similarity = cosine_similarity(q_emb, r_emb) if similarity > 0.3: # 最低相关性阈值 filtered.append(sample) return filtered def fix_format(self, data: list): """统一格式""" for sample in data: response = sample["messages"][-1]["content"] # 统一使用 Markdown 格式 response = self._normalize_markdown(response) # 确保以句号或换行结尾 if not response.endswith(('.', '。', '!', '！', '?', '？', '\n')): response += '。' sample["messages"][-1]["content"] = response return data 4. 数据质量与训练效果的关系 实验设计 控制变量：基础模型 Qwen2.5-7B，训练参数相同，只变化数据质量。 ...

对齐技术全景 预训练模型 (Base) ↓ SFT (Supervised Fine-Tuning) ← 监督微调，学习指令遵循 ↓ DPO / RLHF ← 偏好对齐，学习人类价值观 ↓ 对齐模型 (Chat/Instruct) 阶段 目标 数据 方法 SFT 学会跟随指令 (指令, 回答) 对 监督学习 RLHF 对齐人类偏好 (prompt, chosen, rejected) 强化学习 DPO 对齐人类偏好 同上 直接优化，无需 RL SFT：监督微调 数据格式 Alpaca 格式： { "instruction": "将以下句子翻译为英文", "input": "今天天气真好", "output": "The weather is really nice today." } ShareGPT 格式（多轮对话）： { "conversations": [ {"from": "human", "value": "你好"}, {"from": "gpt", "value": "你好！有什么可以帮你的？"}, {"from": "human", "value": "解释一下量子计算"}, {"from": "gpt", "value": "量子计算利用量子力学原理..."} ] } 数据构造 import json import random class SFTDataBuilder: def __init__(self, llm_call): self.llm = llm_call def generate_alpaca_style(self, seed_topics: list[str], n_samples: int = 1000): """用 LLM 自动生成 SFT 数据""" dataset = [] for topic in seed_topics: prompt = f"""生成 10 条多样化的指令-回答对。 主题：{topic} 要求： 1. 指令类型多样（问答、翻译、摘要、代码、推理等） 2. 难度从简单到复杂 3. 回答准确、自然 输出 JSON 数组格式： [{{"instruction": "", "input": "", "output": ""}}]""" result = self.llm(prompt) try: items = json.loads(result) dataset.extend(items) except json.JSONDecodeError: continue # 去重 + 打乱 seen = set() unique = [] for item in dataset: key = item["instruction"] + item.get("input", "") if key not in seen: seen.add(key) unique.append(item) random.shuffle(unique) return unique[:n_samples] def to_sharegpt(self, alpaca_data: list[dict]) -> list[dict]: """Alpaca → ShareGPT 格式转换""" converted = [] for item in alpaca_data: instruction = item["instruction"] if item.get("input"): instruction = f"{instruction}\n\n输入: {item['input']}" converted.append({ "conversations": [ {"from": "human", "value": instruction}, {"from": "gpt", "value": item["output"]}, ] }) return converted 训练配置 from transformers import ( AutoModelForCausalLM, AutoTokenizer, TrainingArguments, Trainer, ) from datasets import Dataset from peft import LoraConfig, get_peft_model, TaskType import torch class SFTTrainer: def __init__(self, model_path: str, use_lora: bool = True): self.tokenizer = AutoTokenizer.from_pretrained(model_path) self.tokenizer.pad_token = self.tokenizer.eos_token self.model = AutoModelForCausalLM.from_pretrained( model_path, torch_dtype=torch.bfloat16, device_map="auto", ) if use_lora: self._setup_lora() def _setup_lora(self): """配置 LoRA：只训练少量参数""" lora_config = LoraConfig( task_type=TaskType.CAUSAL_LM, r=16, # LoRA rank lora_alpha=32, # 缩放因子 lora_dropout=0.05, target_modules=[ "q_proj", "k_proj", "v_proj", "o_proj", "gate_proj", "up_proj", "down_proj", ], ) self.model = get_peft_model(self.model, lora_config) self.model.print_trainable_parameters() # 输出: trainable params: 13M || all params: 7B || trainable%: 0.19% def format_prompt(self, instruction: str, input_text: str = "", output: str = "") -> str: """格式化为 ChatML 模板""" user_msg = instruction if input_text: user_msg += f"\n\n{input_text}" prompt = f"<|im_start|>user\n{user_msg}<|im_end|>\n" if output: prompt += f"<|im_start|>assistant\n{output}<|im_end|>" return prompt def train(self, data: list[dict], learning_rate: float = 2e-5, num_epochs: int = 3, batch_size: int = 4, max_length: int = 1024): # 格式化 texts = [ self.format_prompt( d["instruction"], d.get("input", ""), d["output"] ) for d in data ] dataset = Dataset.from_dict({"text": texts}) def tokenize(examples): tokenized = self.tokenizer( examples["text"], truncation=True, max_length=max_length, padding="max_length", ) tokenized["labels"] = tokenized["input_ids"].copy() return tokenized dataset = dataset.map(tokenize, batched=True) training_args = TrainingArguments( output_dir="./sft_output", num_train_epochs=num_epochs, per_device_train_batch_size=batch_size, gradient_accumulation_steps=4, learning_rate=learning_rate, warmup_ratio=0.03, lr_scheduler_type="cosine", logging_steps=10, save_strategy="epoch", bf16=True, optim="adamw_torch", ) trainer = Trainer( model=self.model, args=training_args, train_dataset=dataset, ) trainer.train() LoRA 超参数选择 参数 推荐值 说明 r (rank) 8-64 越大能力越强，但过拟合风险 lora_alpha 2×r 经验值，平衡缩放 target_modules all linear 全覆盖效果最好 dropout 0.05 防过拟合 learning_rate 1e-4 ~ 5e-5 LoRA 可用较大学习率 DPO：直接偏好优化 DPO (Rafailov et al., 2023) 是 RLHF 的简化替代——无需训练奖励模型，无需强化学习： ...