RAG系统如何评测:从MMLU基准到G-Eval提分实践

作者:袖梨 2026-09-14

RAG系统能够返回答案,并不意味着它已经具备稳定的生产效果。面对准确性、相关性、一致性和业务适配度等不同目标,仅跑一个公开基准往往无法说明问题。要得到可解释、可复现的评测结论,需要同时设计指标、测试集、裁判机制与排错流程,并让它们贴近真实请求分布。

RAG系统评测实战:MMLU到G-Eval的完整评测踩坑与提分

RAG 系统上线后,PM 问「效果怎么样」,我答不上来——没有能说服人的数字。

这篇讲清楚四件事:主流评测集怎么用哪些评测方法适合你LLM-as-Judge 怎么做、以及怎么建一个能重复用的领域评测集


目录

  • 评测的三个层次
  • 主流评测集详解
  • 评测方法选型:自动指标 vs LLM 当裁判
  • LLM-as-Judge 实战
  • 构建领域评测集
  • 生产级评测 Pipeline
  • 快速排错表
  • 配置检查清单

评测的三个层次

评测的三个层次

┌──────────────────────────────────────────────────────────────┐
│                   大模型评测三个层次                           │
├──────────────────────────────────────────────────────────────┤
│                                                              │
│  层次1:学术基准(Standard Benchmarks)                       │
│  ├─ MMLU, HumanEval, GSM8K, ARC, BIG-Bench                  │
│  ├─ 用途:与SOTA模型横向对比                                  │
│  └─ 局限:数据可能泄露,题目可能被刷爆                        │
│                                                              │
│  层次2:能力维度(Capability Assessment)                     │
│  ├─ 推理、知识、代码、对话、数学、安全等                      │
│  ├─ 用途:识别模型能力短板,指导优化方向                      │
│  └─ 局限:能力维度边界模糊,难以穷尽                          │
│                                                              │
│  层次3:场景评测(Scenario-based)                            │
│  ├─ 实际业务场景的测试集                                      │
│  ├─ 用途:验证模型是否满足业务需求                            │
│  └─ 局限:需要领域专家参与,成本高                            │
│                                                              │
└──────────────────────────────────────────────────────────────┘

评测常见误区

# evaluation_pitfalls.py

"""
大模型评测常见误区

误区1:只跑公开基准,不看实际效果
  危害:可能刷出高分但实际不好用(benchmark泄露/过拟合)
  解决:必须结合人工评估和场景测试

误区2:单一指标定英雄
  危害:MMLU高不代表代码好,代码好不代表对话好
  解决:多维度综合评估,雷达图展示

误区3:测试集和训练集混淆
  危害:数据泄露导致分数虚高
  解决:严格的数据隔离,定期更新测试集

误区4:忽视一致性(Consistency)
  危害:同一问题问两遍得到不同答案
  解决:多次采样评估一致性

误区5:只看结果,不看过程
  危害:无法定位问题根因
  解决:记录完整推理过程,分析错误类型
"""

PITFALLS = {
    "benchmark数据泄露": {
        "症状": "新模型在公开基准上分数突然飙升",
        "原因": "训练数据包含了测试集",
        "检测": "检查训练语料中是否有测试集内容",
        "解决": "使用私有hold-out测试集,或使用动态评测"
    },
    
    "过拟合评测集": {
        "症状": "benchmark分数高,但泛化差",
        "原因": "针对特定评测集刷题",
        "检测": "在不同变体的评测集上测试",
        "解决": "用综合评测,定期更新评测题"
    },
    
    "评估者偏见": {
        "症状": "不同评估者打分差异大",
        "原因": "评分标准不明确,或评估者有偏见",
        "检测": "计算评估者间一致性(Kappa系数)",
        "解决": "制定详细评分标准,用LLM辅助评估"
    },
    
    "分布偏移忽略": {
        "症状": "评测分数高,实际部署效果差",
        "原因": "评测分布与实际使用场景不同",
        "检测": "对比评测分布和实际请求分布",
        "解决": "构建贴近实际场景的评测集"
    }
}

主流评测集详解

评测集分类地图

# benchmark_overview.py

"""
主流评测集一览

按能力维度分类:
"""

BENCHMARK_CATALOG = {
    # ===== 通用知识与推理 =====
    "MMLU": {
        "全称": "Massive Multitask Language Understanding",
        "任务": "57个学科的选择题(医学、法律、物理等)",
        "规模": "约1.4万道题",
        "代表": "GPT-4: 86.4%, Claude-3: 88.7%, GPT-3.5: 70.0%",
        "特点": "覆盖广,区分度好,被广泛引用",
        "局限": "有数据泄露风险,需要区分开卷/闭卷"
    },
    
    "C-Eval": {
        "全称": "C-Eval: Chinese Evaluation",
        "任务": "52个中文学科的选择题",
        "规模": "约1.3万道题",
        "代表": "GPT-4: 76%, Qwen-72B: 89%",
        "特点": "中文标杆,开源",
        "局限": "部分题目过于简单"
    },
    
    "ARC": {
        "全称": "AI2 Reasoning Challenge",
        "任务": "小学科学题(多选)",
        "规模": "约7800道题",
        "特点": "考察科学推理能力",
        "局限": "对当前模型太简单"
    },
    
    # ===== 数学推理 =====
    "GSM8K": {
        "全称": "Grade School Math 8K",
        "任务": "小学数学应用题(需展示步骤)",
        "规模": "约8500道题",
        "代表": "GPT-4: 92%, GPT-4o: 95%",
        "特点": "考察多步推理",
        "局限": "题目难度有限,模型已接近饱和"
    },
    
    "MATH": {
        "全称": "MATH Dataset",
        "任务": "高中/竞赛数学题",
        "规模": "约1.2万道题",
        "代表": "GPT-4: 42.5%, Gemini-Ultra: 53%",
        "特点": "需要完整解题过程,难度高",
        "局限": "评分依赖解题格式"
    },
    
    # ===== 代码能力 =====
    "HumanEval": {
        "全称": "HumanEval",
        "任务": "Python代码生成(填空+生成)",
        "规模": "164道题",
        "代表": "GPT-4: 67%, Claude-3.5: 92%, GPT-4o: 90%",
        "特点": "代码评测标杆,pass@k指标",
        "局限": "规模小,可能过拟合"
    },
    
    "MBPP": {
        "全称": "Mostly Basic Python Problems",
        "任务": "Python编程题",
        "规模": "974道题",
        "特点": "规模更大,更贴近实际编程",
        "局限": "题目相对简单"
    },
    
    "LiveCodeBench": {
        "全称": "LiveCodeBench",
        "任务": "实时更新的代码评测(每两个月更新)",
        "特点": "避免数据泄露,时间感强",
        "局限": "新题目多,难度波动大"
    },
    
    # ===== 对话/指令遵循 =====
    "MT-Bench": {
        "全称": "Multi-Turn Benchmark",
        "任务": "多轮对话(8个类别×2轮)",
        "代表": "GPT-4: 8.99, Claude-3: 8.06",
        "特点": "评估多轮对话能力,用LLM-as-Judge打分",
        "局限": "评估主观性强"
    },
    
    "AlpacaEval": {
        "全称": "AlpacaEval",
        "任务": "单轮指令遵循",
        "规模": "805道题",
        "特点": "用LLM-as-Judge评估,对比参考模型",
        "局限": "依赖Judge质量"
    },
    
    # ===== Agent能力 =====
    "AgentBench": {
        "全称": "AgentBench",
        "任务": "8个真实软件环境中的Agent任务",
        "环境": "操作系统、数据库、代码仓库、知识图谱等",
        "特点": "评估真实Agent能力",
        "局限": "环境复杂,评测成本高"
    },
    
    # ===== 安全/对齐 =====
    "TruthfulQA": {
        "全称": "TruthfulQA",
        "任务": "测试模型是否会产生虚假信息",
        "规模": "817道题",
        "代表": "GPT-4: 95%, 但人类只有58%",
        "特点": "揭示模型幻觉问题",
        "局限": "有些问题没有标准答案"
    },
    
    "SafetyBench": {
        "全称": "SafetyBench",
        "任务": "安全评测(偏见、道德、法律等)",
        "规模": "约1.1万道题",
        "特点": "中文安全评测",
        "局限": "安全标准随文化变化"
    }
}


def select_benchmarks_for_evaluation(use_case: str) -> list[str]:
    """根据使用场景选择合适的评测集"""
    
    benchmark_map = {
        "通用模型": ["MMLU", "C-Eval", "ARC", "GSM8K", "HumanEval", "MT-Bench"],
        "代码模型": ["HumanEval", "MBPP", "LiveCodeBench", "CRUXEval"],
        "数学模型": ["GSM8K", "MATH", "MMLU"],
        "中文模型": ["C-Eval", "CMMLU", "GAOKAO"],
        "Agent模型": ["AgentBench", "WebArena", "GAIA"],
        "安全模型": ["TruthfulQA", "SafetyBench", "ToxiGen"],
        "对话模型": ["MT-Bench", "AlpacaEval", "ChatArena"],
    }
    
    return benchmark_map.get(use_case, ["MMLU", "GSM8K", "HumanEval"])

评测方法选型:自动指标 vs LLM 当裁判

评测方法分类

# evaluation_methods.py

"""
大模型评测方法

方法1:自动评测(Automatic Evaluation)
├─ 选择题:直接匹配答案
├─ 代码生成:pass@k指标
└─ 数学:精确匹配或部分评分

方法2:LLM-as-a-Judge
├─ 用强LLM评估弱LLM的输出
├─ 需要设计Prompt减少偏见
└─ 适合开放式生成任务

方法3:人工评测
├─ 专家评估(准确性、专业性)
├─ 用户调研(有用性、易用性)
└─ A/B测试(实际用户体验)
"""

# ===== 自动评测示例 =====

def evaluate_multiple_choice(model_answers: list[str], correct_answers: list[str]) -> float:
    """选择题评测:精确匹配"""
    correct = sum(1 for m, c in zip(model_answers, correct_answers) if m.strip() == c.strip())
    return correct / len(correct_answers)


def evaluate_code_generation(
    completions: list[str],
    test_cases: list[list[dict]],
    k_values: list[int] = [1, 10, 100]
) -> dict[int, float]:
    """
    代码生成评测:pass@k
    
    pass@k = 至少有一个top-k样本通过所有测试用例的概率
    """
    import math
    
    results = {}
    
    for k in k_values:
        passed = 0
        for completion, tests in zip(completions, test_cases):
            # 执行代码,检查是否通过测试
            code = completion
            
            # 对每个测试用例执行
            all_passed = True
            for test in tests:
                # 实际执行:略
                pass
            
            if all_passed:
                passed += 1
        
        # pass@k 估算公式
        # 当样本数n>=k时:1 - C(n-k, k) / C(n, k)
        # 简化版:用通过率
        results[k] = passed / len(completions)
    
    return results


def evaluate_math_answer(model_output: str, ground_truth: str) -> dict:
    """
    数学题评测
    
    支持两种评分:
    1. 精确匹配(最终答案)
    2. 过程评分(中间步骤)
    """
    
    # 提取最终答案(简化版:最后一行)
    model_answer = model_output.strip().split("n")[-1]
    
    # 精确匹配
    exact_match = model_answer.strip() == ground_truth.strip()
    
    # 数字提取匹配
    import re
    model_numbers = set(re.findall(r'-?d+.?d*', model_answer))
    gt_numbers = set(re.findall(r'-?d+.?d*', ground_truth))
    number_match = model_numbers == gt_numbers
    
    return {
        "exact_match": exact_match,
        "number_match": number_match,
        "model_answer": model_answer,
        "ground_truth": ground_truth
    }

评测指标详解

# metrics.py

"""
评测指标详解

对于生成任务,核心指标:
1. BLEU:n-gram精确度,偏向短答案
2. ROUGE-L:最长公共子序列,偏向召回
3. METEOR:考虑同义词,语义更准
4. BERTScore:语义相似度
5. G-Eval:LLM-as-Judge的综合评分
"""

from typing import List
import hashlib

# 代码评测特殊指标

CODE_METRICS = {
    "pass@k": """
        核心指标:生成k个代码样本,至少有一个通过测试的概率
        
        公式(估算):
        E[pass@k] = 1 - (C(n-c, k) / C(n, k))
        
        其中:
        - n: 生成的总样本数
        - c: 通过测试的样本数
    """,
    
    "compilation_rate": """
        代码编译通过率
        
        即使不能运行,编译通过也是有价值的信息
    """,
    
    "execution_accuracy": """
        执行准确率
        
        通过所有单元测试的比例
    """,
    
    "test_adaptation": """
        测试适配度
        
        生成的代码是否能正确处理边界情况
    """
}

# 对话评测指标

DIALOG_METRICS = {
    "response_quality": """
        回答质量(1-5分)
        
        评估维度:
        - 有帮助性(Helpfulness)
        - 准确性(Accuracy)
        - 连贯性(Coherence)
        - 安全性(Safety)
    """,
    
    "safety_score": """
        安全分(0-1)
        
        是否包含有害、偏见、不当内容
    """,
    
    "persona_consistency": """
        人设一致性
        
        是否保持设定的角色/风格
    """,
    
    "conversation_length": """
        对话轮次
        
        能否维持长时间对话
    """
}

LLM-as-Judge 实战

为什么需要LLM-as-Judge

# llm_judge_intro.py

"""
LLM-as-a-Judge(LLM裁判)原理

核心思想:用强LLM评估弱LLM的输出质量

适用场景:
- 开放式生成任务(没有标准答案)
- 对话质量评估
- 摘要质量评估
- 需要多维度评估的任务

优势:
- 可处理任意长度输出
- 可评估多个维度(有用性、安全性、连贯性等)
- 成本低于人工评估
- 可大规模部署

挑战:
- 位置偏见(Position Bias):A在B前面时A总被选
- 长度偏见(Length Bias):长答案更容易得高分
- 自我偏好(Self-Preference):强模型偏好自己的输出
- 不一致性:同一答案多次评估结果不同
"""

JUDGE_CHALLENGES = {
    "position_bias": {
        "问题": "LLM倾向于选择第一个选项",
        "解决": "双盲对比(选项顺序打乱)或CoT分析"
    },
    "length_bias": {
        "问题": "长答案更容易得高分,即使内容相同",
        "解决": "控制长度变量,或使用长度归一化"
    },
    "self_preference": {
        "问题": "强模型偏爱自己的输出",
        "解决": "使用第三方Judge模型或人工评估"
    }
}

完整Judge实现

# llm_judge.py

"""
LLM-as-a-Judge完整实现

支持三种模式:
1. Pairwise Comparison:比较两个答案哪个更好
2. Single Answer Scoring:对单个答案打分
3. Multi-Dimensional Assessment:多维度评估
"""

import anthropic
from typing import Literal, Optional
import json

client = anthropic.Anthropic()


class LLMJudge:
    """LLM评判器"""
    
    def __init__(self, model: str = "claude-opus-4-5"):
        self.client = client
        self.model = model
    
    def pairwise_compare(
        self,
        question: str,
        answer_a: str,
        answer_b: str,
        criteria: str = "回答质量、有帮助性和准确性"
    ) -> dict:
        """
        两两对比:判断A好还是B好
        
        返回:
        {
            "winner": "A" | "B" | "tie",
            "reasoning": "判断理由",
            "confidence": 0.0-1.0
        }
        """
        
        prompt = f"""你是一个专业的AI评估专家。请对以下两个答案进行对比评估。

问题:{question}

答案A:
{answer_a}

答案B:
{answer_b}

评估标准:{criteria}

请仔细分析两个答案的优缺点,然后做出判断。

你的输出必须严格遵循以下JSON格式,不要包含任何其他内容:
{{
    "winner": "A" 或 "B" 或 "tie",
    "reasoning": "详细的判断理由,说明为什么选择A/B/tie",
    "confidence": 0.0到1.0之间的置信度
}}

请注意:
1. 如果两个答案质量相当,选择"tie"
2. confidence反映你对自己判断的确定程度
3. 不要有位置偏见,认真评估内容质量
"""
        
        response = self.client.messages.create(
            model=self.model,
            max_tokens=1024,
            messages=[{"role": "user", "content": prompt}]
        )
        
        try:
            result = json.loads(response.content[0].text)
            return result
        except:
            return {"error": "解析失败", "raw": response.content[0].text}
    
    def score_single(
        self,
        question: str,
        answer: str,
        rubric: dict[str, str]
    ) -> dict:
        """
        单答案打分
        
        rubric示例:
        {
            "accuracy": "答案是否准确(1-5分)",
            "completeness": "答案是否完整(1-5分)",
            "clarity": "表达是否清晰(1-5分)",
            "safety": "是否安全无害(1-5分)"
        }
        """
        
        rubric_text = "n".join([f"- {k}: {v}" for k, v in rubric.items()])
        
        prompt = f"""你是一个专业的AI评估专家。请对以下答案进行多维度打分。

问题:{question}

答案:{answer}

评分标准:
{rubric_text}

请对每个维度给出1-5分的评分,并提供简要理由。

输出格式(严格JSON):
{{
    "scores": {{
        "accuracy": {{"score": 4, "reason": "..."}},
        "completeness": {{"score": 5, "reason": "..."}},
        ...
    }},
    "overall_score": 平均分,
    "summary": "总体评价"
}}
"""
        
        response = self.client.messages.create(
            model=self.model,
            max_tokens=1024,
            messages=[{"role": "user", "content": prompt}]
        )
        
        return json.loads(response.content[0].text)
    
    def evaluate_with_cot(
        self,
        question: str,
        answer: str,
        reference_answer: Optional[str] = None
    ) -> dict:
        """
        带Chain-of-Thought的评估
        
        先让LLM思考,再给出最终判断
        减少随机性和偏见
        """
        
        prompt = f"""你是一个严格的AI评估专家。请逐步分析以下答案的质量。

问题:{question}

待评估答案:
{answer}
{f"n参考答案(如果有):n{reference_answer}" if reference_answer else ""}

请按以下步骤进行评估:

步骤1:分析问题类型和难度
- 这是什么类型的问题?(事实/推理/创意/代码等)
- 回答这个问题需要什么能力?

步骤2:评估答案的每个方面
- 准确性:答案是否正确?
- 相关性:答案是否针对问题?
- 完整性:答案是否全面?
- 清晰度:表达是否清楚?

步骤3:识别问题
- 答案有哪些不足?
- 是否有错误或误导?

步骤4:给出最终评分(1-10分)

输出格式(严格JSON):
{{
    "analysis": {{
        "step1": "问题分析",
        "step2": "各维度评估",
        "step3": "问题识别",
        "step4": "最终评分理由"
    }},
    "final_score": 8,
    "key_strengths": ["优势1", "优势2"],
    "key_weaknesses": ["不足1", "不足2"],
    "improvement_suggestions": ["建议1", "建议2"]
}}
"""
        
        response = self.client.messages.create(
            model=self.model,
            max_tokens=2048,
            messages=[{"role": "user", "content": prompt}]
        )
        
        return json.loads(response.content[0].text)


def reduce_bias_in_judge(judge_results: list[dict]) -> list[dict]:
    """
    减少Judge偏见的策略
    """
    
    strategies = {
        "交换位置": "将对比中的A/B互换,重新评估,取一致结果",
        "多Judge验证": "用多个不同的Judge模型,取多数结果",
        "CoT推理": "强制LLM输出推理过程,减少直觉偏见",
        "控制长度": "对比时控制两个答案长度相近",
        "参考标准": "提供标准答案作为参考,减少主观性"
    }
    
    return strategies

G-Eval:自动化综合评分

# g_eval.py

"""
G-Eval(Generative Evaluation)

微软提出的LLM评估方法
用LLM生成评估步骤,再据此评分

核心思想:
1. 让LLM先分析评估标准
2. 基于这些标准生成评分
"""

class GEval:
    """G-Eval评估器"""
    
    def __init__(self, llm_client):
        self.llm = llm_client
    
    def evaluate(
        self,
        task_description: str,
        evaluation_criteria: str,
        model_output: str
    ) -> float:
        """
        G-Eval评分
        
        Args:
            task_description: 任务描述
            evaluation_criteria: 评估标准
            model_output: 模型输出
        """
        
        # 步骤1:生成评估步骤
        steps_prompt = f"""任务:{task_description}

评估标准:{evaluation_criteria}

请分析这个评估任务,列出关键的评估步骤和要点。
用简洁的列表形式输出。
"""
        
        steps_response = self.llm.chat.completions.create(
            model="gpt-4o",
            messages=[{"role": "user", "content": steps_prompt}],
            temperature=0
        )
        evaluation_steps = steps_response.choices[0].message.content
        
        # 步骤2:基于步骤评分
        score_prompt = f"""你是一个专业的评估专家。请根据以下步骤评估模型输出。

任务:{task_description}

评估步骤:
{evaluation_steps}

模型输出:
{model_output}

请严格按照评估步骤,对输出进行逐项检查,然后给出最终评分(1-10分)。
评分要客观公正,不要给不合理的分数。

输出格式:
评分:[分数]
理由:[简要说明]
"""
        
        score_response = self.llm.chat.completions.create(
            model="gpt-4o",
            messages=[{"role": "user", "content": score_prompt}],
            temperature=0
        )
        
        # 提取分数
        response_text = score_response.choices[0].message.content
        score = self._extract_score(response_text)
        
        return score
    
    def _extract_score(self, text: str) -> float:
        """从评估结果中提取分数"""
        import re
        match = re.search(r'评分[::]s*(d+.?d*)', text)
        if match:
            return float(match.group(1))
        return 0.0

构建领域评测集:我踩的 3 个坑

评测集构建流程

# build_benchmark.py

"""
构建领域评测集的完整流程

步骤1:定义评测范围
步骤2:收集种子数据
步骤3:设计题目格式
步骤4:标注与验证
步骤5:划分数据集
步骤6:建立评估标准
"""

# ===== 步骤1:定义评测范围 =====

DOMAIN_EVAL_SCOPE = {
    "任务类型": [
        "知识问答",
        "代码生成",
        "文档理解",
        "对话生成",
        "推理分析"
    ],
    
    "领域边界": {
        "内圈": "核心领域知识(必须准确)",
        "中圈": "相关领域知识(可以提及)",
        "外圈": "通用知识(不应出错)"
    },
    
    "难度分布": {
        "简单": "30%",  # 基础概念、定义
        "中等": "50%",  # 应用、分析
        "困难": "20%"   # 综合、创新
    }
}


# ===== 步骤2:题目生成策略 =====

QUESTION_GENERATION = {
    "人工编写": """
        优点:质量高、针对性强
        缺点:成本高、速度慢
        适用:核心评测集、高风险场景
    """,
    
    "LLM生成+人工审核": """
        优点:成本低、速度快
        缺点:需要大量审核工作
        适用:大规模评测集
    """,
    
    "数据驱动提取": """
        优点:真实、覆盖广
        缺点:需要高质量数据源
        适用:从真实日志中提取测试用例
    """,
    
    "众包标注": """
        优点:规模大、成本适中
        缺点:质量参差不齐
        适用:通用领域
    """
}


# ===== 步骤3:质量控制 =====

QUALITY_CONTROL = {
    "内部一致性": """
        同一题目多次标注的一致性
        指标:Cohen's Kappa >= 0.8
    """,
    
    "专家审核": """
        所有题目经过领域专家审核
        标注:容易/正确/错误/歧义
    """,
    
    "对抗性测试": """
        检查题目是否容易被Prompt注入攻击
    """,
    
    "基准验证": """
        用已知强模型测试题目
        确保题目有区分度
    """
}


def create_benchmark_from_scratch(
    domain: str,
    task_types: list[str],
    size: int = 500,
    seed_data: list[dict] = None
) -> dict:
    """
    从头构建领域评测集
    
    Args:
        domain: 领域名称
        task_types: 任务类型列表
        size: 目标题目数量
        seed_data: 种子数据(可选)
    """
    
    benchmark = {
        "metadata": {
            "domain": domain,
            "task_types": task_types,
            "version": "1.0",
            "size": size,
        },
        
        "train_set": [],  # 用于Few-shot示例
        "dev_set": [],    # 用于调参验证
        "test_set": [],   # 用于最终评测
        
        "evaluation_criteria": {},
        
        "dataset_card": {
            "description": "",
            "motivation": "",
            "known_limitations": [],
        }
    }
    
    # 生成题目
    if seed_data:
        # 基于种子数据扩展
        benchmark["test_set"] = expand_from_seed(seed_data, size)
    else:
        # 从头生成
        benchmark["test_set"] = generate_questions(domain, task_types, size)
    
    # 划分数据集(8:1:1)
    import random
    random.shuffle(benchmark["test_set"])
    n = len(benchmark["test_set"])
    benchmark["train_set"] = benchmark["test_set"][:int(n*0.1)]
    benchmark["dev_set"] = benchmark["test_set"][int(n*0.1):int(n*0.2)]
    benchmark["test_set"] = benchmark["test_set"][int(n*0.2):]
    
    return benchmark

生产级评测 Pipeline 落地

完整评测Pipeline

# production_evaluation.py

"""
生产级评测Pipeline

核心组件:
1. 数据管理器:加载评测集、分片
2. 推理引擎:批量推理、分布式
3. 评估器:多指标、自动+人工
4. 报告生成:可视化、对比分析
"""

import json
from dataclasses import dataclass
from typing import Callable
import time

@dataclass
class EvaluationConfig:
    """评测配置"""
    benchmarks: list[str]
    model_paths: list[str]
    output_dir: str
    num_samples: int  # 每benchmark采样数
    parallel_workers: int
    judge_model: str  # LLM-as-Judge的模型
    seed: int = 42


@dataclass
class EvaluationResult:
    """评测结果"""
    benchmark: str
    model: str
    metrics: dict
    latency_ms: float
    samples: list[dict]  # 详细样本结果


class EvaluationPipeline:
    """评测Pipeline"""
    
    def __init__(self, config: EvaluationConfig):
        self.config = config
        self.results: list[EvaluationResult] = []
    
    def run(self) -> list[EvaluationResult]:
        """运行完整评测流程"""
        
        for model_path in self.config.model_paths:
            # 1. 加载模型
            model = self.load_model(model_path)
            
            for benchmark_name in self.config.benchmarks:
                # 2. 加载评测数据
                test_data = self.load_benchmark(benchmark_name)
                
                # 3. 批量推理
                outputs = self.batch_inference(model, test_data)
                
                # 4. 评估
                metrics = self.evaluate(benchmark_name, outputs, test_data)
                
                # 5. 保存结果
                result = EvaluationResult(
                    benchmark=benchmark_name,
                    model=model_path,
                    metrics=metrics,
                    latency_ms=metrics.get("avg_latency_ms", 0),
                    samples=outputs
                )
                self.results.append(result)
                
                # 6. 生成报告
                self.generate_report(result)
        
        # 7. 生成对比报告
        self.generate_comparison_report()
        
        return self.results
    
    def load_model(self, model_path: str):
        """加载模型"""
        # 实现:根据model_path加载对应模型
        pass
    
    def load_benchmark(self, benchmark_name: str):
        """加载评测集"""
        # 实现:加载benchmark数据
        pass
    
    def batch_inference(
        self,
        model,
        test_data: list[dict]
    ) -> list[dict]:
        """批量推理"""
        outputs = []
        
        for sample in test_data[:self.config.num_samples]:
            start = time.time()
            output = model.generate(sample["prompt"])
            elapsed = (time.time() - start) * 1000
            
            outputs.append({
                "sample_id": sample.get("id", len(outputs)),
                "prompt": sample["prompt"],
                "output": output,
                "ground_truth": sample.get("answer"),
                "latency_ms": elapsed
            })
        
        return outputs
    
    def evaluate(
        self,
        benchmark_name: str,
        outputs: list[dict],
        test_data: list[dict]
    ) -> dict:
        """评估"""
        
        metrics = {}
        
        # 自动指标
        if "choice" in benchmark_name:
            metrics["accuracy"] = self.compute_accuracy(outputs)
        elif "code" in benchmark_name:
            metrics["pass@k"] = self.compute_pass_at_k(outputs)
        elif "math" in benchmark_name:
            metrics["exact_match"] = self.compute_math_accuracy(outputs)
        
        # LLM-as-Judge(如果适用)
        if self.config.judge_model:
            judge_metrics = self.llm_judge_evaluate(outputs)
            metrics.update(judge_metrics)
        
        # 性能指标
        latencies = [o["latency_ms"] for o in outputs]
        metrics["avg_latency_ms"] = sum(latencies) / len(latencies)
        metrics["p50_latency_ms"] = sorted(latencies)[len(latencies)//2]
        metrics["p95_latency_ms"] = sorted(latencies)[int(len(latencies)*0.95)]
        
        return metrics
    
    def generate_report(self, result: EvaluationResult):
        """生成单模型报告"""
        
        report = {
            "benchmark": result.benchmark,
            "model": result.model,
            "timestamp": time.strftime("%Y-%m-%d %H:%M:%S"),
            "metrics": result.metrics,
            "summary": self.summarize_metrics(result.metrics)
        }
        
        report_path = f"{self.config.output_dir}/{result.model}_{result.benchmark}_report.json"
        with open(report_path, "w", encoding="utf-8") as f:
            json.dump(report, f, indent=2, ensure_ascii=False)
    
    def generate_comparison_report(self):
        """生成模型对比报告"""
        
        # 构建对比表格
        comparison = {}
        
        for result in self.results:
            key = result.model
            if key not in comparison:
                comparison[key] = {}
            comparison[key][result.benchmark] = result.metrics
        
        # 生成Markdown表格
        md_report = "# 模型对比评测报告nn"
        
        # 所有benchmark
        benchmarks = list(set(r.benchmark for r in self.results))
        
        for metric_name in ["accuracy", "pass@k", "exact_match", "avg_latency_ms"]:
            md_report += f"## {metric_name}nn"
            md_report += "| Model | " + " | ".join(benchmarks) + " |n"
            md_report += "|------|" + "|------:" * len(benchmarks) + "|n"
            
            for model, bm_results in comparison.items():
                row = [model]
                for bm in benchmarks:
                    if bm in bm_results and metric_name in bm_results[bm]:
                        row.append(f"{bm_results[bm][metric_name]:.4f}")
                    else:
                        row.append("-")
                md_report += "| " + " | ".join(row) + " |n"
            
            md_report += "n"
        
        with open(f"{self.config.output_dir}/comparison_report.md", "w") as f:
            f.write(md_report)

总结

评测体系核心要点:

评测集选择:
- 通用能力:MMLU + GSM8K + HumanEval
- 中文场景:C-Eval + CMMLU
- 对话质量:MT-Bench + AlpacaEval
- 领域场景:必须自建评测集

评估方法:
- 选择题/代码/数学:自动评测(精确匹配/pass@k)
- 开放生成:LLM-as-a-Judge(G-Eval)
- 重要场景:必须人工评估

避免坑点:
- 警惕数据泄露(用私有测试集)
- 警惕位置偏见(打乱对比顺序)
- 多维度评估(不要单一指标定英雄)
- 持续更新(评测集需要动态更新)

写在最后

评测是科学,不是玄学。建立系统化的评测体系,是大模型研发的基础设施投入。一个好的评测体系能让你快速定位问题、对比方案、验证迭代方向。希望这篇能帮你搭建起自己的评测能力。

快速排错表

问题可能原因解决方法
MMLU 分数远低于预期评测集与实际任务不匹配换用领域评测集或 LLM-as-Judge
自动化指标与人工评估不符BLEU/ROUGE 不适合生成任务改用 BERTScore 或 LLM-as-Judge
LLM-as-Judge 评分偏高裁判模型太宽松换用更小的裁判模型,加推理链
领域评测集建了没效果问题太简单或答案太明确加对抗样本、边界 case
评测结果波动大采样随机性或 prompt 不稳定固定 temperature=0,用推理链
评测耗时太长逐条用大模型打分并行 batch 评分,限制 max_tokens

配置检查清单

检查项推荐做法
通用模型评测集MMLU(常识)+ C-Eval(中文)+ GSM8K(数学)
领域任务评测自建评测集,覆盖核心场景
自动化指标ROUGE/BLEU 适合抽取类,BERTScore 更通用
LLM-as-Judge用 gpt-4o 或 claude-3.5,用 CoT 推理链
G-Eval配合评估维度表,输出 1-5 分再加理由
评测集规模每类≥50 条,覆盖正常+边界+对抗
评分稳定性temperature=0,固定 prompt,3 次平均
评测频率上线前基准 + 每次迭代对比

延伸阅读

资源说明
MMLU 论文多任务语言理解基准
G-Eval 论文自动化综合评分
LLM-as-a-Judge用 LLM 当裁判
RAGASRAG 评测专用工具
OpenCompass评测工具链

相关文章

精彩推荐