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，训练参数相同，只变化数据质量。
...
