SFT 数据质量评估：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": "去重（精确去重 + 语义去重）"
    }

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),
        }

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

数据量 vs 数据质量对效果的影响：

准确率
  85% |  ★(1K, 高质量)
      |       ★(5K, 高质量)
  80% |              
      |                    ★(10K, 中质量)
  75% |  
      |                              ★(50K, 低质量)
  70% |
      |
  65% |
      +------+------+------+------→ 数据量
       1K    5K    10K   50K

class DataQualityMonitor:
    """持续监控训练数据质量"""
    
    def monitor_during_training(self, trainer):
        """训练过程中监控数据质量"""
        for batch in trainer.train_dataloader:
            quality_issues = self._check_batch(batch)
            
            if quality_issues["critical"]:
                logger.warning(f"Critical data quality issue: {quality_issues}")
                # 可以选择跳过这个 batch
                continue
            
            # 记录质量指标
            trainer.log({"data_quality": quality_issues["score"]})
    
    def post_training_audit(self, model, eval_data):
        """训练后审计：哪些训练数据导致了问题"""
        bad_outputs = []
        
        for sample in eval_data:
            output = model.generate(sample["prompt"])
            
            if self._is_low_quality(output):
                # 追溯：哪些训练数据最相似
                similar_training = self._find_similar_training(sample["prompt"])
                bad_outputs.append({
                    "eval_prompt": sample["prompt"],
                    "bad_output": output,
                    "suspected_training_data": similar_training[:3]
                })
        
        return bad_outputs

class HighQualityDataBuilder:
    """构建高质量 SFT 数据的流水线"""
    
    def build(self, seed_prompts: list, target_count: int = 5000):
        # 1. 用强模型生成回复
        raw_data = self._generate_with_strong_model(seed_prompts)
        
        # 2. 自动清洗
        cleaned = self.cleaner.clean(raw_data)
        
        # 3. 质量评估
        quality = self.assessor.assess_dataset(cleaned)
        
        # 4. 人工抽检 (5%)
        sample_size = int(len(cleaned) * 0.05)
        human_checked = random.sample(cleaned, sample_size)
        human_scores = self._human_review(human_checked)
        
        # 5. 根据人工抽检校准自动评估
        if human_scores["agreement_rate"] < 0.8:
            logger.warning("自动评估与人工评估一致性低，需调整评估标准")
        
        # 6. 最终筛选
        final_data = [d for d in cleaned if d["quality_score"] > 4.0]
        
        return final_data