llm-evaluation

大模型评估

掌握大模型应用的全面评估策略，从自动化指标到人工评估和 A/B 测试。

何时使用此技能

• 系统地衡量大模型应用的性能
• 比较不同模型或提示词
• 在部署前检测性能回退
• 验证提示词变更带来的改进
• 提高对生产系统的信心
• 建立基线并长期跟踪进展
• 调试意外的模型行为
• 评估 RAG 工作流质量（检索 + 生成）
• 衡量 Agent 任务成功率
• 测试结构化输出是否符合架构要求

核心评估类型

1. 自动化指标

使用计算得分进行快速、可重复、可扩展的评估。

文本生成：

• BLEU：N 元组重合度（翻译）
• ROUGE：侧重召回率（摘要）
• METEOR：语义相似度
• BERTScore：基于嵌入的相似度
• Perplexity：语言模型置信度

分类：

• Accuracy：正确率
• Precision/Recall/F1：特定类别的性能
• Confusion Matrix：错误模式
• AUC-ROC：排序质量

检索（RAG）：

• MRR：平均倒数排名
• NDCG：归一化折损累计增益
• Precision@K：前 K 个结果中的相关结果比例
• Recall@K：前 K 个结果中的覆盖率

2. 人工评估

对难以自动化的质量维度进行人工评估。

维度：

• 准确性：事实是否正确
• 连贯性：逻辑是否流畅
• 相关性：是否回答了问题
• 流畅度：自然语言质量
• 安全性：不含有害内容
• 实用性：对用户有帮助

3. LLM-as-Judge

使用能力更强的大模型评估较弱模型的输出。这是 2025/2026 年开放式任务中占主导地位的方法。

方法：

• 逐项评分：为单个回答评分（0-10 李克特量表）
• 成对比较：比较两个回答（MT-Bench、Chatbot Arena 均偏好此方法）
• 基于参考答案：与标准答案进行比较
• 无参考答案：在没有真实答案的情况下进行评判（适用于创意类或开放式任务）
• 基于评分标准：依据明确的标准进行评判（最有利于保持一致性）
• 基于宪法：依据一组原则或规则进行检查

主要挑战：

• 位置偏差：评判模型偏好先出现的回答
• 冗长偏差：无论质量如何，较长的回答通常得分更高
• 自我偏好偏差：模型可能偏好自己的输出
• 缓解方法：调换顺序并取平均值；使用结构化评分标准；使用第三方评判模型

4. RAG 专项评估（RAGAS）

对于检索增强生成工作流，可使用 RAGAS 指标：

from ragas import evaluate
from ragas.metrics import (
    faithfulness,        # Is answer grounded in retrieved context?
    answer_relevancy,   # Does answer address the question?
    context_precision,  # Is retrieved context relevant?
    context_recall,     # Is all necessary info retrieved?
)
from datasets import Dataset

# Prepare evaluation dataset
data = {
    "question": ["What is the capital of France?"],
    "answer": ["Paris is the capital of France."],
    "contexts": [["Paris is a city in France. It is the capital."]],
    "ground_truth": ["Paris"]
}

dataset = Dataset.from_dict(data)
result = evaluate(dataset, metrics=[faithfulness, answer_relevancy, context_precision, context_recall])
print(result)

RAGAS 指标解读：

• 忠实度（0-1）：得分低 = 存在幻觉。对事实性应用至关重要。
• 答案相关性（0-1）：得分低 = 答案偏离主题或含糊其辞。
• 上下文精确率（0-1）：得分低 = 检索结果包含无关片段。
• 上下文召回率（0-1）：得分低 = 检索结果缺少相关文档。

5. Agent 任务评估

对于执行多步任务的 Agent：

class AgentTaskEvaluator:
    """Evaluate agentic task completion."""

    def evaluate_task(self, task, agent_trajectory, expected_result):
        return {
            "task_success": self._check_task_success(agent_trajectory, expected_result),
            "tool_use_accuracy": self._check_tool_selection(agent_trajectory),
            "step_efficiency": self._measure_step_efficiency(agent_trajectory),
            "hallucination_rate": self._check_for_hallucinations(agent_trajectory),
        }

    def _check_task_success(self, trajectory, expected):
        # Did agent achieve the goal? (binary or partial credit)
        final_state = trajectory[-1]["state"]
        return compare_states(final_state, expected)

    def _measure_step_efficiency(self, trajectory):
        # How many extra steps did agent take? (vs. optimal path)
        actual_steps = len(trajectory)
        optimal_steps = self.get_optimal_path_length(trajectory[0]["task"])
        return optimal_steps / actual_steps  # 1.0 = optimal

    def _check_tool_selection(self, trajectory):
        # Did agent use correct tools in correct order?
        correct_tools = sum(1 for step in trajectory if step["tool_correct"])
        return correct_tools / len(trajectory)

Agent 关键指标：

• 任务完成率：完全完成的任务百分比
• 步骤效率：最优步骤数 / 实际执行步骤数
• 工具选择准确率：选择正确工具的次数 / 工具调用总次数
• 错误恢复能力：Agent 是否从错误中恢复？
• 虚构的工具调用：使用编造参数调用工具

快速入门

from llm_eval import EvaluationSuite, Metric

# Define evaluation suite
suite = EvaluationSuite([
    Metric.accuracy(),
    Metric.bleu(),
    Metric.bertscore(),
    Metric.custom(name="groundedness", fn=check_groundedness)
])

# Prepare test cases
test_cases = [
    {
        "input": "What is the capital of France?",
        "expected": "Paris",
        "context": "France is a country in Europe. Paris is its capital."
    },
    # ... more test cases
]

# Run evaluation
results = suite.evaluate(
    model=your_model,
    test_cases=test_cases
)

print(f"Overall Accuracy: {results.metrics['accuracy']}")
print(f"BLEU Score: {results.metrics['bleu']}")

自动化指标实现

BLEU 分数

from nltk.translate.bleu_score import sentence_bleu, SmoothingFunction

def calculate_bleu(reference, hypothesis):
    """Calculate BLEU score between reference and hypothesis."""
    smoothie = SmoothingFunction().method4

    return sentence_bleu(
        [reference.split()],
        hypothesis.split(),
        smoothing_function=smoothie
    )

# Usage
bleu = calculate_bleu(
    reference="The cat sat on the mat",
    hypothesis="A cat is sitting on the mat"
)

ROUGE 分数

from rouge_score import rouge_scorer

def calculate_rouge(reference, hypothesis):
    """Calculate ROUGE scores."""
    scorer = rouge_scorer.RougeScorer(['rouge1', 'rouge2', 'rougeL'], use_stemmer=True)
    scores = scorer.score(reference, hypothesis)

    return {
        'rouge1': scores['rouge1'].fmeasure,
        'rouge2': scores['rouge2'].fmeasure,
        'rougeL': scores['rougeL'].fmeasure
    }

BERTScore

from bert_score import score

def calculate_bertscore(references, hypotheses):
    """Calculate BERTScore using pre-trained BERT."""
    P, R, F1 = score(
        hypotheses,
        references,
        lang='en',
        model_type='microsoft/deberta-xlarge-mnli'
    )

    return {
        'precision': P.mean().item(),
        'recall': R.mean().item(),
        'f1': F1.mean().item()
    }

自定义指标

def calculate_groundedness(response, context):
    """Check if response is grounded in provided context."""
    # Use NLI model to check entailment
    from transformers import pipeline

    nli = pipeline("text-classification", model="microsoft/deberta-large-mnli")

    result = nli(f"{context} [SEP] {response}")[0]

    # Return confidence that response is entailed by context
    return result['score'] if result['label'] == 'ENTAILMENT' else 0.0

def calculate_toxicity(text):
    """Measure toxicity in generated text."""
    from detoxify import Detoxify

    results = Detoxify('original').predict(text)
    return max(results.values())  # Return highest toxicity score

def calculate_factuality(claim, knowledge_base):
    """Verify factual claims against knowledge base."""
    # Implementation depends on your knowledge base
    # Could use retrieval + NLI, or fact-checking API
    pass

LLM-as-Judge 模式

单个输出评估（基于评分标准）

from openai import OpenAI
import json

client = OpenAI()

def llm_judge_quality(response, question):
    """Use GPT-4.1 to judge response quality with structured output."""
    prompt = f"""You are an impartial evaluator. Rate the following response on a scale of 1-10 for each criterion.

**Criteria:**
1. Accuracy (1=many factual errors, 10=completely correct)
2. Helpfulness (1=doesn't address question, 10=fully resolves question)  
3. Clarity (1=confusing/unclear, 10=perfectly clear and well-structured)

**Question:** {question}
**Response:** {response}

Evaluate objectively. Provide ratings in JSON format:
{{
  "accuracy": <1-10>,
  "helpfulness": <1-10>,
  "clarity": <1-10>,
  "reasoning": "<2-3 sentence justification>",
  "overall": <1-10>
}}
"""

    result = client.chat.completions.create(
        model="gpt-4.1",
        messages=[{"role": "user", "content": prompt}],
        temperature=0,
        response_format={"type": "json_object"}  # Structured output
    )

    return json.loads(result.choices[0].message.content)


### Pairwise Comparison (with position bias mitigation)
```python
def compare_responses(question, response_a, response_b):
    """Compare two responses using LLM judge with position bias mitigation."""

    def _judge(q, r1, r2, label1, label2):
        prompt = f"""Compare these two responses to the question. Which is better?

Question: {q}

Response {label1}: {r1}

Response {label2}: {r2}

Which response is better and why? Consider accuracy, helpfulness, and clarity.

Answer with JSON:
{{
  "winner": "{label1}" or "{label2}" or "tie",
  "reasoning": "<explanation>",
  "confidence": <1-10>
}}
"""
        result = client.chat.completions.create(
            model="gpt-4.1",
            messages=[{"role": "user", "content": prompt}],
            temperature=0,
            response_format={"type": "json_object"}
        )
        return json.loads(result.choices[0].message.content)

    # Run twice with swapped order to detect position bias
    result_ab = _judge(question, response_a, response_b, "A", "B")
    result_ba = _judge(question, response_b, response_a, "B", "A")  # swapped

    # Normalize result_ba back to A/B labels
    winner_ba_normalized = "A" if result_ba["winner"] == "B" else ("B" if result_ba["winner"] == "A" else "tie")

    # Check for consistency
    consistent = result_ab["winner"] == winner_ba_normalized
    if not consistent:
        final_winner = "tie"  # Disagree = call it a tie
    else:
        final_winner = result_ab["winner"]

    return {
        "winner": final_winner,
        "consistent": consistent,
        "reasoning_ab": result_ab["reasoning"],
        "reasoning_ba": result_ba["reasoning"],
    }

人工评估框架

标注指南

class AnnotationTask:
    """Structure for human annotation task."""

    def __init__(self, response, question, context=None):
        self.response = response
        self.question = question
        self.context = context

    def get_annotation_form(self):
        return {
            "question": self.question,
            "context": self.context,
            "response": self.response,
            "ratings": {
                "accuracy": {
                    "scale": "1-5",
                    "description": "Is the response factually correct?"
                },
                "relevance": {
                    "scale": "1-5",
                    "description": "Does it answer the question?"
                },
                "coherence": {
                    "scale": "1-5",
                    "description": "Is it logically consistent?"
                }
            },
            "issues": {
                "factual_error": False,
                "hallucination": False,
                "off_topic": False,
                "unsafe_content": False
            },
            "feedback": ""
        }

评审者间一致性

from sklearn.metrics import cohen_kappa_score

def calculate_agreement(rater1_scores, rater2_scores):
    """Calculate inter-rater agreement."""
    kappa = cohen_kappa_score(rater1_scores, rater2_scores)

    interpretation = {
        kappa < 0: "Poor",
        kappa < 0.2: "Slight",
        kappa < 0.4: "Fair",
        kappa < 0.6: "Moderate",
        kappa < 0.8: "Substantial",
        kappa <= 1.0: "Almost Perfect"
    }

    return {
        "kappa": kappa,
        "interpretation": interpretation[True]
    }

A/B 测试

统计检验框架

from scipy import stats
import numpy as np

class ABTest:
    def __init__(self, variant_a_name="A", variant_b_name="B"):
        self.variant_a = {"name": variant_a_name, "scores": []}
        self.variant_b = {"name": variant_b_name, "scores": []}

    def add_result(self, variant, score):
        """Add evaluation result for a variant."""
        if variant == "A":
            self.variant_a["scores"].append(score)
        else:
            self.variant_b["scores"].append(score)

    def analyze(self, alpha=0.05):
        """Perform statistical analysis."""
        a_scores = self.variant_a["scores"]
        b_scores = self.variant_b["scores"]

        # T-test
        t_stat, p_value = stats.ttest_ind(a_scores, b_scores)

        # Effect size (Cohen's d)
        pooled_std = np.sqrt((np.std(a_scores)**2 + np.std(b_scores)**2) / 2)
        cohens_d = (np.mean(b_scores) - np.mean(a_scores)) / pooled_std

        return {
            "variant_a_mean": np.mean(a_scores),
            "variant_b_mean": np.mean(b_scores),
            "difference": np.mean(b_scores) - np.mean(a_scores),
            "relative_improvement": (np.mean(b_scores) - np.mean(a_scores)) / np.mean(a_scores),
            "p_value": p_value,
            "statistically_significant": p_value < alpha,
            "cohens_d": cohens_d,
            "effect_size": self.interpret_cohens_d(cohens_d),
            "winner": "B" if np.mean(b_scores) > np.mean(a_scores) else "A"
        }

    @staticmethod
    def interpret_cohens_d(d):
        """Interpret Cohen's d effect size."""
        abs_d = abs(d)
        if abs_d < 0.2:
            return "negligible"
        elif abs_d < 0.5:
            return "small"
        elif abs_d < 0.8:
            return "medium"
        else:
            return "large"

回归测试

回归检测

class RegressionDetector:
    def __init__(self, baseline_results, threshold=0.05):
        self.baseline = baseline_results
        self.threshold = threshold

    def check_for_regression(self, new_results):
        """Detect if new results show regression."""
        regressions = []

        for metric in self.baseline.keys():
            baseline_score = self.baseline[metric]
            new_score = new_results.get(metric)

            if new_score is None:
                continue

            # Calculate relative change
            relative_change = (new_score - baseline_score) / baseline_score

            # Flag if significant decrease
            if relative_change < -self.threshold:
                regressions.append({
                    "metric": metric,
                    "baseline": baseline_score,
                    "current": new_score,
                    "change": relative_change
                })

        return {
            "has_regression": len(regressions) > 0,
            "regressions": regressions
        }

基准测试

运行基准测试

class BenchmarkRunner:
    def __init__(self, benchmark_dataset):
        self.dataset = benchmark_dataset

    def run_benchmark(self, model, metrics):
        """Run model on benchmark and calculate metrics."""
        results = {metric.name: [] for metric in metrics}

        for example in self.dataset:
            # Generate prediction
            prediction = model.predict(example["input"])

            # Calculate each metric
            for metric in metrics:
                score = metric.calculate(
                    prediction=prediction,
                    reference=example["reference"],
                    context=example.get("context")
                )
                results[metric.name].append(score)

        # Aggregate results
        return {
            metric: {
                "mean": np.mean(scores),
                "std": np.std(scores),
                "min": min(scores),
                "max": max(scores)
            }
            for metric, scores in results.items()
        }

CI/CD 集成：将评估作为代码

在 CI/CD 管道中自动运行评估：

# .github/workflows/eval.yml
name: LLM Evaluation
on:
  pull_request:
    paths: ['prompts/**', 'src/llm/**']

jobs:
  evaluate:
    runs-on: ubuntu-latest
    steps:
      - uses: actions/checkout@v4
      - name: Run evaluation suite
        env:
          OPENAI_API_KEY: ${{ secrets.OPENAI_API_KEY }}
        run: |
          pip install -r requirements-eval.txt
          python scripts/run_evaluations.py --baseline main --compare HEAD
          
      - name: Check for regression
        run: |
          python scripts/check_regression.py \
            --threshold 0.05 \
            --fail-on-regression

# scripts/run_evaluations.py
import argparse
import json
from pathlib import Path

def run_eval_suite(model_fn, test_cases, metrics):
    """Run complete evaluation suite and return results."""
    results = []
    for case in test_cases:
        prediction = model_fn(case["input"])
        scores = {m.name: m.calculate(prediction, case["reference"]) for m in metrics}
        results.append({"case": case["id"], "scores": scores})
    
    aggregated = {
        metric: sum(r["scores"][metric] for r in results) / len(results)
        for metric in results[0]["scores"]
    }
    return aggregated

def main():
    # Load test cases
    test_cases = json.loads(Path("evals/test_cases.json").read_text())
    
    # Run evaluation
    results = run_eval_suite(your_model, test_cases, your_metrics)
    
    # Save results with git commit hash
    import subprocess
    commit = subprocess.check_output(["git", "rev-parse", "HEAD"]).decode().strip()
    
    output = {"commit": commit, "metrics": results}
    Path(f"eval_results/{commit[:8]}.json").write_text(json.dumps(output, indent=2))
    print(json.dumps(results, indent=2))

if __name__ == "__main__":
    main()

结构化输出评估

对于需要结构化输出（JSON 模式、函数调用）的应用：

from pydantic import BaseModel, ValidationError

class ExpectedOutput(BaseModel):
    name: str
    age: int
    email: str

def evaluate_structured_output(model_response: str, expected: dict) -> dict:
    """Evaluate whether model output conforms to schema and matches expected values."""
    
    # 1. Schema compliance
    try:
        parsed = ExpectedOutput.model_validate_json(model_response)
        schema_valid = True
    except ValidationError as e:
        return {"schema_valid": False, "error": str(e), "field_accuracy": 0}
    
    # 2. Field accuracy
    expected_obj = ExpectedOutput(**expected)
    fields = ExpectedOutput.model_fields.keys()
    correct = sum(1 for f in fields if getattr(parsed, f) == getattr(expected_obj, f))
    
    return {
        "schema_valid": True,
        "field_accuracy": correct / len(fields),
        "fields_correct": correct,
        "fields_total": len(fields),
    }

最佳实践

1. 使用多种指标：使用多样化的指标进行全面评估；单一指标无法反映全貌
2. 使用有代表性的数据：使用真实、多样的示例进行测试；也要包含对抗性示例
3. 设置基线：始终与基线表现进行比较（之前版本的提示词、较弱的模型）
4. 保证统计严谨性：使用恰当的统计检验；当 n < 100 时，使用自助法计算置信区间
5. 持续评估：集成到 CI/CD 管道中；每次更改提示词时都进行评估
6. 人工验证：定期将 LLM-as-judge 的输出与人工评分者的结果进行验证
7. 错误分析：对失败案例进行聚类，找出系统性弱点，而不只是统计数量
8. 版本控制：在 git 中跟踪评估数据集和结果；将评估视为代码
9. 缓解位置偏差：在成对比较中交换 A/B 顺序；对结果取平均值
10. 分开开发集和测试集：不要使用测试集调优提示词；保留独立数据

常见陷阱

• 执着于单一指标：为了优化一个指标而牺牲其他指标
• 样本量过小：依据过少的示例得出结论（要获得足够的统计效力，需要 50 个以上示例）
• 数据污染：使用训练数据或用于调优提示词的数据进行测试
• 忽略方差：未考虑统计不确定性；应报告置信区间
• 指标不匹配：对开放式任务使用 BLEU/ROUGE（对于现代大模型，这些指标的相关性较低）
• 评审模型偏差：不检查 LLM 评审模型是否存在系统性偏差
• 基准测试泄漏：热门基准测试可能已包含在训练数据中；优先使用私有评估
• 忽略延迟和成本：如果指标提升导致成本增加三倍，可能并不值得