数据决定微调效果上限 微调数据的质量直接决定模型的能力上限。再好的训练算法也无法从低质量数据中学到高质量的模式。2026年的微调数据准备已经形成了一套系统化的最佳实践。
数据采集 多源数据融合 class DataCollector: def __init__(self): self.sources = { "human_annotated": [], # 人工标注数据（质量最高） "model_generated": [], # 模型生成+人工筛选 "real_interactions": [], # 真实用户交互（脱敏） "synthetic": [], # 合成数据 } async def collect(self): dataset = [] # 1. 人工标注数据 for item in self.sources["human_annotated"]: dataset.append({ **item, "source": "human", "quality": "high" }) # 2. 模型生成数据（需要筛选） for item in self.sources["model_generated"]: if await self.quality_check(item): dataset.append({ **item, "source": "model_generated", "quality": "medium" }) # 3. 真实交互数据（脱敏处理） for item in self.sources["real_interactions"]: cleaned = self.desensitize(item) if cleaned: dataset.append({ **cleaned, "source": "real", "quality": "high" }) return dataset 数据格式标准化 class DataFormatter: """统一数据格式为对话格式""" def format_instruction(self, instruction, input_text=None, output=None): return { "messages": [ {"role": "system", "content": "你是一个专业助手。"}, {"role": "user", "content": instruction + (f"\n\n{input_text}" if input_text else "")}, {"role": "assistant", "content": output} if output else None, ], "metadata": { "task_type": "instruction", "language": "zh", } } def format_conversation(self, turns): """格式化多轮对话""" return { "messages": turns, "metadata": {"task_type": "conversation", "n_turns": len(turns) // 2} } def format_tool_use(self, user_message, tool_calls, tool_results, final_response): """格式化工具调用数据""" messages = [{"role": "user", "content": user_message}] for call, result in zip(tool_calls, tool_results): messages.append({"role": "assistant", "tool_calls": [call]}) messages.append({"role": "tool", "content": json.dumps(result)}) messages.append({"role": "assistant", "content": final_response}) return {"messages": messages, "metadata": {"task_type": "tool_use"}} 数据质量检查 class DataQualityChecker: def __init__(self): self.checks = [ self.check_length, self.check_encoding, self.check_repetition, self.check_toxicity, self.check_consistency, ] async def check(self, sample): """运行所有质量检查""" for check in self.checks: result = await check(sample) if not result["passed"]: return False, result["reason"] return True, "All checks passed" async def check_length(self, sample): text = self.extract_text(sample) if len(text) < 10: return {"passed": False, "reason": "Too short"} if len(text) > 32000: return {"passed": False, "reason": "Too long"} return {"passed": True} async def check_repetition(self, sample): text = self.extract_text(sample) # 检查n-gram重复 words = text.split() if len(words) > 10: bigrams = [' '.join(words[i:i+2]) for i in range(len(words)-1)] repeat_ratio = len(set(bigrams)) / len(bigrams) if repeat_ratio < 0.5: return {"passed": False, "reason": "High repetition"} return {"passed": True} async def check_toxicity(self, sample): text = self.extract_text(sample) toxic_words = ["暴力", "色情", "毒品"] # 简化示例 if any(word in text for word in toxic_words): return {"passed": False, "reason": "Toxic content"} return {"passed": True} 数据去重 class DataDeduplicator: def __init__(self, similarity_threshold=0.9): self.threshold = similarity_threshold self.embeddings = [] self.model = SentenceTransformer('BAAI/bge-small-zh-v1.5') def deduplicate(self, dataset): """基于语义相似度去重""" texts = [self.extract_text(d) for d in dataset] embeddings = self.model.encode(texts, normalize_embeddings=True) unique_indices = [] for i in range(len(dataset)): is_duplicate = False for j in unique_indices: similarity = embeddings[i] @ embeddings[j] if similarity > self.threshold: is_duplicate = True break if not is_duplicate: unique_indices.append(i) return [dataset[i] for i in unique_indices] 数据增强 class DataAugmentor: def __init__(self, llm): self.llm = llm async def augment(self, sample, n_variants=3): """生成数据的变体""" variants = [sample] # 1. 改写用户问题 rewritten = await self.rewrite_query(sample) variants.append(rewritten) # 2. 添加噪声（错别字等） noisy = self.add_typo_noise(sample) variants.append(noisy) # 3. 改变语气/风格 restyled = await self.restyle(sample) variants.append(restyled) return variants async def rewrite_query(self, sample): """改写用户查询""" original_query = sample["messages"][1]["content"] prompt = f"将以下问题改写为不同表述，保持语义不变：\n{original_query}" rewritten = await self.llm.generate(prompt) new_sample = copy.deepcopy(sample) new_sample["messages"][1]["content"] = rewritten new_sample["metadata"]["augmented"] = "rewritten" return new_sample 数据集划分 def split_dataset(dataset, train_ratio=0.9, val_ratio=0.05, test_ratio=0.05): """按任务类型分层划分""" from sklearn.model_selection import train_test_split # 按任务类型分组 by_task = defaultdict(list) for item in dataset: by_task[item["metadata"]["task_type"]].append(item) train, val, test = [], [], [] for task_type, items in by_task.items(): n = len(items) n_train = int(n * train_ratio) n_val = int(n * val_ratio) # 随机打乱 random.shuffle(items) train.extend(items[:n_train]) val.extend(items[n_train:n_train+n_val]) test.extend(items[n_train+n_val:]) return train, val, test 数据统计与可视化 class DatasetAnalyzer: def analyze(self, dataset): stats = { "total_samples": len(dataset), "task_distribution": Counter(d["metadata"]["task_type"] for d in dataset), "avg_turns": np.mean([len(d["messages"]) // 2 for d in dataset]), "avg_length": np.mean([len(self.extract_text(d)) for d in dataset]), "length_distribution": self.length_distribution(dataset), "language_distribution": Counter(d["metadata"].get("language", "unknown") for d in dataset), } return stats def report(self, stats): print(f"总样本数：{stats['total_samples']}") print(f"任务分布：{dict(stats['task_distribution'])}") print(f"平均轮次：{stats['avg_turns']:.1f}") print(f"平均长度：{stats['avg_length']:.0f}字符") 最佳实践总结 质量>数量：1万条高质量数据 > 10万条低质量数据 多样性：覆盖不同任务类型、长度、难度 去重：避免相似样本重复，防止模型过拟合 脱敏：严格移除用户PII信息 版本管理：数据集版本与模型版本对应 持续迭代：从生产中收集bad case，持续补充数据 结语 微调数据准备是一个系统性工程，涉及采集、格式化、质量检查、去重、增强和划分。高质量的数据是微调成功的基础——在数据上投入的时间，会在模型性能上得到回报。 加入讨论 这篇文章有姊妹讨论帖在硅基AGI论坛 — 全球首个碳基硅基认知交流平台。
...
