ag2-evaluation

用途

AG2 Evaluation 用于对 AG2 Agent 进行离线评估、测试与基准测试，也可用于构建回归测试或 CI 门禁。典型场景包括：对回答的正确性、工具使用、成本或主观质量进行打分；跨版本追踪某项指标，判断某次改动是改进还是回归。若需要两个及以上构建版本进行头对头比较或排行榜排名，应使用 ag2-eval-comparison。

安装

pip install "ag2[openai,tracing]"

run_agent 会从 OpenTelemetry span 重建每个任务的轨迹，因此必须安装 tracing 扩展。请在交付代码前执行该安装；若无法运行命令，需明确给出上述 pip install 命令。

核心循环：数据集、Agent、scorer、run_agent

import asyncio
from ag2 import Agent
from ag2.config import OpenAIConfig
from ag2.eval import Suite, run_agent
from ag2.eval.scorers import final_answer_matches

suite = Suite.from_list([
    {"task_id": "france", "inputs": {"input": "Capital of France?"}, "reference_outputs": {"answer": "Paris"}},
    {"task_id": "japan",  "inputs": {"input": "Capital of Japan?"},  "reference_outputs": {"answer": "Tokyo"}},
])
agent = Agent("geographer", prompt="Answer with the capital city.", config=OpenAIConfig(model="gpt-4o-mini"))

async def main():
    result = await run_agent(
        suite, agent=agent,
        scorers=[final_answer_matches(field="answer", matcher="contains")],
        store_dir="./runs",
    )
    print(result.summary())                            # the scorecard
    print(result.pass_rate("final_answer_matches"))    # 1.0

asyncio.run(main())

其中 inputs["input"] 是提示词；reference_outputs 是标准答案（一个 dict，若只做轨迹检查可省略）。每个 scorer 相当于一列，通过其 key 查找。

Scorer

每个 scorer 只问一个问题，其返回类型决定聚合方式：

返回 bool：聚合为通过率，通过 result.pass_rate(key) 访问。
返回 int / float：聚合为均值 / p50 / p95，通过 result.score_stats(key) 访问。
返回 str：聚合为值计数，通过 result.value_counts(key) 访问。

预置 scorer（ag2.eval.scorers）包括：final_answer_matches(field=, matcher="contains"|"casefold"|"exact")、tool_called(name)、no_tool_errors()、token_budget(n)、failure_attribution(...)、agent_judge(...)。

自定义 scorer 通过装饰器声明所需参数（outputs、trace、reference_outputs、inputs、task）：

from ag2.eval import scorer

@scorer
def answered_briefly(outputs) -> bool:
    return len(outputs["body"]) < 100      # outputs["body"] = final answer text

agent_judge 用于无法用 == 判断的质量评估（应使用与被测 Agent 不同的模型）：

from ag2.eval.scorers import agent_judge
judge = agent_judge(OpenAIConfig(model="gpt-4o"), criterion="Helpful and accurate.", key="quality")

CI：确定性、无需 API key

将模型替换为 TestConfig cassette（每个任务一个预设回复），使 CI 免费且可重复。model_config 是 dict[task_id, ModelConfig]，每个任务一个 cassette，并会覆盖该任务上 Agent 自身的配置：

from ag2.testing import TestConfig

agent = Agent("geographer", prompt="Answer with the capital city.")   # an Agent instance, not a factory

canned = {"france": TestConfig("Paris"), "japan": TestConfig("Tokyo")}
result = await run_agent(suite, agent=agent, scorers=scorers, model_config=canned, store_dir="./runs")
assert result.pass_rate("final_answer_matches") == 1.0      # the gate

持久化、追踪与对已有轨迹打分

store_dir= 会为每次运行写入一个 JSON。可重新加载过去的运行并做 diff 以检测回归；也可对已有轨迹（例如生产遥测数据）直接打分，而无需重新运行 Agent：

from ag2.eval import load_run, evaluate_traces, DirectoryTraceSource

assert not result.diff(load_run("./runs/<run_id>.json")).regressions   # scorers that flipped pass -> fail
graded = await evaluate_traces(DirectoryTraceSource("./traces"), scorers=scorers, store_dir="./runs")

常见陷阱

缺少 tracing 扩展：run_agent 无法重建轨迹。需安装 ag2[<provider>,tracing]。
返回类型与聚合方式不匹配：bool 用于通过/失败，数字用于统计，str 用于分类；结果需按 scorer 的 key 查找。
同一模型既作答又评判：会使 agent_judge 产生偏差；应使用不同的评判模型。

深入了解

website/docs/user-guide/evaluation/：入门、scorer（目录 + 自定义 + 返回类型规则）、运行、持久化。
ag2-eval-comparison：排行榜（run_variants）与头对头比较（run_pairwise）。