大模型训练数据治理：从清洗到质量评估全链路

# 合成数据生成的典型pipeline
synthetic_pipeline = {
    "seed_topics": "从高质量知识图谱采样种子主题",
    "generation": "使用强模型生成多角度、多深度的内容",
    "filtering": "使用reward model过滤低质量生成",
    "deduplication": "MinHash + LSH去重",
    "verification": "使用验证器检查事实准确性"
}

def extract_text(html: str) -> str:
    # 1. 使用trafilatura提取正文
    text = trafilatura.extract(html, include_links=False, include_tables=True)
    
    # 2. 去除模板化内容（导航栏、页脚等）
    boilerplate_ratio = compute_boilerplate_ratio(text, html)
    if boilerplate_ratio > 0.5:
        text = clean_boilerplate(text)
    
    # 3. 语言检测
    lang = fasttext.predict(text)
    if lang not in TARGET_LANGS:
        return None
    
    # 4. 编码修复
    text = ftfy.fix_text(text)
    
    return text

quality_prompt = """
请评估以下文本的质量，从1-5分打分：
1分：垃圾内容，无法理解
2分：质量很低，大量语法错误
3分：质量一般，可读但信息量少
4分：质量较好，信息丰富，表达清晰
5分：质量极高，专业、准确、深入

文本：{text}

评分（1-5）：
"""

# MinHash-LSH去重实现
from datasketch import MinHash, MinHashLSH

def deduplicate_documents(docs, threshold=0.8, num_perm=128):
    lsh = MinHashLSH(threshold=threshold, num_perm=num_perm)
    
    for doc_id, doc in enumerate(docs):
        mh = MinHash(num_perm=num_perm)
        # 使用3-gram作为shingle
        shingles = set(get_ngrams(doc, n=3))
        for s in shingles:
            mh.update(s.encode('utf-8'))
        
        # 检查相似文档
        similar = lsh.query(mh)
        if not similar:
            lsh.insert(str(doc_id), mh)
            yield doc  # 保留该文档

class DataQualityMetrics:
    def __init__(self):
        self.dimensions = {
            "diversity": "主题覆盖度，使用LDA或BERT聚类计算",
            "richness": "词汇丰富度，Type-Token Ratio",
            "coherence": "文本连贯性，使用连贯性模型评分",
            "factuality": "事实准确性，与知识库交叉验证",
            "freshness": "时效性，基于发布时间",
            "coverage": "领域覆盖度，按预定义领域分类"
        }
    
    def compute_overall_score(self, dataset):
        scores = {}
        for dim, method in self.dimensions.items():
            scores[dim] = evaluate(dataset, dim)
        
        # 加权综合评分
        weights = {
            "diversity": 0.25,
            "richness": 0.15,
            "coherence": 0.20,
            "factuality": 0.20,
            "freshness": 0.10,
            "coverage": 0.10
        }
        return sum(scores[k] * weights[k] for k in weights)

def detect_contamination(train_data, eval_data, ngram_size=13):
    """使用n-gram匹配检测benchmark污染"""
    eval_ngrams = set()
    for text in eval_data:
        for i in range(len(text) - ngram_size + 1):
            eval_ngrams.add(text[i:i+ngram_size].lower())
    
    contaminated = []
    for doc_id, text in enumerate(train_data):
        matches = 0
        for i in range(len(text) - ngram_size + 1):
            if text[i:i+ngram_size].lower() in eval_ngrams:
                matches += 1
        contamination_rate = matches / max(1, len(text) - ngram_size + 1)
        if contamination_rate > 0.01:  # 1%阈值
            contaminated.append((doc_id, contamination_rate))
    
    return contaminated