如何评测一个 Agent：不只是看它回答得好不好

from dataclasses import dataclass, field
from typing import Optional, Callable

@dataclass
class TestCase:
    id: str
    description: str
    input: str                          # 用户输入
    expected_outcome: str               # 预期结果的文字描述
    success_criteria: list[Callable]    # 可执行的成功判断函数
    tags: list[str]                     # 分类标签，如 ["tool_failure", "multi_step"]
    difficulty: str                     # "easy" | "medium" | "hard"
    required_tools: list[str]           # 这个任务理论上需要用到哪些工具
    max_steps: int = 15                 # 允许的最大执行步数
    timeout_seconds: int = 60          # 超时时间
    notes: Optional[str] = None        # 测试设计说明


@dataclass
class TestResult:
    case_id: str
    success: bool
    completion_rate: float              # 0.0 - 1.0，子任务完成比例
    steps_taken: int
    tools_called: list[str]
    error_recoveries: int               # 遇到工具失败后成功恢复的次数
    final_output: str
    execution_trace: list[dict]         # 完整执行轨迹
    latency_seconds: float
    total_tokens: int
    failure_reason: Optional[str] = None

def make_criteria():
    return [
        # 结果中包含具体数字
        lambda output: any(char.isdigit() for char in output),
        # 结果长度合理（不是空的，也不是截断的）
        lambda output: 100 < len(output) < 5000,
        # 包含特定关键词（根据任务定制）
        lambda output: "分析" in output and "建议" in output,
    ]

import asyncio
import statistics
from typing import Any

async def run_evaluation(
    agent_fn: Callable,
    test_cases: list[TestCase],
    repeat: int = 3              # 每个用例重复运行次数，用于测稳定性
) -> dict[str, Any]:

    results: dict[str, list[TestResult]] = {}

    for case in test_cases:
        case_results = []
        for run_idx in range(repeat):
            try:
                result = await run_single_case(agent_fn, case)
                case_results.append(result)
            except Exception as e:
                case_results.append(TestResult(
                    case_id=case.id,
                    success=False,
                    completion_rate=0.0,
                    steps_taken=0,
                    tools_called=[],
                    error_recoveries=0,
                    final_output="",
                    execution_trace=[],
                    latency_seconds=0,
                    total_tokens=0,
                    failure_reason=str(e)
                ))
        results[case.id] = case_results

    return compute_metrics(results, test_cases)


def compute_metrics(
    results: dict[str, list[TestResult]],
    test_cases: list[TestCase]
) -> dict[str, Any]:

    all_success_rates = []
    stability_scores  = []
    recovery_rates    = []
    step_efficiencies = []

    for case in test_cases:
        case_results = results[case.id]
        successes = [r.success for r in case_results]

        # 任务完成率
        success_rate = sum(successes) / len(successes)
        all_success_rates.append(success_rate)

        # 稳定性：成功率的一致程度（1.0 = 完全稳定，0.0 = 完全随机）
        # 稳定失败（0.0）和稳定成功（1.0）都是高稳定性
        stability = 1.0 - (2 * abs(success_rate - round(success_rate)))
        stability_scores.append(stability)

        # 错误恢复率：遇到工具失败时成功恢复的比例
        tool_failures = sum(
            1 for r in case_results
            for step in r.execution_trace
            if step.get("tool_failed")
        )
        recoveries = sum(r.error_recoveries for r in case_results)
        if tool_failures > 0:
            recovery_rates.append(recoveries / tool_failures)

        # 步骤效率：实际步骤数 / 理论最少步骤数
        for r in case_results:
            if r.success and r.steps_taken > 0:
                efficiency = len(case.required_tools) / r.steps_taken
                step_efficiencies.append(min(efficiency, 1.0))

    return {
        "task_success_rate":    statistics.mean(all_success_rates),
        "stability_score":      statistics.mean(stability_scores),
        "error_recovery_rate":  statistics.mean(recovery_rates) if recovery_rates else None,
        "step_efficiency":      statistics.mean(step_efficiencies) if step_efficiencies else None,
        "by_difficulty": _breakdown_by_tag(results, test_cases, "difficulty"),
        "by_tag":        _breakdown_by_tag(results, test_cases, "tags"),
    }

async def judge_execution_trace(
    task: str,
    trace: list[dict],
    final_output: str,
    judge_model: str = "gpt-4o"
) -> dict[str, float]:

    trace_text = "\n".join([
        f"[{step['type']}] {step['content'][:300]}"
        for step in trace
    ])

    response = await client.chat.completions.create(
        model=judge_model,
        messages=[{
            "role": "user",
            "content": f"""
你是一个 Agent 执行质量评审员。请评估以下 Agent 的执行过程。

任务：{task}

执行轨迹：
{trace_text}

最终输出：
{final_output}

请对以下维度打分（0.0 - 1.0），并给出简短理由：
1. 推理连贯性：每步 Thought 是否有依据、逻辑是否自洽
2. 工具使用合理性：工具调用是否必要、参数是否准确
3. 错误处理质量：遇到工具失败或意外结果时处理是否合理
4. 输出质量：最终输出是否完整、准确、有价值

只输出 JSON，格式：
{{
  "reasoning_coherence": 0.0-1.0,
  "tool_usage_quality":  0.0-1.0,
  "error_handling":      0.0-1.0,
  "output_quality":      0.0-1.0,
  "reasoning_notes":     "简短说明"
}}
"""
        }],
        response_format={"type": "json_object"}
    )

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

FAILURE_CATEGORIES = {
    "tool_selection":    "选错了工具，或该调用时没有调用",
    "tool_parameter":    "工具参数填写错误",
    "no_recovery":       "工具失败后没有恢复，继续在错误前提上执行",
    "goal_drift":        "执行过程中偏离了原始目标",
    "premature_stop":    "信息不足时过早给出答案",
    "over_execution":    "信息已足够但继续调用工具",
    "context_loss":      "在长任务中忘记了之前的关键信息",
    "format_error":      "输出格式不符合要求",
    "hallucination":     "引用了工具未返回的信息",
}