GEPA 评估流程详解

GEPA 评估流程详解

评估是 GEPA 优化过程中的核心步骤，类似于神经网络的前向传播。

评估流程图

┌─────────────────────────────────────────────────────────────────────────┐
│                         GEPA 评估 (前向传播)                             │
├─────────────────────────────────────────────────────────────────────────┤
│                                                                         │
│  输入:                                                                  │
│    - batch: [样本1, 样本2, ...]                                         │
│    - candidate: {"prompt": "You are a helpful assistant..."}            │
│                                                                         │
│  ┌─────────────────────────────────────────────────────────────────┐   │
│  │                    Step 1: 构建消息                              │   │
│  │  for 每个样本:                                                   │   │
│  │    messages = [                                                 │   │
│  │      {"role": "system", "content": candidate["prompt"]},        │   │
│  │      {"role": "user", "content": data["input"]}                 │   │
│  │    ]                                                             │   │
│  └─────────────────────────────────────────────────────────────────┘   │
│                              ↓                                          │
│  ┌─────────────────────────────────────────────────────────────────┐   │
│  │              Step 2: 调用 LLM (前向传播) ✅                      │   │
│  │  responses = lm.batch_complete(                                 │   │
│  │      messages,          # 批量消息                              │   │
│  │      max_workers=10     # 并发调用                              │   │
│  │  )                                                              │   │
│  │                                                                  │   │
│  │  实际调用: litellm.completion()                                  │   │
│  │  API: OpenAI/Claude/其他                                         │   │
│  └─────────────────────────────────────────────────────────────────┘   │
│                              ↓                                          │
│  ┌─────────────────────────────────────────────────────────────────┐   │
│  │           Step 3: 计算分数 (评估器) ❌ 不调用 LLM                │   │
│  │  for 每个 LLM 响应:                                             │   │
│  │    eval_result = evaluator(data, response)                     │   │
│  │    score = eval_result.score      # 例如: 0.0 或 1.0            │   │
│  │    feedback = eval_result.feedback  # 文本反馈                  │   │
│  │                                                                  │   │
│  │  评估器类型:                                                      │   │
│  │  • 字符串匹配 (默认)                                              │   │
│  │  • 正则表达式                                                     │   │
│  │  • 数值比较                                                       │   │
│  │  • 自定义规则                                                     │   │
│  └─────────────────────────────────────────────────────────────────┘   │
│                              ↓                                          │
│  输出: EvaluationBatch                                                 │
│    - outputs: LLM 生成的文本                                          │
│    - scores: 每个样本的分数                                           │
│    - trajectories: 用于反思的数据                                     │
│                                                                         │
└─────────────────────────────────────────────────────────────────────────┘

代码实现

DefaultAdapter 评估

# src/gepa/adapters/default_adapter/default_adapter.py
def evaluate(
    self,
    batch: list[DefaultDataInst],
    candidate: dict[str, str],
    capture_traces: bool = False,
) -> EvaluationBatch:
    outputs = []
    scores = []
    trajectories = [] if capture_traces else None

    # Step 1: 构建消息
    system_content = candidate["system_prompt"]
    litellm_requests = []

    for data in batch:
        messages = [
            {"role": "system", "content": system_content},
            {"role": "user", "content": data["input"]}
        ]
        litellm_requests.append(messages)

    # Step 2: 调用 LLM ✅
    responses = self._lm.batch_complete(
        litellm_requests,
        max_workers=10
    )

    # Step 3: 计算分数 ❌
    for data, response in zip(batch, responses):
        eval_result = self.evaluator(data, response)
        score = eval_result.score
        feedback = eval_result.feedback

        outputs.append({"full_assistant_response": response})
        scores.append(score)

        if trajectories is not None:
            trajectories.append({
                "data": data,
                "full_assistant_response": response,
                "feedback": feedback
            })

    return EvaluationBatch(
        outputs=outputs,
        scores=scores,
        trajectories=trajectories
    )

评估器详解

默认评估器: ContainsAnswerEvaluator

class ContainsAnswerEvaluator:
    """默认评估器：检查答案是否在响应中"""

    def __call__(self, data, response):
        # 字符串匹配 - 不调用 LLM！
        is_correct = data["answer"] in response
        score = 1.0 if is_correct else 0.0

        # 生成反馈 (文本模板)
        if is_correct:
            feedback = f"Correct. Includes '{data['answer']}'"
        else:
            feedback = f"Wrong. Answer is '{data['answer']}'"

        return EvaluationResult(score=score, feedback=feedback)

重要：评估器通常不调用 LLM，而是使用规则判断。

自定义评估器

def my_evaluator(data, response):
    """自定义评估逻辑"""
    import re

    # 提取数字
    numbers = re.findall(r'\d+', response)

    if not numbers:
        return 0.0, "No number found"

    predicted = int(numbers[0])
    expected = int(data["answer"])

    # 计算误差
    error = abs(predicted - expected)
    score = max(0.0, 1.0 - error / 10.0)

    feedback = f"Predicted {predicted}, expected {expected}"
    return score, feedback

置信度评估器

class ConfidenceAdapter:
    """使用 LLM 的 logprob 信息"""

    def evaluate(self, batch, candidate):
        # 调用 LLM，请求 logprobs
        responses = self._lm.batch_complete(
            messages,
            logprobs=True,
            top_logprobs=5
        )

        # 提取置信度 (不需要额外 LLM 调用)
        for response in responses:
            prob = extract_probability(response)
            score = correctness * confidence

类比神经网络

实际例子

# 输入数据
batch = [
    {"input": "1+1=?", "answer": "2"},
    {"input": "2+2=?", "answer": "4"},
    {"input": "3+3=?", "answer": "6"}
]

# 当前 prompt
candidate = {"system_prompt": "You are a math assistant."}

# 评估
eval_batch = adapter.evaluate(batch, candidate, capture_traces=True)

# 内部发生的事情:
# 1. 构建消息
messages_list = [
    [{"role": "system", "content": "..."}, {"role": "user", "content": "1+1=?"}],
    [{"role": "system", "content": "..."}, {"role": "user", "content": "2+2=?"}],
    [{"role": "system", "content": "..."}, {"role": "user", "content": "3+3=?"}]
]

# 2. 调用 LLM ✅
responses = llm.batch_complete(messages_list)
# => ["The answer is 2", "The answer is 4", "The answer is 6"]

# 3. 计算分数 ❌
scores = [1.0, 1.0, 1.0]  # 全部正确

关键要点

1. 前向传播 = LLM 调用 ✅
   • 通过 lm.batch_complete() 调用 LLM API
   • 并发处理多个样本
2. 分数计算 = 规则判断 ❌
   • 评估器通常不调用 LLM
   • 使用字符串匹配、正则表达式等规则
3. 反馈生成 = 文本模板 ❌
   • 基于规则生成反馈文本
   • 用于后续的 LLM 反思
4. 分数越高越好
   • 与神经网络的 loss 相反
   • 可以理解为 -loss 或 accuracy

下一步

• 反思流程详解 - 了解如何使用评估结果
• 与神经网络对比 - 深入理解类比