Anthropic众包训练Claude传闻：1000工程师如何重塑AI训练范式

from dataclasses import dataclass, field
from typing import List, Dict, Tuple
from collections import Counter
import re

@dataclass
class QualityScore:
    completeness: float    # 完整性：回答是否完整（0-1）
    consistency: float     # 一致性：与已有知识的符合度（0-1）
    professionalism: float # 专业性：技术深度（0-1）
    diversity: float       # 多样性：覆盖不同场景的程度（0-1）
    
    @property
    def overall(self) -> float:
        return (
            self.completeness * 0.3 +
            self.consistency * 0.3 +
            self.professionalism * 0.25 +
            self.diversity * 0.15
        )

@dataclass
class CrowdSourceItem:
    contributor_id: str
    prompt: str
    response: str
    category: str  # 分类：coding/math/writing 等


class CrowdDataQualityEvaluator:
    """众包训练数据质量评估器"""
    
    def __init__(self):
        self.reference_knowledge: Dict = {}  # 参考知识库
        self.completeness_threshold = 100    # 最小字数
        self.technical_terms: Dict[str, List] = {  # 各分类的关键词
            'coding': ['def', 'class', 'import', 'function', 'return'],
            'math': ['theorem', 'proof', 'equation', 'integral', 'derivative'],
        }
    
    def evaluate_item(self, item: CrowdSourceItem) -> QualityScore:
        return QualityScore(
            completeness=self._score_completeness(item),
            consistency=self._score_consistency(item),
            professionalism=self._score_professionalism(item),
            diversity=self._score_diversity(item)
        )
    
    def _score_completeness(self, item: CrowdSourceItem) -> float:
        """完整性评分：检查回答是否完整"""
        if len(item.response) < self.completeness_threshold:
            return max(0.0, len(item.response) / self.completeness_threshold)
        # 检查是否包含完整的代码块、解释和示例
        has_code = 'python' in item.response.lower() or 'def ' in item.response
        has_explanation = len(re.split(r'[.。!?！？]', item.response)) > 2
        return min(1.0, (0.5 if has_code else 0) + (0.5 if has_explanation else 0))
    
    def _score_consistency(self, item: CrowdSourceItem) -> float:
        """一致性评分：检查与参考知识的一致性"""
        # 简化实现：检查是否包含明显的错误关键词
        error_patterns = ['不知道', '无法回答', '不确定', 'I don\'t know']
        if any(p in item.response for p in error_patterns):
            return 0.0
        return 1.0  # 实际应用中需要与参考知识库进行语义比对
    
    def _score_professionalism(self, item: CrowdSourceItem) -> float:
        """专业性评分：检查技术深度"""
        category_terms = self.technical_terms.get(item.category, [])
        if not category_terms:
            return 0.5  # 未知分类
        
        term_count = sum(1 for t in category_terms if t.lower() in item.response.lower())
        return min(1.0, term_count / max(1, len(category_terms) * 0.5))
    
    def _score_diversity(self, item: CrowdSourceItem) -> float:
        """多样性评分：评估 prompt 的覆盖范围"""
        # 简化实现：检查 prompt 的唯一性
        words = set(item.prompt.lower().split())
        return min(1.0, len(words) / 20.0)  # 20+ 独特词汇 = 满分
    
    def batch_evaluate(self, items: List[CrowdSourceItem]) -> Dict:
        """批量评估并生成质量报告"""
        scores = [self.evaluate_item(item) for item in items]
        
        return {
            'total_items': len(items),
            'avg_completeness': sum(s.completeness for s in scores) / len(scores),
            'avg_consistency': sum(s.consistency for s in scores) / len(scores),
            'avg_professionalism': sum(s.professionalism for s in scores) / len(scores),
            'avg_diversity': sum(s.diversity for s in scores) / len(scores),
            'avg_overall': sum(s.overall for s in scores) / len(scores),
            'high_quality_count': sum(1 for s in scores if s.overall >= 0.7),
            'low_quality_count': sum(1 for s in scores if s.overall < 0.5),
        }

from typing import List, Dict
from collections import defaultdict
import hashlib


def compute_semantic_hash(text: str, max_words: int = 50) -> str:
    """计算文本的语义哈希（简化版）"""
    # 提取关键词并排序
    words = sorted(set(text.lower().split()))[:max_words]
    content = ' '.join(words)
    return hashlib.md5(content.encode()).hexdigest()[:8]


def detect_duplicates(items: List[Dict]) -> Dict:
    """检测众包数据中的重复项
    
    Args:
        items: [{'contributor_id': str, 'prompt': str, 'response': str}]
    
    Returns:
        去重统计报告
    """
    prompt_hashes = defaultdict(list)
    response_hashes = defaultdict(list)
    
    for idx, item in enumerate(items):
        p_hash = compute_semantic_hash(item['prompt'])
        r_hash = compute_semantic_hash(item['response'])
        prompt_hashes[p_hash].append(idx)
        response_hashes[r_hash].append(idx)
    
    # 统计重复情况
    duplicate_prompts = sum(1 for indices in prompt_hashes.values() if len(indices) > 1)
    duplicate_responses = sum(1 for indices in response_hashes.values() if len(indices) > 1)
    
    # 找出高频贡献者（可能刷单）
    contributor_counts = defaultdict(int)
    for item in items:
        contributor_counts[item['contributor_id']] += 1
    
    high_frequency = {cid: cnt for cid, cnt in contributor_counts.items() 
                      if cnt > len(items) / len(contributor_counts) * 3}
    
    return {
        'total_items': len(items),
        'unique_prompts': len(prompt_hashes),
        'duplicate_prompt_groups': duplicate_prompts,
        'duplicate_response_groups': duplicate_responses,
        'duplicate_rate': f'{duplicate_prompts / len(prompt_hashes) * 100:.1f}%',
        'high_frequency_contributors': high_frequency,
    }


# 使用示例
if __name__ == '__main__':
    sample_items = [
        {'contributor_id': 'eng_001', 'prompt': '如何实现快速排序', 'response': 'def quicksort...'},
        {'contributor_id': 'eng_002', 'prompt': '快排算法的 Python 实现', 'response': 'def quick_sort...'},
        {'contributor_id': 'eng_003', 'prompt': '写一个冒泡排序', 'response': 'def bubble_sort...'},
    ]
    report = detect_duplicates(sample_items)
    print(f"重复率: {report['duplicate_rate']}")
    print(f"高频贡献者: {report['high_frequency_contributors']}")