ag2-evaluation

Compare original and translation side by side

🇺🇸

Original

English
🇨🇳

Translation

Chinese

Evaluation — run, grade, and track an agent

评估——运行、评分并追踪Agent

When to use

适用场景

  • Evaluate / test / benchmark an AG2
    Agent
    , or build a regression / CI gate
  • Grade answers for correctness, tool use, cost, or subjective quality
  • Track a metric across versions (did this change help or regress?)
To compare two-plus builds head-to-head or on a leaderboard, use
ag2-eval-comparison
.
  • 评估/测试/基准测试AG2
    Agent
    ,或构建回归/CI准入门槛
  • 针对正确性、工具使用、成本或主观质量对答案进行评分
  • 跨版本追踪指标(此次变更带来了提升还是导致了退化?)
如需对两个及以上版本进行直接对比或查看排行榜,请使用
ag2-eval-comparison

Install

安装

bash
pip install "ag2[openai,tracing]"
run_agent
reconstructs each task's trace from OpenTelemetry spans, so the
tracing
extra is required. Run this install before delivering the code. If you cannot run commands, state the exact
pip install
command.
bash
pip install "ag2[openai,tracing]"
run_agent
会从OpenTelemetry追踪数据中重建每个任务的追踪记录,因此必须安装
tracing
扩展包。在交付代码前请执行此安装命令。若无法运行命令,请明确说明具体的
pip install
指令。

The loop — dataset, agent, scorers, run_agent

流程——数据集、Agent、评分器、run_agent

python
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"]
is the prompt;
reference_outputs
is the gold answer (a dict — omit it for trace-only checks). Each scorer is a column, looked up by its key.
python
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())                            # 计分卡
    print(result.pass_rate("final_answer_matches"))    # 1.0

asyncio.run(main())
inputs["input"]
为提示词;
reference_outputs
为标准答案(字典格式——若仅需检查追踪记录可省略)。每个评分器对应一列数据,可通过其键值查找。

Scorers

评分器

A scorer asks ONE question. Its RETURN TYPE picks the aggregation:
returnaggregationaccessor
bool
pass rate
result.pass_rate(key)
int
/
float
mean / p50 / p95
result.score_stats(key)
str
value counts
result.value_counts(key)
Prebuilt (
ag2.eval.scorers
):
final_answer_matches(field=, matcher="contains"|"casefold"|"exact")
,
tool_called(name)
,
no_tool_errors()
,
token_budget(n)
,
failure_attribution(...)
,
agent_judge(...)
.
Custom — decorate a function that declares what it needs by name (
outputs
,
trace
,
reference_outputs
,
inputs
,
task
):
python
from ag2.eval import scorer

@scorer
def answered_briefly(outputs) -> bool:
    return len(outputs["body"]) < 100      # outputs["body"] = final answer text
agent_judge
grades quality you can't check with
==
(use a different model than the agent under test):
python
from ag2.eval.scorers import agent_judge
judge = agent_judge(OpenAIConfig(model="gpt-4o"), criterion="Helpful and accurate.", key="quality")
每个评分器仅针对一个问题。其返回类型决定了聚合方式:
返回类型聚合方式访问方法
bool
通过率
result.pass_rate(key)
int
/
float
平均值/中位数/95分位数
result.score_stats(key)
str
数值统计
result.value_counts(key)
预构建评分器(
ag2.eval.scorers
):
final_answer_matches(field=, matcher="contains"|"casefold"|"exact")
tool_called(name)
no_tool_errors()
token_budget(n)
failure_attribution(...)
agent_judge(...)
自定义评分器——通过装饰器定义函数,并声明所需参数(
outputs
trace
reference_outputs
inputs
task
):
python
from ag2.eval import scorer

@scorer
def answered_briefly(outputs) -> bool:
    return len(outputs["body"]) < 100      # outputs["body"] = 最终答案文本
agent_judge
用于评估无法通过
==
校验的质量(请使用与被测Agent不同的模型):
python
from ag2.eval.scorers import agent_judge
judge = agent_judge(OpenAIConfig(model="gpt-4o"), criterion="Helpful and accurate.", key="quality")

CI — deterministic, no API key

CI集成——确定性、无需API密钥

Swap the model for a
TestConfig
cassette (a canned reply per task) so CI is free and repeatable.
model_config
is a
dict[task_id, ModelConfig]
— one cassette per task — and overrides the agent's own config for that task:
python
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
将模型替换为
TestConfig
测试用例(每个任务对应一个预设回复),使CI流程免费且可重复。
model_config
dict[task_id, ModelConfig]
——每个任务对应一个测试用例——会覆盖Agent针对该任务的自有配置:
python
from ag2.testing import TestConfig

agent = Agent("geographer", prompt="Answer with the capital city.")   # Agent实例,而非工厂类

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      # 准入校验

Persist, track, grade existing traces

持久化、追踪、评估现有追踪数据

store_dir=
writes one JSON per run. Reload a past run and diff for regressions; or grade traces you already have (e.g. production telemetry) without re-running the agent:
python
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")
store_dir=
参数会为每次运行生成一个JSON文件。可重新加载过往运行结果并对比以发现回归问题;也可对已有的追踪数据(如生产环境遥测数据)进行评分,无需重新运行Agent:
python
from ag2.eval import load_run, evaluate_traces, DirectoryTraceSource

assert not result.diff(load_run("./runs/<run_id>.json")).regressions   # 从通过变为失败的评分器
graded = await evaluate_traces(DirectoryTraceSource("./traces"), scorers=scorers, store_dir="./runs")

Common pitfalls

常见陷阱

  • Missing
    tracing
    extra
    run_agent
    can't reconstruct traces. Install
    ag2[<provider>,tracing]
    .
  • Return type vs aggregation
    bool
    for pass/fail, a number for stats, a
    str
    for categories; look results up by the scorer's
    key
    .
  • Same model answers and judges — biases
    agent_judge
    ; use a different judge model.
  • 缺少
    tracing
    扩展包
    ——
    run_agent
    无法重建追踪记录。请安装
    ag2[<provider>,tracing]
  • 返回类型与聚合方式不匹配——
    bool
    用于通过/失败判断,数值类型用于统计,
    str
    用于分类;需通过评分器的
    key
    查找结果。
  • 评判与Agent使用同一模型——会导致
    agent_judge
    产生偏差;请使用不同的评判模型。

Going deeper

深入学习

  • website/docs/user-guide/evaluation/
    getting-started
    ,
    scorers
    (catalog + custom + return-type rules),
    runs
    ,
    persistence
  • ag2-eval-comparison
    — leaderboard (
    run_variants
    ) + head-to-head (
    run_pairwise
    )
  • website/docs/user-guide/evaluation/
    ——入门指南、评分器(目录+自定义+返回类型规则)、运行记录、持久化
  • ag2-eval-comparison
    ——排行榜(
    run_variants
    )+ 直接对比(
    run_pairwise