微调数据管线工程：清洗、去重、质量评估全流程

原始数据源
    │
    ├──→ 格式标准化
    ├──→ 基础清洗（去噪、截断、编码修复）
    ├──→ 去重（精确去重 + 语义去重）
    ├──→ 质量评估（规则 + 模型打分）
    ├──→ 安全过滤（有害内容、PII脱敏）
    ├──→ 分布均衡（领域采样）
    └──→ 最终数据集

import json
from dataclasses import dataclass
from typing import List

@dataclass
class TrainingSample:
    """标准化训练样本结构"""
    instruction: str          # 用户指令
    input: str = ""           # 附加输入（可选）
    output: str = ""          # 期望输出
    metadata: dict = None     # 来源、质量分数等元信息

def normalize_to_alpaca(raw_samples: List[dict]) -> List[dict]:
    """将各种格式统一为Alpaca格式"""
    normalized = []
    for sample in raw_samples:
        # 处理ShareGPT格式
        if "conversations" in sample:
            convs = sample["conversations"]
            instruction = convs[0]["value"] if len(convs) > 0 else ""
            output = convs[1]["value"] if len(convs) > 1 else ""
            normalized.append({"instruction": instruction, "output": output})
        # 处理Alpaca格式
        elif "instruction" in sample:
            normalized.append(sample)
        # 处理OpenAI格式
        elif "messages" in sample:
            msgs = sample["messages"]
            instruction = msgs[0]["content"] if msgs[0]["role"] == "user" else ""
            output = next((m["content"] for m in msgs if m["role"] == "assistant"), "")
            normalized.append({"instruction": instruction, "output": output})
    
    return normalized

import re

class TextCleaner:
    def __init__(self):
        self.patterns = {
            # HTML标签
            "html": re.compile(r'<[^>]+>'),
            # 连续空白
            "whitespace": re.compile(r'\s{3,}'),
            # URL（可选保留）
            "url": re.compile(r'https?://\S+'),
            # 乱码字符
            "garbage": re.compile(r'[\x00-\x08\x0b\x0c\x0e-\x1f\x7f-\x9f]'),
            # 重复标点（超过3个）
            "repeat_punct": re.compile(r'([！。，,.!?？]){4,}'),
        }
    
    def clean(self, text: str) -> str:
        text = self.patterns["garbage"].sub('', text)
        text = self.patterns["html"].sub('', text)
        text = self.patterns["whitespace"].sub('\n', text)
        text = self.patterns["repeat_punct"].sub(r'\1\1\1', text)
        return text.strip()
    
    def is_valid(self, text: str, min_len=10, max_len=4096) -> bool:
        """基础有效性检查"""
        if not text or len(text) < min_len or len(text) > max_len:
            return False
        # 数字与字母占比过高（可能是日志/代码误入）
        alpha_ratio = sum(c.isalpha() for c in text) / len(text)
        if alpha_ratio < 0.3:
            return False
        return True

def fix_encoding(text: str) -> str:
    """修复常见的编码问题"""
    # 修复UTF-8被误解码为Latin-1的情况
    try:
        return text.encode('latin-1').decode('utf-8')
    except (UnicodeDecodeError, UnicodeEncodeError):
        pass
    return text

import hashlib

def exact_deduplication(samples: List[dict]) -> List[dict]:
    """基于内容哈希的精确去重"""
    seen = set()
    unique = []
    
    for sample in samples:
        # 对instruction+output做哈希
        content = sample["instruction"] + sample.get("output", "")
        content_hash = hashlib.md5(content.encode('utf-8')).hexdigest()
        
        if content_hash not in seen:
            seen.add(content_hash)
            unique.append(sample)
    
    print(f"精确去重: {len(samples)} → {len(unique)} (移除 {len(samples)-len(unique)} 条)")
    return unique

from sentence_transformers import SentenceTransformer
from sklearn.metrics.pairwise import cosine_similarity
import numpy as np

def semantic_deduplication(samples: List[dict], 
                           threshold=0.92,
                           batch_size=512) -> List[dict]:
    """基于向量相似度的语义去重"""
    model = SentenceTransformer("BAAI/bge-large-zh-v1.5")
    
    # 编码所有样本
    texts = [s["instruction"] + s.get("output", "") for s in samples]
    embeddings = model.encode(texts, batch_size=batch_size, 
                              normalize_embeddings=True, show_progress_bar=True)
    
    # 聚类去重：保留每个簇中质量最高的样本
    kept_indices = []
    removed = set()
    
    for i in range(len(samples)):
        if i in removed:
            continue
        kept_indices.append(i)
        # 批量计算与后续样本的相似度
        similarities = cosine_similarity(
            embeddings[i:i+1], 
            embeddings[i+1:]
        )[0]
        
        for j, sim in enumerate(similarities):
            if sim > threshold:
                removed.add(i + 1 + j)
    
    unique = [samples[i] for i in kept_indices]
    print(f"语义去重: {len(samples)} → {len(unique)} (移除 {len(samples)-len(unique)} 条)")
    return unique

class QualityScorer:
    def __init__(self):
        self.rules = {
            "length_balance": self._check_length_balance,
            "instruction_clarity": self._check_instruction_clarity,
            "output_completeness": self._check_output_completeness,
            "no_template_repetition": self._check_no_template,
        }
    
    def score(self, sample: dict) -> float:
        scores = [rule(sample) for rule in self.rules.values()]
        return sum(scores) / len(scores)
    
    def _check_length_balance(self, sample):
        """检查指令与输出的长度比"""
        inst_len = len(sample["instruction"])
        out_len = len(sample.get("output", ""))
        if out_len == 0:
            return 0.0
        ratio = inst_len / (inst_len + out_len)
        # 指令占比在10%-60%之间为合理
        if 0.1 <= ratio <= 0.6:
            return 1.0
        return 0.5
    
    def _check_instruction_clarity(self, sample):
        """检查指令是否清晰具体"""
        inst = sample["instruction"]
        # 过短的指令通常不够清晰
        if len(inst) < 5:
            return 0.2
        # 包含明确动词的指令更清晰
        verbs = ["解释", "分析", "写", "总结", "翻译", "列出", "比较", "设计"]
        if any(v in inst for v in verbs):
            return 1.0
        return 0.6
    
    def _check_output_completeness(self, sample):
        """检查输出是否完整"""
        output = sample.get("output", "")
        if len(output) < 20:
            return 0.3
        # 检查是否有截断标志
        if output.endswith(("...", "等等", "诸如此类")):
            return 0.5
        return 1.0
    
    def _check_no_template(self, sample):
        """检查是否包含模板化重复"""
        output = sample.get("output", "")
        # 检测连续重复的句子
        sentences = output.split("。")
        if len(sentences) > 2:
            unique_ratio = len(set(sentences)) / len(sentences)
            return min(1.0, unique_ratio + 0.2)
        return 1.0

from openai import OpenAI

def model_quality_score(sample: dict) -> float:
    """使用强模型对训练样本打分"""
    client = OpenAI()
    
    prompt = f"""请对以下训练样本的质量打分（0-10分），评估维度：
1. 指令是否清晰、具体
2. 回答是否准确、完整
3. 回答是否直接针对指令
4. 是否有事实错误或有害内容

指令：{sample['instruction']}
回答：{sample.get('output', '')}

只输出一个0到10的数字。"""

    response = client.chat.completions.create(
        model="gpt-4o-mini",
        messages=[{"role": "user", "content": prompt}],
        temperature=0,
    )
    
    score_text = response.choices[0].message.content.strip()
    try:
        return float(score_text) / 10.0
    except ValueError:
        return 0.5  # 解析失败给中等分

class SafetyFilter:
    def __init__(self):
        # PII正则模式
        self.pii_patterns = {
            "phone": re.compile(r'1[3-9]\d{9}'),
            "id_card": re.compile(r'\d{17}[\dXx]'),
            "email": re.compile(r'[\w.-]+@[\w.-]+\.\w+'),
            "bank_card": re.compile(r'\d{16,19}'),
        }
    
    def filter_pii(self, text: str) -> str:
        """脱敏个人信息"""
        text = self.pii_patterns["phone"].sub("[电话]", text)
        text = self.pii_patterns["id_card"].sub("[身份证]", text)
        text = self.pii_patterns["email"].sub("[邮箱]", text)
        text = self.pii_patterns["bank_card"].sub("[银行卡]", text)
        return text
    
    def is_safe(self, sample: dict) -> bool:
        """安全检查"""
        text = sample["instruction"] + sample.get("output", "")
        harmful_keywords = ["制造炸弹", "黑客攻击教程", "毒品合成"]
        return not any(kw in text for kw in harmful_keywords)

from collections import Counter

def report_dataset_stats(samples: List[dict]):
    """生成数据集统计报告"""
    print(f"总样本数: {len(samples)}")
    
    # 长度分布
    lengths = [len(s.get("output", "")) for s in samples]
    print(f"输出长度: 平均{np.mean(lengths):.0f}, 中位数{np.median(lengths):.0f}")
    
    # 质量分数分布
    scorer = QualityScorer()
    scores = [scorer.score(s) for s in samples]
    print(f"质量分数: 平均{np.mean(scores):.2f}, >0.7占比: {sum(s>0.7 for s in scores)/len(scores)*100:.1f}%")
    
    # 领域分布（基于简单关键词分类）
    domains = []
    for s in samples:
        text = s["instruction"]
        if any(k in text for k in ["代码", "编程", "函数"]):
            domains.append("编程")
        elif any(k in text for k in ["翻译", "英语", "中文"]):
            domains.append("翻译")
        elif any(k in text for k in ["数学", "计算", "方程"]):
            domains.append("数学")
        else:
            domains.append("通用")
    
    dist = Counter(domains)
    for domain, count in dist.most_common():
        print(f"  {domain}: {count} ({count/len(samples)*100:.1f}%)")