第 7 章：评估体系设计

┌─────────────────┬──────────────┬─────────────┬──────────────┐
│     基准名称     │   任务类型    │  数据规模   │    特点      │
├─────────────────┼──────────────┼─────────────┼──────────────┤
│ MMBench         │ 多选题       │ 3000题      │ 循环评估     │
│ SEED-Bench      │ 多选题       │ 19K题       │ 多维度覆盖   │
│ MME             │ 是/否判断    │ 14个子任务  │ 感知+认知    │
│ MMMU            │ 多选题       │ 11.5K题     │ 学科知识     │
│ MathVista       │ 数学推理     │ 6K题        │ 数学图表     │
└─────────────────┴──────────────┴─────────────┴──────────────┘

评估维度矩阵：
          ┌────────────────────────────────┐
          │         应用场景需求            │
          ├────────────────────────────────┤
          │  对话 │ OCR │ 推理 │ 创作    │
┌─────────┼───────┼─────┼──────┼──────────┤
│基础能力 │  ✓    │     │      │          │ → MMBench, SEED
│专业知识 │       │     │  ✓   │          │ → MMMU, MathVista  
│文字识别 │       │  ✓  │      │          │ → TextVQA, OCRBench
│内容生成 │       │     │      │    ✓     │ → COCO Caption
└─────────┴───────┴─────┴──────┴──────────┘

# 示例：检测训练数据与测试集的重叠
def detect_data_leakage(train_data, test_data):
    # 1. 图像级别检测（感知哈希）
    train_hashes = compute_perceptual_hashes(train_data.images)
    test_hashes = compute_perceptual_hashes(test_data.images)
    image_overlap = len(train_hashes & test_hashes) / len(test_hashes)

    # 2. 文本级别检测（n-gram 重叠）
    train_ngrams = extract_ngrams(train_data.texts, n=5)
    test_ngrams = extract_ngrams(test_data.texts, n=5)
    text_overlap = jaccard_similarity(train_ngrams, test_ngrams)

    # 3. 语义级别检测（embedding 相似度）
    semantic_sim = compute_semantic_similarity(train_data, test_data)

    return {
        'image_overlap': image_overlap,
        'text_overlap': text_overlap,
        'semantic_similarity': semantic_sim
    }

问题示例：
输入图像：[一只棕色的狗在草地上奔跑]
模型输出："一只金毛犬在绿色草坪上跑步"
参考答案："一只棕色的狗在草地上奔跑"

BLEU-4: 0.31 （低分，但语义正确）
人类评分：4.5/5 （高分，认为描述准确）

→ 指标与人类判断严重不一致

# GPT-4V 评估框架示例
class ModelBasedEvaluator:
    def __init__(self, judge_model="gpt-4-vision"):
        self.judge = load_model(judge_model)

    def evaluate(self, image, question, model_answer, reference=None):
        prompt = f"""
        请评估模型回答的质量（1-5分）：

        评估维度：

        1. 事实准确性：回答是否与图像内容一致
        2. 完整性：是否充分回答了问题
        3. 相关性：回答是否切题
        4. 清晰度：表达是否清楚易懂

        图像：[图像]
        问题：{question}
        模型回答：{model_answer}
        {"参考答案：" + reference if reference else ""}

        请给出：

        - 总分（1-5）
        - 各维度得分
        - 评价理由
        """

        return self.judge.generate(prompt, image)

# CHAIR (Caption Hallucination Assessment with Image Relevance)
def calculate_chair(generated_caption, image_objects):
    """
    计算描述中的幻觉率
    """
    mentioned_objects = extract_objects(generated_caption)

    # 句子级幻觉率
    hallucinated_sentences = 0
    total_sentences = len(generated_caption.split('.'))

    for sentence in generated_caption.split('.'):
        sentence_objects = extract_objects(sentence)
        if any(obj not in image_objects for obj in sentence_objects):
            hallucinated_sentences += 1

    chairs = hallucinated_sentences / total_sentences

    # 物体级幻觉率
    hallucinated_objects = len([obj for obj in mentioned_objects 
                               if obj not in image_objects])
    chairi = hallucinated_objects / len(mentioned_objects)

    return {'CHAIRs': chairs, 'CHAIRi': chairi}

def evaluate_spatial_understanding(prediction, ground_truth):
    """
    评估模型的空间关系理解能力
    """
    spatial_relations = ['左', '右', '上', '下', '前', '后', '内', '外']

    correct_relations = 0
    total_relations = 0

    for relation in spatial_relations:
        if relation in ground_truth:
            total_relations += 1
            if check_spatial_relation(prediction, ground_truth, relation):
                correct_relations += 1

    return correct_relations / total_relations if total_relations > 0 else 0

def instruction_following_score(instruction, response):
    """
    评估模型对指令的遵循程度
    """
    requirements = parse_requirements(instruction)

    scores = {
        'format_compliance': check_format(response, requirements.format),
        'length_compliance': check_length(response, requirements.length),
        'content_coverage': check_content(response, requirements.topics),
        'constraint_satisfaction': check_constraints(response, requirements.constraints)
    }

    return sum(scores.values()) / len(scores)

评估维度体系：
                    ┌─────────────┐
                    │  VLM 评估   │
                    └──────┬──────┘
                           │
        ┌──────────────────┼──────────────────┐
        │                  │                  │
   ┌────▼────┐      ┌─────▼─────┐     ┌─────▼─────┐
   │基础能力 │      │ 高级能力  │     │ 安全对齐  │
   └────┬────┘      └─────┬─────┘     └─────┬─────┘
        │                  │                  │

   - 物体识别         - 推理能力         - 有害内容过滤
   - 属性理解         - 创造性           - 偏见检测
   - 关系理解         - 知识运用         - 隐私保护
   - 计数能力         - 多轮对话         - 事实性

❌ 模糊指令：
"评估这个回答的质量"

✅ 明确指令：
"根据以下标准评估回答质量：

1. 事实准确性（0-2分）：描述是否与图像内容一致
2. 完整性（0-2分）：是否包含所有重要信息
3. 流畅性（0-1分）：语言是否通顺自然"

# 设计可量化的评估标准
evaluation_rubric = {
    "事实准确性": {
        0: "包含明显错误信息",
        1: "基本正确但有小错误",
        2: "完全准确"
    },
    "相关性": {
        0: "答非所问",
        1: "部分相关",
        2: "高度相关"
    }
}

# VLM 输出评估标注指南

## 1. 评估维度定义

### 1.1 事实准确性（Factual Accuracy）
**定义**：模型输出与图像内容的一致程度

**评分标准**：

- **优秀(3分)**：所有描述完全准确，无任何事实错误
- **良好(2分)**：主要信息正确，存在minor细节错误
- **一般(1分)**：部分信息正确，但有明显错误
- **差(0分)**：存在严重事实错误或幻觉

**示例**：
图像：[一只橙色的猫坐在蓝色沙发上]

- 优秀："一只橙色的猫在蓝色沙发上"
- 良好："一只猫在沙发上"（缺少颜色信息）
- 一般："一只狗在沙发上"（物体识别错误）
- 差："多只猫在地板上"（完全错误）

### 1.2 完整性（Completeness）
[详细定义和示例...]

## 2. 标注流程

1. **初步浏览**：快速查看图像，理解场景
2. **仔细对比**：逐句对比模型输出与图像
3. **评分记录**：按维度给分并记录理由
4. **一致性检查**：确保评分标准一致

## 3. 常见问题处理

Q: 如果模型使用同义词怎么办？
A: 同义词不扣分（如"汽车"vs"轿车"）

Q: 如何处理主观描述？
A: 只要合理即可（如"美丽的"风景）

# 计算标注者间一致性
def calculate_inter_rater_agreement(annotations):
    """
    计算 Fleiss' Kappa 系数
    """
    n_items = len(annotations[0])  # 评估项目数
    n_raters = len(annotations)      # 标注者数量
    n_categories = 5                 # 评分等级数（如1-5分）

    # 构建评分矩阵
    rating_matrix = np.zeros((n_items, n_categories))

    for item_idx in range(n_items):
        for rater_idx in range(n_raters):
            rating = annotations[rater_idx][item_idx]
            rating_matrix[item_idx, rating-1] += 1

    # 计算 Kappa
    kappa = fleiss_kappa(rating_matrix)

    # 解释 Kappa 值
    if kappa < 0.2:
        agreement = "微弱"
    elif kappa < 0.4:
        agreement = "一般"
    elif kappa < 0.6:
        agreement = "中等"
    elif kappa < 0.8:
        agreement = "较强"
    else:
        agreement = "极强"

    return kappa, agreement

# 综合分析框架
class EvaluationAnalyzer:
    def __init__(self, annotations):
        self.annotations = annotations

    def basic_statistics(self):
        """基础统计量"""
        scores = np.array(self.annotations)
        return {
            'mean': np.mean(scores, axis=0),
            'std': np.std(scores, axis=0),
            'median': np.median(scores, axis=0),
            'quantiles': np.percentile(scores, [25, 50, 75], axis=0)
        }

    def dimension_correlation(self):
        """维度间相关性分析"""
        # 分析不同评估维度之间的相关性
        dimensions = ['accuracy', 'completeness', 'fluency']
        corr_matrix = np.corrcoef(self.annotations[dimensions].T)
        return corr_matrix

    def error_analysis(self):
        """错误模式分析"""
        error_patterns = {
            'hallucination': 0,
            'missing_info': 0,
            'wrong_attribute': 0,
            'spatial_error': 0
        }
        # 分析常见错误类型
        return error_patterns

    def significance_test(self, model_a_scores, model_b_scores):
        """显著性检验"""
        from scipy import stats

        # 配对t检验
        t_stat, p_value = stats.ttest_rel(model_a_scores, model_b_scores)

        # Bootstrap 置信区间
        diff = model_a_scores - model_b_scores
        bootstrap_means = []
        for _ in range(1000):
            sample = np.random.choice(diff, size=len(diff), replace=True)
            bootstrap_means.append(np.mean(sample))

        ci_lower = np.percentile(bootstrap_means, 2.5)
        ci_upper = np.percentile(bootstrap_means, 97.5)

        return {
            't_statistic': t_stat,
            'p_value': p_value,
            'confidence_interval': (ci_lower, ci_upper),
            'significant': p_value < 0.05
        }

# VLM A/B 测试框架
class VLMABTestFramework:
    def __init__(self, config):
        self.config = config
        self.model_a = load_model(config.model_a_path)
        self.model_b = load_model(config.model_b_path)
        self.metrics_collector = MetricsCollector()

    def assign_user_to_group(self, user_id):
        """用户分组策略"""
        # 使用哈希确保同一用户始终分到同一组
        hash_value = hashlib.md5(user_id.encode()).hexdigest()
        hash_int = int(hash_value[:8], 16)

        if hash_int % 100 < self.config.traffic_split:
            return 'model_b'
        return 'model_a'

    def serve_request(self, user_id, image, query):
        """处理用户请求"""
        group = self.assign_user_to_group(user_id)

        # 记录请求信息
        request_id = str(uuid.uuid4())
        self.log_request(request_id, user_id, group)

        # 生成响应
        if group == 'model_a':
            response = self.model_a.generate(image, query)
        else:
            response = self.model_b.generate(image, query)

        # 收集指标
        self.collect_metrics(request_id, group, response)

        return response

    def collect_metrics(self, request_id, group, response):
        """收集评估指标"""
        metrics = {
            'latency': response.latency,
            'token_count': len(response.tokens),
            'user_feedback': None,  # 异步收集
            'downstream_success': None  # 追踪下游任务成功率
        }
        self.metrics_collector.record(request_id, group, metrics)

流量分配方案：

1. 渐进式发布（Progressive Rollout）
   第1天：5% 流量
   第3天：10% 流量（如果指标正常）
   第7天：25% 流量
   第14天：50% 流量
   第21天：100% 流量（全量发布）

2. 分层实验（Stratified Testing）
   ┌──────────────┬───────────┬────────────┐
   │    用户群体   │  流量占比  │   优先级    │
   ├──────────────┼───────────┼────────────┤
   │  内部用户     │    100%   │     1      │
   │  Beta 用户    │     50%   │     2      │
   │  VIP 用户     │     10%   │     3      │
   │  普通用户     │      5%   │     4      │
   └──────────────┴───────────┴────────────┘

3. 地域分配（Geographic Split）
   - 先在延迟容忍度高的地区测试
   - 逐步扩展到核心地区

class ABTestMonitor:
    def __init__(self, config):
        self.config = config
        self.alert_thresholds = config.alert_thresholds

    def check_guardrail_metrics(self, metrics):
        """检查护栏指标"""
        alerts = []

        # 1. 性能护栏
        if metrics.p95_latency > self.alert_thresholds.max_latency:
            alerts.append(('CRITICAL', f'P95延迟超标: {metrics.p95_latency}ms'))

        # 2. 质量护栏
        if metrics.error_rate > self.alert_thresholds.max_error_rate:
            alerts.append(('CRITICAL', f'错误率过高: {metrics.error_rate:.2%}'))

        # 3. 用户体验护栏
        if metrics.user_complaints > self.alert_thresholds.max_complaints:
            alerts.append(('WARNING', f'用户投诉增加: {metrics.user_complaints}'))

        return alerts

    def statistical_significance(self, control_metrics, treatment_metrics):
        """统计显著性检验"""
        # 计算提升和置信区间
        lift = (treatment_metrics.mean - control_metrics.mean) / control_metrics.mean

        # 计算统计功效
        sample_size = len(treatment_metrics.data)
        power = self.calculate_statistical_power(
            sample_size, 
            lift, 
            control_metrics.std
        )

        # 判断是否达到显著性
        p_value = self.calculate_p_value(control_metrics, treatment_metrics)

        return {
            'lift': lift,
            'p_value': p_value,
            'power': power,
            'significant': p_value < 0.05 and power > 0.8,
            'confidence_interval': self.calculate_ci(control_metrics, treatment_metrics)
        }

    def early_stopping_decision(self, current_results):
        """早停决策"""
        # 1. 如果严重负向，立即停止
        if current_results.lift < -0.1 and current_results.significant:
            return 'STOP', '显著负向影响'

        # 2. 如果已经显著正向，可以提前结束
        if current_results.lift > 0.05 and current_results.significant:
            if current_results.sample_size > self.config.min_sample_size:
                return 'SUCCESS', '显著正向提升'

        # 3. 如果样本量足够但无显著差异
        if current_results.sample_size > self.config.max_sample_size:
            return 'STOP', '无显著差异'

        return 'CONTINUE', None

# 长期效果追踪系统
class LongTermTracking:
    def __init__(self):
        self.metrics_history = defaultdict(list)

    def track_metric_degradation(self, metric_name, current_value):
        """追踪指标退化"""
        history = self.metrics_history[metric_name]
        history.append({
            'timestamp': datetime.now(),
            'value': current_value
        })

        # 检测趋势
        if len(history) > 7:  # 至少一周数据
            recent = [h['value'] for h in history[-7:]]
            baseline = [h['value'] for h in history[-14:-7]]

            # Mann-Kendall 趋势检验
            trend = self.mann_kendall_test(recent)

            if trend == 'decreasing':
                self.alert(f'{metric_name} 出现下降趋势')

    def track_user_behavior_shift(self, user_queries):
        """追踪用户行为变化"""
        # 分析查询分布变化
        query_distribution = self.analyze_query_distribution(user_queries)

        # 检测分布漂移
        if self.detect_distribution_shift(query_distribution):
            self.trigger_retraining_alert()

    def generate_weekly_report(self):
        """生成周报"""
        report = {
            'performance_trends': self.analyze_performance_trends(),
            'user_satisfaction': self.analyze_user_feedback(),
            'error_patterns': self.analyze_error_patterns(),
            'recommendations': self.generate_recommendations()
        }
        return report

# CircularEval 实现
def circular_eval(model, question, image, options):
    """
    通过打乱选项顺序多次评估，消除位置偏见
    """
    n_options = len(options)
    votes = defaultdict(int)

    # 生成所有循环排列
    for shift in range(n_options):
        # 循环移动选项
        shifted_options = options[shift:] + options[:shift]
        option_map = {chr(65+i): shifted_options[i] for i in range(n_options)}

        # 构造prompt
        prompt = format_question(question, shifted_options)

        # 获取模型预测
        answer = model.predict(image, prompt)

        # 映射回原始选项
        if answer in option_map:
            original_option = options.index(option_map[answer])
            votes[original_option] += 1

    # 投票确定最终答案
    final_answer = max(votes, key=votes.get)
    confidence = votes[final_answer] / n_options

    return final_answer, confidence

能力分类体系：
├── 感知能力（Perception）
│   ├── 物体定位（Object Localization）
│   ├── 属性识别（Attribute Recognition）
│   ├── 场景理解（Scene Understanding）
│   └── 空间关系（Spatial Relationship）
│
├── 推理能力（Reasoning）
│   ├── 逻辑推理（Logical Reasoning）
│   ├── 数值计算（Numerical Calculation）
│   ├── 常识推理（Commonsense Reasoning）
│   └── 因果推断（Causal Inference）
│
└── 知识能力（Knowledge）
    ├── 学科知识（Subject Knowledge）
    ├── 社会常识（Social Convention）
    ├── 历史文化（Historical Culture）
    └── 名人地标（Celebrity & Landmark）

# MMBench 结果分析工具
class MMBenchAnalyzer:
    def __init__(self, results):
        self.results = results

    def capability_radar_chart(self):
        """生成能力雷达图"""
        capabilities = {
            'Object Localization': 0.85,
            'Attribute Recognition': 0.92,
            'Spatial Relationship': 0.76,
            'Logical Reasoning': 0.68,
            'Commonsense': 0.81,
            'Subject Knowledge': 0.73
        }

        # 生成雷达图数据
        angles = np.linspace(0, 2*np.pi, len(capabilities), endpoint=False)
        values = list(capabilities.values())

        return angles, values

    def error_case_analysis(self):
        """错误案例分析"""
        error_patterns = {
            'position_bias': [],      # 位置偏好错误
            'language_bias': [],      # 语言偏见错误
            'hallucination': [],      # 幻觉错误
            'reasoning_fail': [],     # 推理失败
            'knowledge_gap': []       # 知识缺失
        }

        for item in self.results:
            if not item['correct']:
                error_type = self.classify_error(item)
                error_patterns[error_type].append(item)

        return error_patterns

    def compare_with_baselines(self):
        """与基准模型对比"""
        baselines = {
            'GPT-4V': 0.776,
            'Gemini-Pro': 0.739,
            'Claude-3': 0.768,
            'Qwen-VL-Plus': 0.726
        }

        our_score = np.mean([r['score'] for r in self.results])

        comparison = {
            name: {
                'score': score,
                'delta': our_score - score,
                'relative': (our_score - score) / score * 100
            }
            for name, score in baselines.items()
        }

        return comparison

class POPEEvaluator:
    def __init__(self, object_detector):
        self.detector = object_detector

    def generate_pope_questions(self, image):
        """生成 POPE 评估问题"""
        # 检测图像中的真实物体
        real_objects = self.detector.detect(image)

        # 构造三种类型的问题
        questions = {
            'random': self.random_sampling(real_objects),
            'popular': self.popular_sampling(real_objects),
            'adversarial': self.adversarial_sampling(real_objects)
        }

        return questions

    def random_sampling(self, real_objects):
        """随机采样策略"""
        questions = []
        # 50% 真实物体
        for obj in random.sample(real_objects, len(real_objects)//2):
            questions.append((f"Is there a {obj} in the image?", "Yes"))

        # 50% 不存在的物体
        fake_objects = self.get_random_objects(exclude=real_objects)
        for obj in fake_objects[:len(real_objects)//2]:
            questions.append((f"Is there a {obj} in the image?", "No"))

        return questions

    def popular_sampling(self, real_objects):
        """频繁共现物体采样"""
        # 选择经常一起出现但实际不在图中的物体
        co_occurring = self.get_co_occurring_objects(real_objects)
        fake_objects = [obj for obj in co_occurring if obj not in real_objects]

        questions = []
        for obj in real_objects[:len(real_objects)//2]:
            questions.append((f"Is there a {obj} in the image?", "Yes"))
        for obj in fake_objects[:len(real_objects)//2]:
            questions.append((f"Is there a {obj} in the image?", "No"))

        return questions

    def evaluate_hallucination(self, model, questions):
        """评估幻觉率"""
        results = {
            'accuracy': 0,
            'yes_bias': 0,  # 倾向于回答"是"
            'hallucination_rate': 0
        }

        correct = 0
        yes_count = 0
        false_positive = 0

        for question, ground_truth in questions:
            prediction = model.answer(question)

            if prediction == ground_truth:
                correct += 1
            if prediction == "Yes":
                yes_count += 1
                if ground_truth == "No":
                    false_positive += 1

        results['accuracy'] = correct / len(questions)
        results['yes_bias'] = yes_count / len(questions)
        results['hallucination_rate'] = false_positive / len([q for q in questions if q[1] == "No"])

        return results

class CoTEvaluator:
    def __init__(self):
        self.reasoning_patterns = {
            'visual_grounding': r'首先.*图像.*看到',
            'step_by_step': r'第[一二三四五]步',
            'logical_connector': r'因此|所以|由于|因为',
            'evidence_based': r'根据.*可以.*判断'
        }

    def evaluate_reasoning_quality(self, cot_response):
        """评估推理链质量"""
        scores = {}

        # 1. 推理步骤完整性
        steps = self.extract_reasoning_steps(cot_response)
        scores['completeness'] = min(len(steps) / 3, 1.0)  # 期望至少3步

        # 2. 逻辑连贯性
        scores['coherence'] = self.check_logical_flow(steps)

        # 3. 视觉grounding
        scores['grounding'] = self.check_visual_grounding(cot_response)

        # 4. 最终答案一致性
        scores['consistency'] = self.check_answer_consistency(cot_response)

        return scores

    def check_visual_grounding(self, response):
        """检查推理是否基于视觉信息"""
        visual_references = [
            '图像', '图中', '看到', '显示', '出现',
            '左边', '右边', '上方', '下方', '中间',
            '颜色', '形状', '大小'
        ]

        reference_count = sum(1 for ref in visual_references if ref in response)
        return min(reference_count / 5, 1.0)  # 期望至少5次视觉引用

    def compare_cot_vs_direct(self, model, test_set):
        """对比 CoT 和直接回答的效果"""
        results = {
            'direct': {'accuracy': 0, 'confidence': []},
            'cot': {'accuracy': 0, 'confidence': [], 'reasoning_quality': []}
        }

        for item in test_set:
            # 直接回答
            direct_answer = model.answer_direct(item.image, item.question)
            results['direct']['accuracy'] += (direct_answer == item.ground_truth)

            # CoT 回答
            cot_response = model.answer_with_cot(item.image, item.question)
            cot_answer = self.extract_final_answer(cot_response)
            results['cot']['accuracy'] += (cot_answer == item.ground_truth)

            # 评估推理质量
            reasoning_scores = self.evaluate_reasoning_quality(cot_response)
            results['cot']['reasoning_quality'].append(reasoning_scores)

        # 计算平均值
        n = len(test_set)
        results['direct']['accuracy'] /= n
        results['cot']['accuracy'] /= n

        return results

class AdversarialEvaluator:
    def __init__(self):
        self.attack_types = [
            'typographic',  # 文字类攻击
            'compositional', # 组合性攻击  
            'logical',       # 逻辑陷阱
            'visual'         # 视觉对抗
        ]

    def generate_adversarial_examples(self, original_sample):
        """生成对抗样本"""
        adversarial_samples = []

        # 1. 打字错误攻击
        typo_sample = self.add_typos(original_sample)
        adversarial_samples.append(('typographic', typo_sample))

        # 2. 否定词攻击
        negation_sample = self.add_negation(original_sample)
        adversarial_samples.append(('negation', negation_sample))

        # 3. 组合关系攻击
        comp_sample = self.shuffle_relationships(original_sample)
        adversarial_samples.append(('compositional', comp_sample))

        return adversarial_samples

    def evaluate_robustness(self, model, test_set):
        """评估模型鲁棒性"""
        robustness_scores = defaultdict(list)

        for original in test_set:
            # 原始样本得分
            original_score = model.evaluate(original)

            # 生成对抗样本
            adversarial_samples = self.generate_adversarial_examples(original)

            for attack_type, adv_sample in adversarial_samples:
                adv_score = model.evaluate(adv_sample)

                # 计算性能下降
                degradation = (original_score - adv_score) / original_score
                robustness_scores[attack_type].append(1 - degradation)

        # 汇总结果
        summary = {
            attack: {
                'mean_robustness': np.mean(scores),
                'std': np.std(scores),
                'min': np.min(scores)
            }
            for attack, scores in robustness_scores.items()
        }

        return summary

class CrossLingualEvaluator:
    def __init__(self, languages=['en', 'zh', 'ja', 'fr', 'es']):
        self.languages = languages
        self.translators = {lang: load_translator(lang) for lang in languages}

    def evaluate_language_consistency(self, model, image, question_en):
        """评估跨语言一致性"""
        results = {}

        # 英文基准答案
        answer_en = model.generate(image, question_en, lang='en')

        for lang in self.languages[1:]:  # 跳过英文
            # 翻译问题
            question_translated = self.translators[lang].translate(question_en)

            # 获取目标语言答案
            answer_lang = model.generate(image, question_translated, lang=lang)

            # 翻译回英文进行比较
            answer_back = self.translators[lang].translate_back(answer_lang)

            # 计算语义相似度
            similarity = self.semantic_similarity(answer_en, answer_back)

            results[lang] = {
                'answer': answer_lang,
                'back_translation': answer_back,
                'similarity': similarity,
                'consistent': similarity > 0.85
            }

        return results

    def identify_language_specific_challenges(self):
        """识别特定语言的挑战"""
        challenges = {
            'zh': [
                '量词使用（一个、一只、一条）',
                '方位词差异（上下左右 vs 东南西北）',
                '颜色描述（深浅 vs dark/light）'
            ],
            'ja': [
                '敬语级别',
                '汉字/假名选择',
                '计数词系统'
            ],
            'ar': [
                '从右到左的空间描述',
                '双数形式',
                '性别一致性'
            ]
        }
        return challenges

Item1: [4, 3, 4]  # 三位标注者的评分
Item2: [5, 5, 4]
...

# ❌ 错误做法
if model.accuracy > 0.9:
    deploy_model()

# ✅ 正确做法
evaluation_criteria = {
    'accuracy': (0.9, 'min'),
    'latency_p99': (500, 'max'),  # ms
    'hallucination_rate': (0.05, 'max'),
    'fairness_gap': (0.1, 'max'),
    'user_satisfaction': (4.0, 'min')  # 1-5 scale
}

all_pass = all(
    check_criterion(model, metric, threshold, direction)
    for metric, (threshold, direction) in evaluation_criteria.items()
)

if all_pass:
    deploy_model()

# 数据泄露检测
def detect_leakage(train_data, test_data):
    # 1. 精确匹配检测
    train_hashes = {hash(img.tobytes()) for img in train_data.images}
    test_hashes = {hash(img.tobytes()) for img in test_data.images}
    exact_overlap = len(train_hashes & test_hashes)

    # 2. 近似匹配检测（使用感知哈希）
    train_phash = {imagehash.phash(img) for img in train_data.images}
    test_phash = {imagehash.phash(img) for img in test_data.images}

    near_duplicates = 0
    for test_h in test_phash:
        for train_h in train_phash:
            if test_h - train_h < 5:  # 汉明距离阈值
                near_duplicates += 1
                break

    print(f"精确重复: {exact_overlap}")
    print(f"近似重复: {near_duplicates}")
    print(f"泄露率: {(exact_overlap + near_duplicates) / len(test_data) * 100:.2f}%")

# Bootstrap 置信区间
def calculate_confidence_interval(scores, n_bootstrap=1000):
    bootstrap_means = []
    n = len(scores)

    for _ in range(n_bootstrap):
        sample = np.random.choice(scores, size=n, replace=True)
        bootstrap_means.append(np.mean(sample))

    ci_lower = np.percentile(bootstrap_means, 2.5)
    ci_upper = np.percentile(bootstrap_means, 97.5)

    return {
        'mean': np.mean(scores),
        'ci_95': (ci_lower, ci_upper),
        'std_error': np.std(bootstrap_means)
    }

# 报告格式
result = calculate_confidence_interval(model_scores)
print(f"准确率: {result['mean']:.3f} (95% CI: {result['ci_95'][0]:.3f}-{result['ci_95'][1]:.3f})")

# 设置合理的停止标准
class ABTestStoppingCriteria:
    def __init__(self):
        self.min_sample_size = 10000
        self.min_test_days = 7
        self.required_power = 0.8

    def should_stop(self, test_stats):
        # 检查多个条件
        conditions = {
            'sample_size': test_stats.n >= self.min_sample_size,
            'duration': test_stats.days >= self.min_test_days,
            'statistical_power': test_stats.power >= self.required_power,
            'significance': test_stats.p_value < 0.05
        }

        # 只有所有条件满足才能停止
        can_stop = all(conditions.values())

        return can_stop, conditions

class AnnotationQualityController:
    def __init__(self):
        self.gold_standards = []  # 黄金标准样本
        self.annotator_history = defaultdict(list)

    def insert_quality_checks(self, task_batch):
        """插入质量检查样本"""
        # 每 10 个任务插入 1 个黄金标准
        mixed_batch = []
        for i, task in enumerate(task_batch):
            mixed_batch.append(task)
            if (i + 1) % 10 == 0:
                gold = random.choice(self.gold_standards)
                mixed_batch.append(gold)
        return mixed_batch

    def monitor_annotator_quality(self, annotator_id, annotations):
        """监控标注者质量"""
        gold_performance = []

        for ann in annotations:
            if ann.is_gold_standard:
                score = self.calculate_agreement(ann, ann.gold_answer)
                gold_performance.append(score)
                self.annotator_history[annotator_id].append({
                    'timestamp': datetime.now(),
                    'score': score
                })

        # 检测质量下降
        if len(gold_performance) > 5:
            recent_quality = np.mean(gold_performance[-5:])
            if recent_quality < 0.8:
                self.alert(f"标注者 {annotator_id} 质量下降到 {recent_quality:.2f}")
                self.trigger_retraining(annotator_id)

def create_edge_case_tests():
    edge_cases = [
        {
            'name': '空图像',
            'image': np.ones((224, 224, 3)) * 255,
            'question': '描述这张图片',
            'expected_behavior': '合理处理，不崩溃'
        },
        {
            'name': '超长输入',
            'image': normal_image,
            'question': 'a' * 10000,
            'expected_behavior': '截断或拒绝，不OOM'
        },
        {
            'name': '特殊字符',
            'image': normal_image,
            'question': '���这是什么？🤔',
            'expected_behavior': '正确解析，不报错'
        },
        {
            'name': '极端宽高比',
            'image': np.ones((10, 1000, 3)),
            'question': '这是什么形状？',
            'expected_behavior': '正确处理或优雅拒绝'
        }
    ]
    return edge_cases