📚 AI 工程与智能体教程站

第 12 章 · LLM-as-judge 评估器 ​

本章目标：

  • 理解 LLM-as-judge 的核心概念与适用场景
  • 掌握 Judge 提示词设计：rubric + few-shot examples
  • 学会批量评估脚本的编写与执行
  • 建立自动化评估管线：generate → judge → 统计
  • 实战：评估 RAG 回答的准确性（faithfulness + relevance）

12.1 什么是 LLM-as-judge ​

在构建和迭代 Agent 系统时，一个核心问题是：如何知道你的 Agent 是否"做得好"？

传统做法是人工审查输出——找 10 条测试数据，逐条判断质量。这在数据量少时可行，但当你的 Agent 每天处理数千条请求、或者你需要反复迭代提示词时，人工评估成为瓶颈。

LLM-as-judge 的思路很简单：让另一个 LLM 来评估第一个 LLM 的输出质量。

┌─────────────┐     ┌─────────────┐
│  Test Data  │────▶│   Judge     │
│  (question) │     │  (LLM)      │
└─────────────┘     └─────────────┘
                            │
                            ▼
                     ┌─────────────┐
                     │  Score +     │
                     │  Explanation │
                     └─────────────┘

LLM-as-judge 的优势在于速度快、成本低、可批量执行。一个 GPT-4o-mini 作为 judge，每次调用成本不到 $0.001，可以在几秒内完成数百条数据的评估。

何时该用 LLM-as-judge ​

场景适合度说明
开放式文本生成⭐⭐⭐主观质量判断，LLM 擅长
RAG 回答准确性⭐⭐⭐可评估 faithfulness + relevance
代码生成质量⭐⭐可评估正确性、可读性
分类/抽取任务⭐有客观答案，用精确匹配更可靠
数学计算✗用解析解验证，不要用 judge

关键认知：Judge 本身也会犯错 ​

LLM-as-judge 不是完美的——judge 模型有自己的偏差、会"过于严格"或"过于宽松"。但研究表明，当 judge 和被评估的是同代或更代模型时，评估质量足够可靠。

实践建议：

  • 始终保留一条人工抽检管道（每 100 条随机抽 5 条）；
  • 记录 judge 的解释，便于调试；
  • 不同 judge 模型的结果可做交叉验证。

12.2 Judge 提示词设计 ​

Judge 的效果核心在于提示词设计。一个优秀的 judge prompt 应包含三个要素：

  1. 角色定义：明确 judge 的职责
  2. 评分标准（Rubric）：量化的打分依据
  3. 示例（Few-shot）：让 judge 理解标准的具体表现

12.2.1 Rubric 设计原则 ​

Rubric 必须具体、可操作、避免模糊形容词。

❌ 坏的 rubric：

"评估回答是否质量高"

✅ 好的 rubric（RAG faithfulness 评估）：

- 2分：回答完全基于参考文本，无额外信息
- 1分：回答基本基于参考文本，包含少量合理推断
- 0分：回答与参考文本无关或包含明显错误

12.2.2 Few-shot 示例的价值 ​

示例让 judge 将抽象标准具象化。一般提供 2-3 个正例和 1-2 个反例效果最佳。

markdown
### 评分示例

**问题**: 什么是量子纠缠？
**参考文本**: 量子纠缠是量子力学中的一种现象，当两个粒子相互作用后，即使相隔很远，它们的状态仍保持关联。
**回答A**: 量子纠缠是粒子间的一种神秘连接，无论多远都能瞬间影响对方。
→ 评分：2分（完全基于参考文本，未添加额外信息）

**回答B**: 量子纠缠是量子力学现象，粒子状态保持关联。此外，薛定谔猫悖论也与此相关。
→ 评分：1分（基于参考文本，但添加了超出参考范围的信息）

**回答C**: 量子纠缠就是两个物体之间的磁力作用。
→ 评分：0分（与参考文本无关，且包含错误信息）

12.3 批量评估脚本 ​

有了 judge prompt，下一步是编写批量评估脚本。核心逻辑：

for each test_case in test_data:
    answer = generate(test_case.question)
    score = judge.evaluate(answer, test_case.reference)
    log(score, answer)

12.3.1 数据结构定义 ​

python
from dataclasses import dataclass
from typing import Optional

@dataclass
class TestCase:
    question: str
    reference_context: str  # 用于 RAG 场景的参考文本
    expected_answer: Optional[str] = None  # 可选的期望答案
    metadata: dict = None

12.3.2 Judge Prompt 构造器 ​

python
from typing import Dict, Any
import json

class RagJudgePrompt:
    """RAG 回答评估的 Judge prompt 模板"""
    
    SYSTEM_PROMPT = """\
你是一个专业的 RAG 系统评估员。你的任务是评估 AI 助手的回答质量。

评估维度：Faithfulness（忠实度）
- 衡量回答是否忠实于参考文本，没有添加参考文本外的信息

评分标准（0-2分）：
- 2分：回答完全基于参考文本，无任何额外信息
- 1分：回答基于参考文本，但包含少量合理推断或补充
- 0分：回答与参考文本无关，或包含明显错误信息

请只输出 JSON 格式的结果，格式如下：
{"score": 整数分数, "explanation": "简短理由"}
"""
    
    def build_prompt(
        self,
        question: str,
        context: str,
        answer: str
    ) -> str:
        return f"""\
{self.SYSTEM_PROMPT}

请评估以下回答：

【问题】
{question}

【参考文本】
{context}

【AI 回答】
{answer}

请输出评估结果："""

12.4 自动化评估管线 ​

完整的评估管线包含三个阶段：

┌──────────────┐    ┌──────────────┐    ┌──────────────┐
│  Generate    │───▶│    Judge     │───▶│   Stats      │
│  (Step 1)    │    │  (Step 2)    │    │  (Step 3)    │
└──────────────┘    └──────────────┘    └──────────────┘

12.4.1 完整评估管线代码 ​

python
import asyncio
import json
from dataclasses import dataclass
from typing import List, Dict, Any
from openai import AsyncOpenAI

@dataclass
class EvaluationResult:
    test_case: TestCase
    generated_answer: str
    score: int
    explanation: str
    latency_ms: float

class RagEvaluator:
    def __init__(self, api_key: str):
        self.client = AsyncOpenAI(api_key=api_key)
        self.prompt_builder = RagJudgePrompt()
    
    async def generate_answer(
        self,
        question: str,
        context: str
    ) -> str:
        """模拟 RAG 生成步骤"""
        prompt = f"""\
基于以下参考文本回答问题。如果参考文本中没有相关信息，请如实说明。

【参考文本】
{context}

【问题】
{question}

【回答】"""
        
        response = await self.client.chat.completions.create(
            model="gpt-4o-mini",
            messages=[{"role": "user", "content": prompt}],
            max_tokens=500
        )
        return response.choices[0].message.content.strip()
    
    async def evaluate(
        self,
        test_case: TestCase,
        generated_answer: str
    ) -> EvaluationResult:
        """执行单次评估"""
        prompt = self.prompt_builder.build_prompt(
            question=test_case.question,
            context=test_case.reference_context,
            answer=generated_answer
        )
        
        import time
        start = time.time()
        
        response = await self.client.chat.completions.create(
            model="gpt-4o-mini",
            messages=[{"role": "user", "content": prompt}],
            max_tokens=200
        )
        
        latency = (time.time() - start) * 1000
        
        # 解析 judge 输出
        raw_output = response.choices[0].message.content.strip()
        try:
            result = json.loads(raw_output)
            score = result.get("score", 0)
            explanation = result.get("explanation", "")
        except json.JSONDecodeError:
            # 降级处理：提取数字
            import re
            match = re.search(r'\b([0-2])\b', raw_output)
            score = int(match.group(1)) if match else 0
            explanation = raw_output[:100]
        
        return EvaluationResult(
            test_case=test_case,
            generated_answer=generated_answer,
            score=score,
            explanation=explanation,
            latency_ms=latency
        )
    
    async def run_batch(
        self,
        test_cases: List[TestCase],
        max_concurrent: int = 5
    ) -> List[EvaluationResult]:
        """批量评估"""
        semaphore = asyncio.Semaphore(max_concurrent)
        
        async def evaluate_one(tc: TestCase) -> EvaluationResult:
            async with semaphore:
                answer = await self.generate_answer(tc.question, tc.reference_context)
                return await self.evaluate(tc, answer)
        
        tasks = [evaluate_one(tc) for tc in test_cases]
        return await asyncio.gather(*tasks)
    
    def summarize(self, results: List[EvaluationResult]) -> Dict[str, Any]:
        """生成评估报告"""
        total = len(results)
        scores = [r.score for r in results]
        
        # 计算平均分数
        avg_score = sum(scores) / total if total > 0 else 0
        
        # 计算各分数段分布
        score_dist = {0: 0, 1: 0, 2: 0}
        for s in scores:
            score_dist[s] = score_dist.get(s, 0) + 1
        
        # 计算平均延迟
        avg_latency = sum(r.latency_ms for r in results) / total if total > 0 else 0
        
        # 生成详细报告
        report = {
            "total_tests": total,
            "average_score": round(avg_score, 2),
            "score_distribution": score_dist,
            "pass_rate": round(score_dist.get(2, 0) / total * 100, 1) if total > 0 else 0,
            "avg_latency_ms": round(avg_latency, 1),
            "detailed_results": [
                {
                    "question": r.test_case.question[:50] + "...",
                    "score": r.score,
                    "explanation": r.explanation[:80],
                    "latency_ms": round(r.latency_ms, 1)
                }
                for r in results
            ]
        }
        
        return report

12.4.2 运行评估 ​

python
async def main():
    # 测试数据集
    test_cases = [
        TestCase(
            question="什么是 LangGraph？",
            reference_context="LangGraph 是 LangChain 团队开发的图结构工作流引擎，"
                           "用于构建具有循环、分支和状态的复杂 Agent 应用。"
                           "它基于 StateGraph API，支持工具调用、子图组合和持久化。"
            ),
        TestCase(
            question="Deep Agents 是什么？",
            reference_context="Deep Agents 是 LangChain 推出的高层 Agent SDK，"
                           "提供 create_deep_agent 函数快速构建具有文件操作、"
                           "子代理、记忆等能力的 Agent 系统。"
            ),
        TestCase(
            question="RAG 系统的主要组件有哪些？",
            reference_context="RAG（检索增强生成）系统通常包含四个核心组件："
                           "1) 文档加载与分割；2) 向量嵌入与存储；"
                           "3) 语义检索；4) 上下文注入与 LLM 生成。"
            ),
    ]
    
    # 初始化评估器
    evaluator = RagEvaluator(api_key="your-openai-api-key")
    
    # 运行批量评估
    results = await evaluator.run_batch(test_cases)
    
    # 生成报告
    report = evaluator.summarize(results)
    print(json.dumps(report, indent=2, ensure_ascii=False))

if __name__ == "__main__":
    asyncio.run(main())

12.5 实战：评估 RAG 回答质量 ​

下面是一个完整的端到端示例，评估一个简易 RAG 系统的回答质量。

12.5.1 项目结构 ​

rag-evaluator/
├── config.py              # 配置
├── judge.py               # Judge prompt 和评分逻辑
├── evaluator.py           # 评估管线
├── test_cases.py          # 测试数据集
└── run_evaluation.py      # 入口脚本

12.5.2 测试数据集 ​

python
# test_cases.py
from dataclasses import dataclass

@dataclass
class TestCase:
    question: str
    context: str
    expected_keywords: list = None
    
TEST_CASES = [
    {
        "question": "LangGraph 是什么？",
        "context": "LangGraph 是 LangChain 开发的图结构 Agent 编排引擎。"
                  "核心抽象是 StateGraph，允许开发者定义节点（nodes）和边（edges）。"
                  "支持工具调用（ToolNode）、条件路由（conditional edges）和循环结构。",
        "expected_keywords": ["LangChain", "StateGraph", "节点", "边"]
    },
    {
        "question": "Deep Agents 和 LangGraph 有什么区别？",
        "context": "Deep Agents 是 LangChain 推出的高级 Agent SDK，"
                  "封装了 LangGraph 的能力。"
                  "它提供 create_deep_agent 函数，开箱即用地支持：文件操作、"
                  "子代理（subagents）、记忆（memory）和技能（skills）。"
                  "LangGraph 是底层编排引擎，Deep Agents 是其上层封装。",
        "expected_keywords": ["封装", "高级", "File", "子代理"]
    },
    {
        "question": "什么是 RAG？",
        "context": "RAG（Retrieval-Augmented Generation，检索增强生成）是一种将"
                  "外部知识库与 LLM 结合的技术。"
                  "工作流程：1) 将文档分块并生成嵌入；2) 用户提问时检索相关文档；"
                  "3) 将检索结果作为上下文注入 prompt；4) LLM 基于上下文生成回答。"
                  "优势：减少幻觉，提供可溯源的回答。",
        "expected_keywords": ["检索", "增强", "嵌入", "上下文"]
    },
]

12.5.3 完整评估脚本 ​

python
# run_evaluation.py
import asyncio
import json
from openai import AsyncOpenAI
from test_cases import TEST_CASES
from judge import RagJudgePrompt

class RagEvaluator:
    def __init__(self, api_key: str, judge_model: str = "gpt-4o-mini"):
        self.client = AsyncOpenAI(api_key=api_key)
        self.judge_model = judge_model
        self.prompt_builder = RagJudgePrompt()
    
    async def generate_rag_answer(self, question: str, context: str) -> str:
        """模拟 RAG 生成过程"""
        prompt = f"""\
你是一名技术支持助手。请基于【参考文档】回答用户问题。
如果参考文档中找不到答案，请明确说明"参考文档未提供相关信息"。

【参考文档】
{context}

【用户问题】
{question}

【回答】"""
        
        response = await self.client.chat.completions.create(
            model="gpt-4o-mini",
            messages=[{"role": "user", "content": prompt}],
            max_tokens=300,
            temperature=0.1
        )
        return response.choices[0].message.content.strip()
    
    async def judge_answer(
        self,
        question: str,
        context: str,
        answer: str
    ) -> dict:
        """使用 LLM 评估回答质量"""
        prompt = self.prompt_builder.build_prompt(question, context, answer)
        
        response = await self.client.chat.completions.create(
            model=self.judge_model,
            messages=[{"role": "user", "content": prompt}],
            max_tokens=150,
            temperature=0
        )
        
        raw = response.choices[0].message.content.strip()
        try:
            return json.loads(raw)
        except json.JSONDecodeError:
            # 解析降级：提取 score 数字
            import re
            match = re.search(r'"score"\s*:\s*(\d)', raw)
            score = int(match.group(1)) if match else 0
            return {"score": score, "explanation": raw[:100]}
    
    async def evaluate_one(self, tc: dict) -> dict:
        """评估单个测试用例"""
        # Step 1: 生成回答
        answer = await self.generate_rag_answer(tc["question"], tc["context"])
        
        # Step 2: Judge 评估
        judgment = await self.judge_answer(tc["question"], tc["context"], answer)
        
        return {
            "question": tc["question"],
            "context": tc["context"][:100] + "...",
            "answer": answer,
            "score": judgment["score"],
            "explanation": judgment["explanation"]
        }
    
    async def run_all(self, test_cases: list) -> dict:
        """运行全部评估"""
        results = []
        for tc in test_cases:
            result = await self.evaluate_one(tc)
            results.append(result)
        
        # 统计
        scores = [r["score"] for r in results]
        avg_score = sum(scores) / len(scores) if scores else 0
        
        return {
            "results": results,
            "summary": {
                "total": len(results),
                "average_score": round(avg_score, 2),
                "score_distribution": {
                    0: scores.count(0),
                    1: scores.count(1),
                    2: scores.count(2)
                }
            }
        }

async def main():
    evaluator = RagEvaluator(api_key="your-api-key-here")
    report = await evaluator.run_all(TEST_CASES)
    
    print(json.dumps(report, indent=2, ensure_ascii=False))
    
    # 输出详细结果
    print("\n=== 详细评估 ===")
    for r in report["results"]:
        print(f"\n问题: {r['question']}")
        print(f"得分: {r['score']}/2")
        print(f"理由: {r['explanation']}")
        print(f"回答: {r['answer'][:100]}...")

if __name__ == "__main__":
    asyncio.run(main())

12.5.4 评估结果解读 ​

运行后会得到类似这样的输出：

json
{
  "results": [
    {
      "question": "LangGraph 是什么？",
      "score": 2,
      "explanation": "回答完全基于参考文本，准确描述了 LangGraph 的核心概念。"
    },
    {
      "question": "Deep Agents 和 LangGraph 有什么区别？",
      "score": 1,
      "explanation": "回答基于参考文本，但添加了少量合理推断。"
    }
  ],
  "summary": {
    "total": 3,
    "average_score": 1.67,
    "score_distribution": {"0": 0, "1": 1, "2": 2}
  }
}

关键指标：

  • average_score：整体质量，越高越好（满分 2）
  • score_distribution：质量分布，期望集中在 2 分
  • pass_rate：2 分占比，可设定阈值（如 ≥80% 为通过）

12.6 进阶：多维度评估 ​

生产环境的评估往往需要多维度。以 RAG 为例，常见维度包括：

维度含义评分标准
Faithfulness回答是否忠于参考0-2分（见上文）
Relevance回答是否切题0-2分
Completeness回答是否完整0-2分
Hallucination是否有编造内容0-1分（二分类）

12.6.1 多维 Judge Prompt ​

python
class MultiDimensionalJudge:
    """多维度评估器"""
    
    SYSTEM_PROMPT = """\
你是一个专业的 RAG 系统评估员。请从以下维度评估 AI 回答：

1. Faithfulness（忠实度）: 0-2分
   - 2: 完全基于参考文本
   - 1: 基本基于参考，含少量推断
   - 0: 与参考无关或有错误

2. Relevance（相关性）: 0-2分
   - 2: 直接回答提问
   - 1: 部分相关
   - 0: 无关

3. Completeness（完整性）: 0-2分
   - 2: 覆盖所有关键点
   - 1: 覆盖主要点，略有不完整
   - 0: 严重遗漏

输出 JSON: {"faithfulness": N, "relevance": N, "completeness": N}
"""

12.7 与现有课程的关联 ​

本章内容可关联以下已有课程：

  • Vercel AI SDK ch03：Provider 管理，用于配置 judge 模型的 API key
  • Agent 工程 ch12：CoT 推理，可应用于 judge 的思考链设计
  • FastAPI ch22：Docker 部署，可将评估服务容器化

本章小结 ​

  • LLM-as-judge：用 LLM 评估 LLM 输出，速度快成本低，适合开放式文本质量判断
  • Judge prompt 设计三要素：角色定义 + 量化 rubric + few-shot 示例
  • 评估管线三步走：Generate → Judge → Stats，可完全自动化
  • RAG 评估常用维度：Faithfulness（忠实度）+ Relevance（相关性）+ Completeness（完整性）
  • 关键认知：Judge 本身也会犯错，保留人工抽检和交叉验证

🛠️ 动手实践 ​

  1. 修改 judge.py 中的 RagJudgePrompt，增加 Hallucination 维度（回答是否包含参考文本外的内容），重新运行评估并观察分数变化。
  2. 扩展测试数据集至 10 条，运行批量评估，分析平均分数分布，找出得分最低的测试用例并改进提示词。
  3. 将评估脚本改造为 FastAPI 接口，接受 {question, context} 输入，返回评估结果 JSON，参考 FastAPI ch01-06。

上一章：上下文工程（即将发布） | 下一章：A/B 测试与对照实验（即将发布）