数据构建

引言 指令微调（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%在训练调参上，才能得到最好的效果。 ...