大模型评估与Benchmark完全教程
本教程全面讲解大语言模型评估与Benchmark的核心技术，帮助开发者构建完整的LLM评估体系。
目录
- 为什么需要大模型评估
- 主流Benchmark基准详解
- 自动化评估框架实战
- LLM-as-Judge评估方法
- 人工评估设计
- 多维度评估体系
- Arena排名与ELO评分
- 评估流水线搭建
- 自定义Benchmark构建
- RAG与Agent专项评估
- 评估最佳实践与常见陷阱
1. 为什么需要大模型评估
大语言模型（LLM）的快速发展带来了模型选择的困境：面对数十甚至上百个开源和闭源模型，如何科学地评估它们的能力？模型评估不仅是学术研究的基础，更是企业选型、产品迭代的关键环节。
1.1 评估的核心价值
- 模型选型：在特定业务场景下选择最合适的模型
- 训练监控：跟踪训练过程中模型能力的变化趋势
- 安全对齐：确保模型输出符合安全和伦理标准
- 迭代优化：量化改进效果，指导后续优化方向
- 合规审计：满足监管机构对AI系统的审查要求
1.2 评估的挑战
大模型评估面临诸多挑战：
- 多维性：模型能力涵盖推理、代码、语言理解、创作等多个维度
- 开放性：开放式生成任务难以用自动化指标精确衡量
- 数据污染：训练数据可能包含测试集内容，导致评估失真
- 评估偏差：评估模型自身可能存在的偏见
- 成本问题：大规模评估需要大量计算资源和人力投入
# 一个简单的评估概念演示
def evaluate_model_concept():
"""
展示评估的基本流程概念
"""
# 1. 准备评估数据
eval_data = [
{"question": "什么是机器学习？", "expected": "机器学习是人工智能的一个分支..."},
{"question": "解释梯度下降", "expected": "梯度下降是一种优化算法..."},
]
# 2. 模型推理
model_outputs = []
for item in eval_data:
output = call_model(item["question"]) # 调用待评估模型
model_outputs.append(output)
# 3. 计算评估指标
scores = []
for pred, item in zip(model_outputs, eval_data):
score = compute_score(pred, item["expected"])
scores.append(score)
# 4. 汇总结果
avg_score = sum(scores) / len(scores)
print(f"模型平均得分: {avg_score:.2f}")
return avg_score
2. 主流Benchmark基准详解
2.1 MMLU（Massive Multitask Language Understanding）
MMLU是最广泛使用的综合知识评估基准，涵盖57个学科领域的14,042道选择题，包括STEM、人文、社科等方向。
特点：
- 覆盖面广，从初中到研究生难度
- 四选一的标准化格式，便于自动化评估
- 已成为模型发布的"标配"评估
import json
from datasets import load_dataset
def evaluate_mmlu(model, tokenizer, subject="all"):
"""
MMLU评估实现
"""
# 加载MMLU数据集
dataset = load_dataset("cais/mmlu", subject)
correct = 0
total = 0
for example in dataset["test"]:
question = example["question"]
choices = example["choices"]
answer = example["answer"] # 0-3
# 构造prompt
prompt = f"""Question: {question}
A. {choices[0]}
B. {choices[1]}
C. {choices[2]}
D. {choices[3]}
Answer:"""
# 模型预测
predicted = predict_choice(model, tokenizer, prompt)
if predicted == answer:
correct += 1
total += 1
accuracy = correct / total
print(f"MMLU Accuracy ({subject}): {accuracy:.2%}")
return accuracy
def predict_choice(model, tokenizer, prompt):
"""
让模型预测选择题答案
"""
import torch
choice_tokens = [tokenizer.encode(c, add_special_tokens=False)[0]
for c in ["A", "B", "C", "D"]]
inputs = tokenizer(prompt, return_tensors="pt")
with torch.no_grad():
outputs = model(**inputs)
# 取最后一个token的logits
last_logits = outputs.logits[0, -1, :]
# 获取A/B/C/D对应的logits
choice_logits = [last_logits[t].item() for t in choice_tokens]
predicted = choice_logits.index(max(choice_logits))
return predicted
2.2 HumanEval（代码生成评估）
HumanEval由OpenAI发布，包含164道Python编程题，通过生成代码并通过测试用例来评估模型的编程能力。
评估指标： pass@k（生成k个样本中至少有一个通过所有测试的概率）
import numpy as np
from typing import List
def estimate_pass_at_k(
num_samples: int,
num_correct: int,
k: int
) -> float:
"""
估算pass@k指标
公式: 1 - C(n-c, k) / C(n, k)
"""
from math import comb
if num_samples - num_correct < k:
return 1.0
return 1.0 - comb(num_samples - num_correct, k) / comb(num_samples, k)
def evaluate_humaneval(model, tokenizer, problems, n_samples=200, k_values=[1, 10, 100]):
"""
HumanEval代码生成评估
"""
results = []
for problem in problems:
prompt = problem["prompt"]
test_code = problem["test"]
entry_point = problem["entry_point"]
# 生成多个代码样本
completions = []
for _ in range(n_samples):
code = generate_completion(model, tokenizer, prompt)
completions.append(prompt + code)
# 检查每个样本是否通过测试
num_correct = 0
for code in completions:
if check_solution(code, test_code, entry_point):
num_correct += 1
results.append({
"task_id": problem["task_id"],
"n_samples": n_samples,
"num_correct": num_correct,
})
# 计算pass@k
pass_at_k = {}
for k in k_values:
estimates = [
estimate_pass_at_k(r["n_samples"], r["num_correct"], k)
for r in results
]
pass_at_k[f"pass@{k}"] = np.mean(estimates)
print("HumanEval Results:")
for metric, value in pass_at_k.items():
print(f" {metric}: {value:.2%}")
return pass_at_k
def check_solution(code: str, test_code: str, entry_point: str) -> bool:
"""
安全执行代码并运行测试用例
"""
import signal
def timeout_handler(signum, frame):
raise TimeoutError()
try:
signal.signal(signal.SIGALRM, timeout_handler)
signal.alarm(10) # 10秒超时
exec_globals = {}
exec(code, exec_globals)
exec(test_code, exec_globals)
signal.alarm(0)
return True
except (TimeoutError, AssertionError, Exception):
signal.alarm(0)
return False
2.3 GSM8K（数学推理评估）
GSM8K包含8,500道小学到初中水平的数学应用题，需要多步推理才能得出正确答案。它是评估模型推理能力的重要基准。
def evaluate_gsm8k(model, tokenizer, dataset):
"""
GSM8K数学推理评估
"""
correct = 0
total = 0
for example in dataset:
question = example["question"]
expected_answer = extract_number(example["answer"])
# 使用Chain-of-Thought提示
prompt = f"""Question: {question}
Let's solve this step by step:
"""
# 生成推理过程
response = generate(model, tokenizer, prompt, max_tokens=512)
# 提取最终答案（通常是####后面的数字）
predicted_answer = extract_number(response)
if predicted_answer == expected_answer:
correct += 1
total += 1
accuracy = correct / total
print(f"GSM8K Accuracy: {accuracy:.2%}")
return accuracy
def extract_number(text: str) -> float:
"""
从文本中提取最终数字答案
"""
import re
# 尝试提取####后面的数字
match = re.search(r"####\s*([\d,.-]+)", text)
if match:
return float(match.group(1).replace(",", ""))
# 尝试提取最后一个数字
numbers = re.findall(r"[-+]?\d*\.?\d+", text)
if numbers:
return float(numbers[-1].replace(",", ""))
return None
2.4 MT-Bench（多轮对话评估）
MT-Bench由LMSYS提出，包含80道多轮对话问题，涵盖写作、角色扮演、推理、数学、编程、提取、STEM、人文8个类别。每道题包含两轮对话，评估模型的对话能力和指令跟随能力。
mt_bench_categories = {
"writing": [
"Compose an engaging travel blog post about a recent trip to Hawaii...",
"Draft a professional email to a potential client proposing a meeting..."
],
"reasoning": [
"A farmer has 17 sheep. All but 9 die. How many are left?",
"If it takes 5 machines 5 minutes to make 5 widgets..."
],
"math": [
"What is the sum of the first 100 positive integers?",
"Solve the equation: 2x + 5 = 17"
],
"coding": [
"Write a Python function to find the longest common subsequence...",
"Implement a binary search tree with insert and search operations..."
],
# ... 更多类别
}
def evaluate_mt_bench(model, judge_model, questions):
"""
MT-Bench评估：使用GPT-4作为评判
"""
scores = []
for q in questions:
# 第一轮对话
response1 = generate(model, q["turn1"])
# 第二轮对话（包含第一轮上下文）
prompt2 = f"User: {q['turn1']}\nAssistant: {response1}\nUser: {q['turn2']}"
response2 = generate(model, prompt2)
# 使用评判模型打分（1-10分）
score1 = judge_model_score(judge_model, q["turn1"], response1)
score2 = judge_model_score(judge_model, q["turn2"], response2)
avg_score = (score1 + score2) / 2
scores.append({
"category": q["category"],
"score": avg_score,
"turn1_score": score1,
"turn2_score": score2,
})
# 按类别汇总
category_scores = {}
for s in scores:
cat = s["category"]
if cat not in category_scores:
category_scores[cat] = []
category_scores[cat].append(s["score"])
for cat, cat_scores in category_scores.items():
print(f"{cat}: {sum(cat_scores)/len(cat_scores):.2f}")
overall = sum(s["score"] for s in scores) / len(scores)
print(f"Overall MT-Bench Score: {overall:.2f}")
return overall
2.5 其他重要Benchmark一览
| Benchmark | 评估维度 | 数据量 | 格式 |
|---|---|---|---|
| HellaSwag | 常识推理 | 10K | 选择题 |
| ARC | 科学推理 | 7.8K | 选择题 |
| TruthfulQA | 真实性 | 817 | 开放式 |
| WinoGrande | 常识推理 | 1.7K | 选择题 |
| MBPP | 代码生成 | 974 | 编程题 |
| MATH | 数学推理 | 12.5K | 解答题 |
| BBH | 综合推理 | 6.5K | 混合 |
| IFEval | 指令跟随 | 541 | 开放式 |
| GPQA | 研究生级QA | 448 | 选择题 |
3. 自动化评估框架实战
3.1 lm-eval-harness
lm-eval-harness是EleutherAI开发的开源评估框架，支持数百个Benchmark，是最流行的LLM评估工具之一。
# 安装
pip install lm-eval
# 基本使用：评估GPT-2在多个任务上的表现
lm_eval --model hf \
--model_args pretrained=gpt2 \
--tasks mmlu,hellaswag,arc_easy \
--device cuda:0 \
--batch_size 16 \
--output_path ./results/
# Python API使用方式
from lm_eval import simple_evaluate
results = simple_evaluate(
model="hf",
model_args="pretrained=meta-llama/Llama-2-7b-hf",
tasks=["mmlu", "hellaswag", "gsm8k"],
batch_size=16,
device="cuda:0",
)
# 打印结果
for task, metrics in results["results"].items():
print(f"\n{task}:")
for metric, value in metrics.items():
if isinstance(value, float):
print(f" {metric}: {value:.4f}")
3.2 OpenCompass
OpenCompass是上海AI Lab开发的综合性评估工具，特别适合中文模型评估。
# OpenCompass配置文件示例 (config.py)
from mmengine.config import read_base
with read_base():
# 选择评估数据集
from .datasets.mmlu.mmlu_gen import mmlu_datasets
from .datasets.ceval.ceval_gen import ceval_datasets
from .datasets.gsm8k.gsm8k_gen import gsm8k_datasets
# 数据集配置
datasets = [*mmlu_datasets, *ceval_datasets, *gsm8k_datasets]
# 模型配置
models = [
dict(
type='HuggingFaceCausalLM',
abbr='llama-2-7b',
path='meta-llama/Llama-2-7b-hf',
max_out_len=100,
batch_size=16,
run_cfg=dict(num_gpus=1),
),
dict(
type='HuggingFaceCausalLM',
abbr='qwen-7b',
path='Qwen/Qwen-7B',
max_out_len=100,
batch_size=16,
run_cfg=dict(num_gpus=1),
),
]
# 评估器配置
eval = dict(
partitioner=dict(type='SizePartitioner', max_task_size=1000),
runner=dict(type='LocalRunner', max_num_workers=4),
)
# 运行OpenCompass评估
python run.py config.py --work-dir ./output/
# 评估单个模型
python run.py config.py --models llama-2-7b --work-dir ./output/
3.3 HELM（Holistic Evaluation of Language Models）
HELM由斯坦福CRFM开发，强调全面性和可复现性。
# HELM评估配置示例
# scenario.json
{
"run_specs": [
{
"class_name": "helm.benchmark.run_spec.RunSpec",
"args": {
"name": "mmlu:subject=anatomy,model=huggingface/gpt2",
"scenario_spec": {
"class_name": "helm.benchmark.scenarios.mmlu_scenario.MMLUScenario",
"args": {"subject": "anatomy"}
},
"adapter_spec": {
"class_name": "helm.benchmark.adapters.adapter.AdapterSpec",
"args": {
"method": "generation",
"max_tokens": 5
}
},
"metric_specs": [
{"class_name": "helm.benchmark.metrics.basic_metrics.ExactMatchMetric"}
]
}
}
]
}
# 运行HELM评估
helm-run --run-specs "mmlu:subject=anatomy" --suite my-suite --max-eval-instances 100
# 查看结果
helm-summarize --suite my-suite
4. LLM-as-Judge评估方法
当传统的自动评估指标（如BLEU、ROUGE）无法有效衡量开放式生成质量时，使用强大的LLM作为评判者成为一种流行且有效的方法。
4.1 基本实现
import openai
from typing import Dict, List
class LLMJudge:
"""使用LLM作为评估者"""
def __init__(self, judge_model="gpt-4"):
self.judge_model = judge_model
self.client = openai.OpenAI()
def score_response(
self,
question: str,
response: str,
reference: str = None,
criteria: List[str] = None
) -> Dict:
"""
对单个回答进行评分
"""
if criteria is None:
criteria = [
"准确性：回答是否事实正确",
"完整性：是否全面回答了问题",
"清晰度：表达是否清楚易懂",
"有用性：对用户是否有实际帮助"
]
criteria_text = "\n".join(f"{i+1}. {c}" for i, c in enumerate(criteria))
system_prompt = """你是一个专业的AI回答评估者。请根据以下标准对回答进行评分。
每个标准给出1-10分的评分，并提供简要理由。
最后给出一个总体评分（1-10分）。
请以JSON格式输出：
{
"scores": {"标准名": {"score": 分数, "reason": "理由"}},
"overall_score": 总分,
"overall_reason": "总体评价"
}"""
user_prompt = f"""问题：{question}
待评估的回答：
{response}
"""
if reference:
user_prompt += f"\n参考答案：{reference}\n"
user_prompt += f"\n评估标准：\n{criteria_text}"
result = self.client.chat.completions.create(
model=self.judge_model,
messages=[
{"role": "system", "content": system_prompt},
{"role": "user", "content": user_prompt}
],
response_format={"type": "json_object"},
temperature=0
)
return json.loads(result.choices[0].message.content)
def pairwise_comparison(
self,
question: str,
response_a: str,
response_b: str
) -> Dict:
"""
两个回答的对比评估
"""
prompt = f"""请比较以下两个回答，判断哪个更好。
问题：{question}
回答A：
{response_a}
回答B：
{response_b}
请从准确性、完整性、清晰度、有用性四个维度进行比较。
输出JSON格式：
{{
"winner": "A" 或 "B" 或 "tie",
"reason": "判断理由",
"dimensions": {{
"accuracy": "A更好/B更好/相当",
"completeness": "A更好/B更好/相当",
"clarity": "A更好/B更好/相当",
"helpfulness": "A更好/B更好/相当"
}}
}}"""
result = self.client.chat.completions.create(
model=self.judge_model,
messages=[{"role": "user", "content": prompt}],
response_format={"type": "json_object"},
temperature=0
)
return json.loads(result.choices[0].message.content)
4.2 处理评估偏差
LLM-as-Judge方法存在已知偏差，需要在设计评估时加以控制：
class BiasAwareJudge(LLMJudge):
"""考虑偏差的LLM评估者"""
def unbiased_pairwise_comparison(
self,
question: str,
response_a: str,
response_b: str
) -> Dict:
"""
通过位置去偏的对比评估
分别进行A-B和B-A两次评估，消除位置偏差
"""
# 正序评估
result_ab = self.pairwise_comparison(question, response_a, response_b)
# 反序评估
result_ba = self.pairwise_comparison(question, response_b, response_a)
# 综合判断
winner_ab = result_ab["winner"]
winner_ba = result_ba["winner"]
# 反转第二次结果
if winner_ba == "A":
winner_ba = "B"
elif winner_ba == "B":
winner_ba = "A"
if winner_ab == winner_ba:
final_winner = winner_ab
confidence = "high"
else:
final_winner = "tie"
confidence = "low"
return {
"winner": final_winner,
"confidence": confidence,
"detail_ab": result_ab,
"detail_ba": result_ba,
}
def calibrate_with_reference(
self,
questions: List[str],
responses: List[str],
references: List[str]
) -> float:
"""
使用有标准答案的题目校准评估者
返回校准偏差值
"""
biases = []
for q, r, ref in zip(questions, responses, references):
# 获取模型评分
score_data = self.score_response(q, r, reference=ref)
model_score = score_data["overall_score"]
# 与标准答案比较
ref_score_data = self.score_response(q, ref)
ref_score = ref_score_data["overall_score"]
# 计算偏差
bias = model_score - ref_score
biases.append(bias)
avg_bias = sum(biases) / len(biases)
print(f"Average calibration bias: {avg_bias:.2f}")
return avg_bias
5. 人工评估设计
5.1 评估流程设计
import random
from dataclasses import dataclass
from typing import Optional
@dataclass
class EvalTask:
"""评估任务定义"""
task_id: str
question: str
response_a: str
response_b: str
model_a: str # 匿名化后的标识
model_b: str
category: str
@dataclass
class EvalResult:
"""评估结果"""
task_id: str
annotator_id: str
winner: str # "A", "B", "tie"
quality_score_a: int # 1-5
quality_score_b: int
dimensions: dict # 各维度评分
feedback: Optional[str] = None
class HumanEvalPipeline:
"""人工评估流水线"""
def __init__(self, tasks: List[dict]):
self.tasks = tasks
self.results = []
def prepare_tasks(
self,
models: List[str],
questions: List[dict],
n_annotators_per_task: int = 3
) -> List[EvalTask]:
"""
准备评估任务，包含匿名化和随机化
"""
eval_tasks = []
for question in questions:
# 随机选择两个模型
model_a, model_b = random.sample(models, 2)
# 生成回答
response_a = generate_response(model_a, question["text"])
response_b = generate_response(model_b, question["text"])
# 随机化顺序（消除位置偏差）
if random.random() > 0.5:
model_a, model_b = model_b, model_a
response_a, response_b = response_b, response_a
task = EvalTask(
task_id=f"{question['id']}_{model_a}_{model_b}",
question=question["text"],
response_a=response_a,
response_b=response_b,
model_a=f"Model_A", # 匿名
model_b=f"Model_B",
category=question.get("category", "general"),
)
eval_tasks.append(task)
return eval_tasks
def compute_agreement(self, results: List[EvalResult]) -> dict:
"""
计算评估者间一致性（Cohen's Kappa）
"""
from collections import Counter
from itertools import combinations
# 按任务分组
task_results = {}
for r in results:
if r.task_id not in task_results:
task_results[r.task_id] = []
task_results[r.task_id].append(r)
agreements = []
for task_id, task_res in task_results.items():
if len(task_res) >= 2:
for r1, r2 in combinations(task_res, 2):
agreements.append(r1.winner == r2.winner)
agreement_rate = sum(agreements) / len(agreements) if agreements else 0
print(f"Inter-annotator agreement: {agreement_rate:.2%}")
return {"agreement_rate": agreement_rate, "n_pairs": len(agreements)}
6. 多维度评估体系
一个完整的LLM评估体系需要覆盖多个能力维度：
from enum import Enum
from dataclasses import dataclass, field
class EvalDimension(Enum):
REASONING = "reasoning" # 推理能力
CODE = "code" # 代码能力
KNOWLEDGE = "knowledge" # 知识储备
LANGUAGE = "language" # 语言能力
SAFETY = "safety" # 安全性
ALIGNMENT = "alignment" # 对齐程度
INSTRUCTION = "instruction" # 指令跟随
CREATIVITY = "creativity" # 创造力
@dataclass
class DimensionConfig:
"""维度配置"""
name: str
weight: float
benchmarks: List[str]
metrics: List[str]
description: str
class MultiDimensionEvaluator:
"""多维度评估器"""
def __init__(self):
self.dimensions = {
EvalDimension.REASONING: DimensionConfig(
name="推理能力",
weight=0.2,
benchmarks=["gsm8k", "math", "bbh", "arc"],
metrics=["accuracy", "pass@k"],
description="逻辑推理、数学计算、科学推理能力"
),
EvalDimension.CODE: DimensionConfig(
name="代码能力",
weight=0.15,
benchmarks=["humaneval", "mbpp", "apps"],
metrics=["pass@k", "test_case_accuracy"],
description="代码生成、调试、理解能力"
),
EvalDimension.KNOWLEDGE: DimensionConfig(
name="知识储备",
weight=0.15,
benchmarks=["mmlu", "ceval", "triviaqa"],
metrics=["accuracy"],
description="各领域知识掌握程度"
),
EvalDimension.SAFETY: DimensionConfig(
name="安全性",
weight=0.15,
benchmarks=["toxicity", "bias", "safetybench"],
metrics=["toxicity_rate", "refusal_rate"],
description="有害内容生成风险、偏见程度"
),
EvalDimension.INSTRUCTION: DimensionConfig(
name="指令跟随",
weight=0.15,
benchmarks=["ifeval", "mt_bench"],
metrics=["follow_rate", "score"],
description="遵循用户指令的能力"
),
EvalDimension.CREATIVITY: DimensionConfig(
name="创造力",
weight=0.1,
benchmarks=["creative_writing"],
metrics=["judge_score"],
description="创意写作、头脑风暴能力"
),
EvalDimension.ALIGNMENT: DimensionConfig(
name="对齐程度",
weight=0.1,
benchmarks=["hh-rlhf", "truthfulqa"],
metrics=["helpful_rate", "harmless_rate"],
description="与人类价值观的一致性"
),
}
def comprehensive_evaluate(self, model_name: str, results: dict) -> dict:
"""
综合多维度评估
"""
dimension_scores = {}
weighted_sum = 0
for dim, config in self.dimensions.items():
# 收集该维度下所有benchmark的结果
dim_scores = []
for bench in config.benchmarks:
if bench in results:
dim_scores.append(results[bench])
if dim_scores:
dim_avg = sum(dim_scores) / len(dim_scores)
else:
dim_avg = 0
dimension_scores[dim.value] = {
"score": dim_avg,
"weight": config.weight,
"name": config.name,
"benchmarks_evaluated": [b for b in config.benchmarks if b in results],
}
weighted_sum += dim_avg * config.weight
# 归一化到0-100分
total_weight = sum(d["weight"] for d in dimension_scores.values() if d["benchmarks_evaluated"])
final_score = (weighted_sum / total_weight * 100) if total_weight > 0 else 0
return {
"model": model_name,
"overall_score": round(final_score, 2),
"dimensions": dimension_scores,
"strengths": self._find_strengths(dimension_scores),
"weaknesses": self._find_weaknesses(dimension_scores),
}
def _find_strengths(self, scores: dict) -> List[str]:
sorted_dims = sorted(scores.items(), key=lambda x: x[1]["score"], reverse=True)
return [d[1]["name"] for d in sorted_dims[:3]]
def _find_weaknesses(self, scores: dict) -> List[str]:
sorted_dims = sorted(scores.items(), key=lambda x: x[1]["score"])
return [d[1]["name"] for d in sorted_dims[:3]]
7. Arena排名与ELO评分
7.1 ELO评分系统
LMSYS Chatbot Arena使用ELO评分系统对模型进行排名，通过用户盲测投票来评估模型相对能力。
class EloRatingSystem:
"""ELO评分系统实现"""
def __init__(self, k_factor=32, initial_rating=1000):
self.k_factor = k_factor
self.initial_rating = initial_rating
self.ratings = {}
def get_rating(self, model: str) -> float:
if model not in self.ratings:
self.ratings[model] = self.initial_rating
return self.ratings[model]
def expected_score(self, rating_a: float, rating_b: float) -> float:
"""计算期望得分"""
return 1 / (1 + 10 ** ((rating_b - rating_a) / 400))
def update_ratings(
self,
model_a: str,
model_b: str,
result: str # "A", "B", "tie"
):
"""
根据对战结果更新评分
"""
rating_a = self.get_rating(model_a)
rating_b = self.get_rating(model_b)
expected_a = self.expected_score(rating_a, rating_b)
expected_b = self.expected_score(rating_b, rating_a)
if result == "A":
score_a, score_b = 1.0, 0.0
elif result == "B":
score_a, score_b = 0.0, 1.0
else: # tie
score_a, score_b = 0.5, 0.5
self.ratings[model_a] = rating_a + self.k_factor * (score_a - expected_a)
self.ratings[model_b] = rating_b + self.k_factor * (score_b - expected_b)
def get_leaderboard(self) -> List[dict]:
"""获取排行榜"""
sorted_models = sorted(
self.ratings.items(),
key=lambda x: x[1],
reverse=True
)
return [
{"rank": i + 1, "model": m, "elo_rating": round(r, 1)}
for i, (m, r) in enumerate(sorted_models)
]
7.2 Bradley-Terry模型
import numpy as np
from scipy.optimize import minimize
class BradleyTerryModel:
"""Bradley-Terry模型用于更精确的模型排名"""
def __init__(self, models: List[str]):
self.models = models
self.n_models = len(models)
self.model_to_idx = {m: i for i, m in enumerate(models)}
# 每对模型的胜负记录
self.wins = np.zeros((self.n_models, self.n_models))
def add_result(self, winner: str, loser: str):
"""记录一场比赛结果"""
i = self.model_to_idx[winner]
j = self.model_to_idx[loser]
self.wins[i][j] += 1
def fit(self) -> dict:
"""
拟合Bradley-Terry模型
最大化似然函数求解各模型的强度参数
"""
def neg_log_likelihood(params):
strength = np.exp(params)
ll = 0
for i in range(self.n_models):
for j in range(self.n_models):
if i != j and self.wins[i][j] > 0:
p_ij = strength[i] / (strength[i] + strength[j])
ll += self.wins[i][j] * np.log(p_ij + 1e-10)
return -ll
# 优化
x0 = np.zeros(self.n_models)
result = minimize(neg_log_likelihood, x0, method='L-BFGS-B')
# 转换为评分
strengths = np.exp(result.x)
strengths = strengths / strengths.mean() # 归一化
rankings = sorted(
zip(self.models, strengths),
key=lambda x: x[1],
reverse=True
)
return {
"rankings": [
{"rank": i + 1, "model": m, "score": round(s, 4)}
for i, (m, s) in enumerate(rankings)
]
}
8. 评估流水线搭建
8.1 完整评估流水线
import os
import json
import time
from datetime import datetime
from pathlib import Path
class EvalPipeline:
"""完整的模型评估流水线"""
def __init__(self, config_path: str):
with open(config_path) as f:
self.config = json.load(f)
self.results_dir = Path(self.config["results_dir"])
self.results_dir.mkdir(parents=True, exist_ok=True)
self.results = {}
def run(self, model_path: str, model_name: str):
"""
执行完整评估流程
"""
print(f"开始评估模型: {model_name}")
start_time = time.time()
eval_id = f"{model_name}_{datetime.now().strftime('%Y%m%d_%H%M%S')}"
# Step 1: 加载模型
model, tokenizer = self.load_model(model_path)
# Step 2: 运行各维度评估
for dimension in self.config["dimensions"]:
print(f"\n 评估维度: {dimension['name']}")
for benchmark in dimension["benchmarks"]:
print(f" 运行: {benchmark['name']}")
try:
score = self.run_benchmark(
model, tokenizer, benchmark
)
self.results[benchmark["name"]] = {
"score": score,
"dimension": dimension["name"],
"timestamp": datetime.now().isoformat(),
}
print(f" 结果: {score:.4f}")
except Exception as e:
print(f" 错误: {e}")
self.results[benchmark["name"]] = {
"score": None,
"error": str(e),
"dimension": dimension["name"],
}
# Step 3: 计算综合得分
overall = self.compute_overall_score()
# Step 4: 保存报告
elapsed = time.time() - start_time
report = {
"eval_id": eval_id,
"model_name": model_name,
"model_path": model_path,
"timestamp": datetime.now().isoformat(),
"elapsed_seconds": elapsed,
"results": self.results,
"overall_score": overall,
}
report_path = self.results_dir / f"{eval_id}.json"
with open(report_path, "w") as f:
json.dump(report, f, indent=2, ensure_ascii=False)
print(f"\n评估完成，耗时 {elapsed:.1f} 秒")
print(f"综合得分: {overall:.2f}")
print(f"报告已保存: {report_path}")
return report
def load_model(self, model_path):
"""加载模型和分词器"""
from transformers import AutoModelForCausalLM, AutoTokenizer
tokenizer = AutoTokenizer.from_pretrained(
model_path, trust_remote_code=True
)
model = AutoModelForCausalLM.from_pretrained(
model_path,
torch_dtype="auto",
device_map="auto",
trust_remote_code=True,
)
return model, tokenizer
def run_benchmark(self, model, tokenizer, benchmark_config):
"""运行单个benchmark"""
benchmark_name = benchmark_config["name"]
if benchmark_name == "mmlu":
return evaluate_mmlu(model, tokenizer)
elif benchmark_name == "humaneval":
return evaluate_humaneval(model, tokenizer, benchmark_config["problems"])
elif benchmark_name == "gsm8k":
return evaluate_gsm8k(model, tokenizer, benchmark_config["dataset"])
else:
raise ValueError(f"Unknown benchmark: {benchmark_name}")
def compute_overall_score(self):
"""计算加权综合得分"""
total_weight = 0
weighted_sum = 0
for dim in self.config["dimensions"]:
dim_scores = []
for bench in dim["benchmarks"]:
if bench["name"] in self.results:
score = self.results[bench["name"]]["score"]
if score is not None:
dim_scores.append(score)
if dim_scores:
dim_avg = sum(dim_scores) / len(dim_scores)
weighted_sum += dim_avg * dim["weight"]
total_weight += dim["weight"]
return (weighted_sum / total_weight) if total_weight > 0 else 0
8.2 评估配置文件示例
{
"results_dir": "./eval_results",
"dimensions": [
{
"name": "知识理解",
"weight": 0.2,
"benchmarks": [
{"name": "mmlu", "dataset": "cais/mmlu", "split": "test"},
{"name": "ceval", "dataset": "ceval/ceval-exam", "split": "val"}
]
},
{
"name": "推理能力",
"weight": 0.25,
"benchmarks": [
{"name": "gsm8k", "dataset": "gsm8k", "split": "test"},
{"name": "math", "dataset": "competition_math", "split": "test"}
]
},
{
"name": "代码能力",
"weight": 0.2,
"benchmarks": [
{"name": "humaneval", "dataset": "openai_humaneval", "split": "test"}
]
},
{
"name": "指令跟随",
"weight": 0.15,
"benchmarks": [
{"name": "ifeval", "dataset": "google/ifeval", "split": "test"}
]
},
{
"name": "安全性",
"weight": 0.2,
"benchmarks": [
{"name": "toxicity", "method": "llm_judge"}
]
}
]
}
9. 自定义Benchmark构建
9.1 领域专用Benchmark设计
from dataclasses import dataclass
from typing import List, Callable
import json
import hashlib
@dataclass
class BenchmarkItem:
"""Benchmark条目"""
item_id: str
category: str
difficulty: str # easy, medium, hard
question: str
reference_answer: str
eval_method: str # exact_match, contains, llm_judge, code_exec
metadata: dict = None
class CustomBenchmarkBuilder:
"""自定义Benchmark构建器"""
def __init__(self, name: str, version: str = "1.0"):
self.name = name
self.version = version
self.items: List[BenchmarkItem] = []
self.eval_methods = {}
def add_item(
self,
category: str,
question: str,
reference_answer: str,
difficulty: str = "medium",
eval_method: str = "llm_judge",
**metadata
) -> str:
"""添加评估条目"""
item_id = hashlib.md5(
f"{category}:{question}".encode()
).hexdigest()[:12]
item = BenchmarkItem(
item_id=item_id,
category=category,
difficulty=difficulty,
question=question,
reference_answer=reference_answer,
eval_method=eval_method,
metadata=metadata,
)
self.items.append(item)
return item_id
def add_from_file(self, file_path: str):
"""从文件批量添加"""
with open(file_path) as f:
data = json.load(f)
for item in data["items"]:
self.add_item(**item)
def validate(self) -> dict:
"""验证Benchmark质量"""
issues = []
# 检查类别分布
categories = {}
for item in self.items:
categories[item.category] = categories.get(item.category, 0) + 1
# 检查难度分布
difficulties = {}
for item in self.items:
difficulties[item.difficulty] = difficulties.get(item.difficulty, 0) + 1
# 检查是否有重复
seen = set()
for item in self.items:
if item.question in seen:
issues.append(f"重复问题: {item.question[:50]}...")
seen.add(item.question)
# 检查最小样本量
for cat, count in categories.items():
if count < 10:
issues.append(f"类别 '{cat}' 样本量不足: {count}")
return {
"total_items": len(self.items),
"categories": categories,
"difficulties": difficulties,
"issues": issues,
"is_valid": len(issues) == 0,
}
def export(self, output_path: str):
"""导出Benchmark"""
benchmark = {
"name": self.name,
"version": self.version,
"created_at": datetime.now().isoformat(),
"total_items": len(self.items),
"items": [
{
"item_id": item.item_id,
"category": item.category,
"difficulty": item.difficulty,
"question": item.question,
"reference_answer": item.reference_answer,
"eval_method": item.eval_method,
"metadata": item.metadata,
}
for item in self.items
],
}
with open(output_path, "w") as f:
json.dump(benchmark, f, indent=2, ensure_ascii=False)
print(f"Benchmark已导出: {output_path}")
print(f"总条目数: {len(self.items)}")
10. RAG与Agent专项评估
10.1 RAG系统评估
class RAGEvaluator:
"""RAG系统评估器"""
def __init__(self):
self.metrics = {}
def evaluate_retrieval(
self,
queries: List[str],
retrieved_docs: List[List[str]],
relevant_docs: List[List[str]],
k_values: List[int] = [1, 3, 5, 10]
) -> dict:
"""
评估检索质量
"""
results = {}
for k in k_values:
precision_at_k = []
recall_at_k = []
mrr_at_k = []
for query, retrieved, relevant in zip(queries, retrieved_docs, relevant_docs):
retrieved_k = retrieved[:k]
relevant_set = set(relevant)
# Precision@K
hits = sum(1 for doc in retrieved_k if doc in relevant_set)
precision_at_k.append(hits / k)
# Recall@K
recall_at_k.append(hits / len(relevant_set) if relevant_set else 0)
# MRR@K
for rank, doc in enumerate(retrieved_k, 1):
if doc in relevant_set:
mrr_at_k.append(1.0 / rank)
break
else:
mrr_at_k.append(0)
results[f"precision@{k}"] = sum(precision_at_k) / len(precision_at_k)
results[f"recall@{k}"] = sum(recall_at_k) / len(recall_at_k)
results[f"mrr@{k}"] = sum(mrr_at_k) / len(mrr_at_k)
return results
def evaluate_generation(
self,
questions: List[str],
answers: List[str],
contexts: List[List[str]],
references: List[str]
) -> dict:
"""
评估RAG生成质量
"""
faithfulness_scores = []
relevance_scores = []
answer_scores = []
for q, a, ctx, ref in zip(questions, answers, contexts, references):
# 忠实度：回答是否基于检索到的上下文
faithfulness = self.check_faithfulness(a, ctx)
faithfulness_scores.append(faithfulness)
# 相关性：回答是否与问题相关
relevance = self.check_relevance(q, a)
relevance_scores.append(relevance)
# 正确性：回答是否正确
correctness = self.check_correctness(a, ref)
answer_scores.append(correctness)
return {
"faithfulness": sum(faithfulness_scores) / len(faithfulness_scores),
"answer_relevance": sum(relevance_scores) / len(relevance_scores),
"correctness": sum(answer_scores) / len(answer_scores),
}
def check_faithfulness(self, answer: str, contexts: List[str]) -> float:
"""检查回答是否忠实于上下文"""
# 使用NLI模型或LLM判断
context_text = "\n".join(contexts)
prompt = f"""判断以下回答是否完全基于给定的上下文。
上下文：
{context_text}
回答：
{answer}
如果回答中的所有信息都能在上下文中找到依据，输出1；否则输出0。
只输出数字。"""
result = call_judge_model(prompt)
return float(result.strip())
def check_relevance(self, question: str, answer: str) -> float:
"""检查回答与问题的相关性"""
prompt = f"""判断以下回答是否直接回答了问题。
问题：{question}
回答：{answer}
评分1-5：
1=完全无关
2=部分相关但未回答问题
3=基本相关
4=高度相关
5=完美回答问题
只输出数字。"""
result = call_judge_model(prompt)
return float(result.strip()) / 5.0
def check_correctness(self, answer: str, reference: str) -> float:
"""检查回答的正确性"""
prompt = f"""判断以下回答与参考答案是否一致。
参考答案：{reference}
回答：{answer}
评分1-5：
1=完全错误
2=大部分错误
3=部分正确
4=大部分正确
5=完全正确
只输出数字。"""
result = call_judge_model(prompt)
return float(result.strip()) / 5.0
10.2 Agent系统评估
class AgentEvaluator:
"""Agent系统评估器"""
def evaluate_task_completion(
self,
tasks: List[dict],
agent_trajectories: List[List[dict]]
) -> dict:
"""
评估Agent任务完成率
"""
completion_rates = {}
for task, trajectory in zip(tasks, agent_trajectories):
category = task.get("category", "general")
# 检查是否达到目标状态
completed = self.check_goal_reached(task, trajectory)
if category not in completion_rates:
completion_rates[category] = {"completed": 0, "total": 0}
completion_rates[category]["total"] += 1
if completed:
completion_rates[category]["completed"] += 1
# 计算各类别完成率
results = {}
for cat, data in completion_rates.items():
results[cat] = data["completed"] / data["total"]
overall = sum(d["completed"] for d in completion_rates.values())
total = sum(d["total"] for d in completion_rates.values())
results["overall"] = overall / total if total > 0 else 0
return results
def evaluate_efficiency(
self,
tasks: List[dict],
trajectories: List[List[dict]]
) -> dict:
"""
评估Agent效率
"""
step_counts = []
tool_calls = []
token_usage = []
for task, trajectory in zip(tasks, trajectories):
step_counts.append(len(trajectory))
tools_used = sum(1 for step in trajectory if step.get("tool_call"))
tool_calls.append(tools_used)
tokens = sum(step.get("tokens_used", 0) for step in trajectory)
token_usage.append(tokens)
return {
"avg_steps": sum(step_counts) / len(step_counts),
"avg_tool_calls": sum(tool_calls) / len(tool_calls),
"avg_tokens": sum(token_usage) / len(token_usage),
"max_steps": max(step_counts),
"min_steps": min(step_counts),
}
def check_goal_reached(self, task: dict, trajectory: List[dict]) -> bool:
"""检查是否达到任务目标"""
if not trajectory:
return False
last_step = trajectory[-1]
# 检查最终状态
if task.get("success_condition"):
return task["success_condition"](last_step)
# 默认检查：最后一步是否包含成功标记
return last_step.get("status") == "success"
11. 评估最佳实践与常见陷阱
11.1 最佳实践
- 多维度评估：不要只看单一指标，要从多个维度全面评估
- 数据质量优先：评估数据的质量比数量更重要
- 可复现性：记录完整的评估配置和环境信息
- 定期评估：建立定期评估机制，跟踪模型能力变化
- 结合人工：自动化评估与人工评估相结合
- 关注分布：不仅看平均分，还要关注分数分布和异常值
11.2 常见陷阱
# 陷阱1：数据污染检测
def detect_data_contamination(model_outputs: List[str], eval_dataset: List[str]) -> dict:
"""
检测模型是否在评估数据上训练过
"""
contamination_scores = []
for output, reference in zip(model_outputs, eval_dataset):
# 检查模型输出是否与评估数据高度相似
similarity = compute_similarity(output, reference)
contamination_scores.append(similarity)
avg_similarity = sum(contamination_scores) / len(contamination_scores)
return {
"avg_similarity": avg_similarity,
"suspicious_count": sum(1 for s in contamination_scores if s > 0.9),
"contamination_risk": "high" if avg_similarity > 0.7 else "medium" if avg_similarity > 0.5 else "low",
}
# 陷阱2：评估指标选择
def choose_metrics(task_type: str) -> List[str]:
"""
根据任务类型选择合适的评估指标
"""
metric_mapping = {
"classification": ["accuracy", "f1", "precision", "recall"],
"generation": ["bleu", "rouge", "bertscore", "llm_judge"],
"qa": ["exact_match", "f1", "llm_judge"],
"code": ["pass@k", "test_case_accuracy"],
"summarization": ["rouge", "bertscore", "llm_judge"],
"translation": ["bleu", "comet", "ter"],
}
return metric_mapping.get(task_type, ["llm_judge"])
# 陷阱3：避免过度优化单一基准
def check_overfitting_to_benchmark(
model_scores: dict,
benchmark_variance: dict
) -> dict:
"""
检查模型是否过度优化特定基准
"""
warnings = []
for benchmark, score in model_scores.items():
# 如果分数异常高，可能存在过拟合
if benchmark in benchmark_variance:
mean = benchmark_variance[benchmark]["mean"]
std = benchmark_variance[benchmark]["std"]
z_score = (score - mean) / std if std > 0 else 0
if z_score > 3:
warnings.append(
f"警告: {benchmark} 得分异常高 (z-score={z_score:.2f})，"
f"可能存在数据污染或过度优化"
)
return {"warnings": warnings}
总结
大模型评估是一个系统工程，需要从多个维度、使用多种方法进行综合评估。本教程涵盖了从基础Benchmark到高级评估方法的完整知识体系，帮助你构建专业的LLM评估能力。
关键要点：
- 选择合适的Benchmark组合，覆盖目标能力维度
- 使用自动化框架提高评估效率，同时保留人工评估作为质量保障
- 警惕数据污染和评估偏差，确保评估结果的可信度
- 建立持续的评估流水线，跟踪模型能力演进
- 根据实际需求构建自定义Benchmark，贴近真实使用场景
评估不是终点，而是模型优化的起点。通过科学的评估体系，你可以更精准地识别模型短板，指导后续的训练和优化工作。
