LLM-as-Judge 评估方法实战

SINGLE_GRADING_PROMPT = """你是一个专业的评估专家。请对以下回答进行评分。

## 问题
{question}

## 回答
{answer}

## 评分维度
请按以下维度分别打分（1-10分）：

1. **准确性** (Accuracy)：回答中的事实信息是否正确？
2. **完整性** (Completeness)：是否充分回答了问题的各个方面？
3. **清晰度** (Clarity)：表达是否清晰、逻辑是否连贯？
4. **实用性** (Helpfulness)：对提问者是否有实际帮助？

## 输出格式
请以 JSON 格式输出：
```json
{{
  "accuracy": {{"score": 0, "reason": ""}},
  "completeness": {{"score": 0, "reason": ""}},
  "clarity": {{"score": 0, "reason": ""}},
  "helpfulness": {{"score": 0, "reason": ""}},
  "overall": {{"score": 0, "reason": ""}}
}}

单答案评分的优点是简单直接，缺点是评分容易受 Judge 模型的系统性偏差影响——比如倾向于给高分（lenency bias）或中间分（central tendency bias）。

### 模式二：参考答案对比（Reference-Guided Grading）

给定一个"标准答案"，让 Judge 模型将待评估答案与标准答案对比：

```python
REFERENCE_GRADING_PROMPT = """你是一个专业的评估专家。请将待评估答案与参考答案进行对比评分。

## 问题
{question}

## 参考答案
{reference_answer}

## 待评估答案
{answer}

## 评分要求
请评估待评估答案相对于参考答案的质量。每个维度评分1-10分：
- 10分：与参考答案质量相当或更好
- 7-9分：接近参考答案，有细微差距
- 4-6分：有明显不足但方向正确
- 1-3分：严重偏离或错误

## 维度
1. 事实一致性：与参考答案的事实是否一致？
2. 覆盖度：是否覆盖了参考答案的要点？
3. 额外价值：是否提供了参考答案之外的有用信息？
4. 表达质量：语言组织是否与参考答案同样优秀？

请以 JSON 格式输出评分结果。
"""

PAIRWISE_PROMPT = """你是一个专业的评估专家。请比较以下两个回答的优劣。

## 问题
{question}

## 回答 A
{answer_a}

## 回答 B
{answer_b}

## 评判标准
请从以下方面综合判断哪个回答更好：
- 事实准确性
- 回答完整性
- 逻辑清晰度
- 实用价值

## 输出格式
请输出 JSON：
```json
{{
  "winner": "A" | "B" | "tie",
  "reasoning": "详细分析...",
  "confidence": 0.0-1.0
}}

成对比较是竞技场（Arena）排行榜的核心方法。它的优势在于相对评估比绝对评分更容易，Judge 模型不需要校准分数尺度，只需要做比较。但缺点是评估成本高——N 个模型两两比较需要 O(N²) 次评估。

## 构建可靠的 Judge Pipeline

### 第一步：选择合适的 Judge 模型

Judge 模型的选择直接影响评估质量。一般原则：

- **能力上限原则**：Judge 模型的能力应高于被评估模型，否则"菜鸡互啄"
- **角色分离原则**：如果可能，使用不同家族的模型作为 Judge，避免同家族模型的系统性偏好
- **成本考量**：大批量评估时，需要在 Judge 模型的能力和推理成本之间取得平衡

实践中推荐的 Judge 模型选择：

| 被评估模型 | 推荐 Judge 模型 |
|-----------|----------------|
| GPT-3.5 级别 | GPT-4o / Claude 3.5 Sonnet |
| GPT-4 级别 | Claude 3.5 Opus / GPT-4o + 人工抽样 |
| 开源模型 | GPT-4o / Claude 3.5 Sonnet |
| 特定领域模型 | 领域专家模型 + 通用大模型双重检验 |

### 第二步：设计评分维度

评分维度的设计需要根据具体任务定制。以下是一个通用框架：

```python
EVALUATION_DIMENSIONS = {
    "general_qa": [
        {"name": "accuracy", "weight": 0.35, "description": "事实准确性"},
        {"name": "completeness", "weight": 0.25, "description": "回答完整性"},
        {"name": "clarity", "weight": 0.20, "description": "表达清晰度"},
        {"name": "safety", "weight": 0.20, "description": "内容安全性"},
    ],
    "code_generation": [
        {"name": "correctness", "weight": 0.40, "description": "代码正确性"},
        {"name": "efficiency", "weight": 0.20, "description": "算法效率"},
        {"name": "readability", "weight": 0.20, "description": "代码可读性"},
        {"name": "edge_cases", "weight": 0.20, "description": "边界情况处理"},
    ],
    "creative_writing": [
        {"name": "creativity", "weight": 0.30, "description": "创意性"},
        {"name": "coherence", "weight": 0.25, "description": "连贯性"},
        {"name": "language_quality", "weight": 0.25, "description": "语言质量"},
        {"name": "engagement", "weight": 0.20, "description": "吸引力"},
    ]
}

import json
from dataclasses import dataclass
from typing import Optional

@dataclass
class EvalResult:
    dimensions: dict
    overall_score: float
    reasoning: str
    confidence: float

class LLMJudgePipeline:
    def __init__(self, judge_model_client, dimensions: list):
        self.judge = judge_model_client
        self.dimensions = dimensions
    
    async def evaluate(self, question: str, answer: str,
                       reference: Optional[str] = None) -> EvalResult:
        prompt = self._build_prompt(question, answer, reference)
        
        response = await self.judge.chat(
            messages=[{"role": "user", "content": prompt}],
            temperature=0.1,  # 低温度保证一致性
            response_format={"type": "json_object"}
        )
        
        result = json.loads(response.content)
        
        # 计算加权总分
        overall = sum(
            result[d["name"]]["score"] * d["weight"]
            for d in self.dimensions
        )
        
        return EvalResult(
            dimensions=result,
            overall_score=overall,
            reasoning=result.get("overall", {}).get("reason", ""),
            confidence=result.get("confidence", 0.8)
        )
    
    def _build_prompt(self, question, answer, reference=None):
        dims_desc = "\n".join(
            f"{i+1}. **{d['name']}** ({d['description']}) - 权重: {d['weight']}"
            for i, d in enumerate(self.dimensions)
        )
        
        prompt = f"""你是一个专业的评估专家。请严格按以下维度评分。

## 问题
{question}

## 待评估回答
{answer}
"""
        if reference:
            prompt += f"\n## 参考答案\n{reference}\n"
        
        prompt += f"""
## 评分维度（1-10分）
{dims_desc}

## 输出要求
以 JSON 输出，每个维度包含 score 和 reason 字段。
"""
        return prompt

async def unbiased_pairwise_compare(judge, question, answer_a, answer_b):
    # 正向比较
    result_1 = await judge.compare(question, answer_a, answer_b)
    
    # 反向比较（交换位置）
    result_2 = await judge.compare(question, answer_b, answer_a)
    
    # 综合判断
    if result_1.winner == "A" and result_2.winner == "B":
        return "A"  # 两次都认为第一个更好 → A 确实更好
    elif result_1.winner == "B" and result_2.winner == "A":
        return "B"  # 两次都认为第二个更好 → B 确实更好
    else:
        return "tie"  # 不一致 → 平局

def calculate_agreement(scores_a: list, scores_b: list) -> dict:
    """计算两组评分的一致性指标"""
    import numpy as np
    from scipy import stats
    
    scores_a = np.array(scores_a)
    scores_b = np.array(scores_b)
    
    return {
        "pearson": stats.pearsonr(scores_a, scores_b)[0],
        "spearman": stats.spearmanr(scores_a, scores_b)[0],
        "kappa": calculate_cohen_kappa(scores_a, scores_b),
        "mae": np.mean(np.abs(scores_a - scores_b))
    }