Back to Details

ag2-testing

Compare original and translation side by side

🇺🇸

Original

English
🇨🇳

Translation

Chinese

Testing agents and tools

测试agents与工具

When to use

适用场景

Writing tests for code that builds AG2 
Agent
s, custom 
@tool
functions, middleware, or response schemas — anywhere you don't want to make real LLM API calls.
为构建AG2 
Agent
、自定义
@tool
函数、中间件或响应模式的代码编写测试——适用于任何不想调用真实LLM API的场景。

60-second recipe — mock an LLM response

60秒快速上手——模拟LLM响应

python
import pytest
from ag2 import Agent
from ag2.testing import TestConfig

@pytest.mark.asyncio
async def test_mocked_response():
    agent = Agent("test_agent")
    reply = await agent.ask("Hi!", config=TestConfig("This is a mocked response."))
    assert reply.body == "This is a mocked response."
TestConfig(*responses)
replaces the model client. Each positional arg is the mocked response for the next LLM call within an 
ask()
 — strings for text replies, 
ToolCallEvent
for tool dispatches. (The cursor is per-
ask()
; see "Multi-turn mock" below for what that means across multiple turns.)
python
import pytest
from ag2 import Agent
from ag2.testing import TestConfig

@pytest.mark.asyncio
async def test_mocked_response():
    agent = Agent("test_agent")
    reply = await agent.ask("Hi!", config=TestConfig("This is a mocked response."))
    assert reply.body == "This is a mocked response."
TestConfig(*responses)
会替换模型客户端。每个位置参数对应一次
ask()
内下一次LLM调用的模拟响应——字符串对应文本回复，
ToolCallEvent
对应工具调度。（游标是按每个
ask()
独立的；下文的“多轮模拟”会解释跨多轮调用的含义。）

Simulate a successful tool call

模拟成功的工具调用

Pass a 
ToolCallEvent
first (the model "decides" to call the tool), then the final answer:
python
import pytest
from ag2 import Agent
from ag2.events import ToolCallEvent
from ag2.testing import TestConfig

@pytest.mark.asyncio
async def test_tool_success():
    def my_tool() -> str:
        return "tool execution result"

    agent = Agent("test_agent", tools=[my_tool])
    config = TestConfig(
        ToolCallEvent(name="my_tool"),
        "final result",
    )
    reply = await agent.ask("Please use my_tool", config=config)
    assert reply.body == "final result"
先传入
ToolCallEvent
（模型“决定”调用工具），再传入最终答案：
python
import pytest
from ag2 import Agent
from ag2.events import ToolCallEvent
from ag2.testing import TestConfig

@pytest.mark.asyncio
async def test_tool_success():
    def my_tool() -> str:
        return "tool execution result"

    agent = Agent("test_agent", tools=[my_tool])
    config = TestConfig(
        ToolCallEvent(name="my_tool"),
        "final result",
    )
    reply = await agent.ask("Please use my_tool", config=config)
    assert reply.body == "final result"

Test tool error paths

测试工具错误流程

If a tool raises, the exception propagates to 
ask()
:
python
@pytest.mark.asyncio
async def test_tool_raises():
    def failing_tool() -> str:
        raise ValueError("Something went wrong")

    config = TestConfig(
        ToolCallEvent(name="failing_tool"),
        "result",
    )
    agent = Agent("test_agent", config=config, tools=[failing_tool])

    with pytest.raises(ValueError, match="Something went wrong"):
        await agent.ask("Hi!")
如果工具抛出异常，该异常会传播到
ask()
：
python
@pytest.mark.asyncio
async def test_tool_raises():
    def failing_tool() -> str:
        raise ValueError("Something went wrong")

    config = TestConfig(
        ToolCallEvent(name="failing_tool"),
        "result",
    )
    agent = Agent("test_agent", config=config, tools=[failing_tool])

    with pytest.raises(ValueError, match="Something went wrong"):
        await agent.ask("Hi!")

Tool not found

工具未找到

If the LLM calls a tool the agent doesn't have, the framework raises 
ToolNotFoundError
:
python
from ag2.exceptions import ToolNotFoundError

@pytest.mark.asyncio
async def test_tool_not_found():
    config = TestConfig(ToolCallEvent(name="unregistered_tool"))
    agent = Agent("test_agent", config=config)
    with pytest.raises(ToolNotFoundError, match="Tool `unregistered_tool` not found"):
        await agent.ask("Hi!")
如果LLM调用了agent未注册的工具，框架会抛出
ToolNotFoundError
：
python
from ag2.exceptions import ToolNotFoundError

@pytest.mark.asyncio
async def test_tool_not_found():
    config = TestConfig(ToolCallEvent(name="unregistered_tool"))
    agent = Agent("test_agent", config=config)
    with pytest.raises(ToolNotFoundError, match="Tool `unregistered_tool` not found"):
        await agent.ask("Hi!")

Useful test patterns

实用测试模式

Override 
Depends
dependencies

覆盖
Depends
依赖

python
def get_production_db():
    raise Exception("Do not call in tests!")

@tool
def read_data(db: Annotated[object, Depends(get_production_db)]) -> str:
    return "Data"

agent = Agent("test", tools=[read_data])
agent.dependency_provider.override(get_production_db, lambda: "mock_db")
python
def get_production_db():
    raise Exception("Do not call in tests!")

@tool
def read_data(db: Annotated[object, Depends(get_production_db)]) -> str:
    return "Data"

agent = Agent("test", tools=[read_data])
agent.dependency_provider.override(get_production_db, lambda: "mock_db")

Override 
Inject
dependencies

覆盖
Inject
依赖

Just pass 
dependencies={...}
to 
agent.ask(...)
:
python
await agent.ask("Read", dependencies={"database_pool": fake_pool})
只需在
agent.ask(...)
中传入
dependencies={...}
：
python
await agent.ask("Read", dependencies={"database_pool": fake_pool})

Capture stream events

捕获流事件

python
from ag2 import MemoryStream
from ag2.events import ToolCallEvent

stream = MemoryStream()
collected: list[ToolCallEvent] = []
stream.where(ToolCallEvent).subscribe(lambda e: collected.append(e))

await agent.ask("Test", stream=stream)
assert collected[0].name == "expected_tool"
python
from ag2 import MemoryStream
from ag2.events import ToolCallEvent

stream = MemoryStream()
collected: list[ToolCallEvent] = []
stream.where(ToolCallEvent).subscribe(lambda e: collected.append(e))

await agent.ask("Test", stream=stream)
assert collected[0].name == "expected_tool"

Multi-turn mock — the response list is per-
ask()
, not per-conversation

多轮模拟——响应列表按每个
ask()
划分，而非按对话

TestConfig(...)
's response list is consumed within a single 
ask()
, across that round's repeated LLM calls — that's why 
TestConfig(ToolCallEvent("my_tool"), "final result")
works for a tool-using turn (the LLM emits the tool call, the tool runs, the LLM is called again and gets 
"final result"
). Internally, 
TestConfig.create()
hands back a fresh client whose iterator starts at 
responses[0]
, and that's done once per 
ask()
 — so every new 
ask()
(a 
reply.ask(...)
chain, or each turn the network adapters / an auto-replying agent drive) restarts the cursor at the first response. Listing more responses does not let you say "conversational turn 2 differs from turn 1".
For variation across multiple 
ask()
calls, either:
  • pass a fresh 
    TestConfig(...)
    per turn via the per-
    ask()
    config=
    override (
    await agent.ask("…", config=TestConfig("turn-2 reply"))
    ), or
  • mock the model with a 
    ToolCallEvent
    and put the per-turn logic in a stateful tool — a closure or class instance that tracks how many times it's been called and returns accordingly. (Useful when something other than your test code drives the turn loop — e.g. a 
    workflow
    / 
    discussion
    channel — so you can't inject a per-turn 
    config=
    .)
TestConfig(...)
的响应列表会在单次
ask()
内被消耗，覆盖该轮次中所有重复的LLM调用——这也是
TestConfig(ToolCallEvent("my_tool"), "final result")
能在工具调用轮次中生效的原因（LLM发出工具调用，工具执行，LLM再次被调用并获得
"final result"
）。内部实现上，
TestConfig.create()
会返回一个全新的客户端，其迭代器从
responses[0]
开始，且该操作每个
ask()
执行一次——因此每次新的
ask()
（比如
reply.ask(...)
链式调用，或网络适配器/自动回复agent驱动的每个轮次）都会将游标重置到第一个响应。列出更多响应并不能实现“对话第二轮与第一轮不同”的效果。
若要在多个
ask()
调用中实现不同的响应，可选择以下两种方式：
  • 通过每个
    ask()
    的
    config=
    参数覆盖，传入全新的
    TestConfig(...)
    （
    await agent.ask("…", config=TestConfig("turn-2 reply"))
    ），或
  • 用
    ToolCallEvent
    模拟模型，并将每轮逻辑放入有状态的工具中——比如一个闭包或类实例，用于跟踪调用次数并返回相应结果。（当测试代码之外的其他组件驱动轮次循环时非常有用——例如
    workflow
    /
    discussion
    通道，此时无法注入每轮的
    config=
    。）

Going deeper

深入了解

  • Source doc: 
    website/docs/user-guide/testing.mdx
    .
  • Test markers / async config — repo 
    pyproject.toml
    . Use 
    @pytest.mark.asyncio
    (the project uses pytest-asyncio).
  • Streams (for asserting events): 
    website/docs/user-guide/advanced/stream.mdx
    .
  • 源文档：
    website/docs/user-guide/testing.mdx
    。
  • 测试标记/异步配置——仓库
    pyproject.toml
    。使用
    @pytest.mark.asyncio
    （项目依赖pytest-asyncio）。
  • 流（用于断言事件）：
    website/docs/user-guide/advanced/stream.mdx
    。

Common pitfalls

常见陷阱

  • Forgetting 
    @pytest.mark.asyncio
     — the test will skip or fail oddly.
  • Mismatched response count — 
    TestConfig
    runs out of responses if the agent makes more LLM calls than you expect within one 
    ask()
     (e.g. tool error → another LLM call); 
    StopIteration
    propagates. Add more positional args or assert the call sequence. (The cursor is per-
    ask()
    , so the count you need is "LLM calls in one round", not "turns in the conversation" — see Multi-turn mock above.)
  • Mocking the LLM but not the tool — your tool function still runs (and may hit real APIs / disk). Mock the tool if you're isolating LLM behaviour, or override its 
    Depends
    to inject test doubles.
  • Asserting on 
    reply.body
    when you set a 
    response_schema
     — 
    body
    is the raw text. Use 
    await reply.content()
    for the validated value.
  • Sharing 
    Agent
    instances across async tests     — agents carry mutable state (variables, dependencies). Construct fresh agents per test for isolation.
  • Using real provider clients in CI — wrap the provider config with 
    TestConfig
    per-test or via a fixture; never rely on 
    OPENAI_API_KEY
    etc. being available in test environments.
  • 忘记添加
    @pytest.mark.asyncio
    ——测试会被跳过或出现异常失败。
  • 响应数量不匹配——如果agent在单次
    ask()
    内发起的LLM调用次数超出预期（例如工具出错→再次调用LLM），
    TestConfig
    会耗尽响应，
    StopIteration
    异常会传播。需添加更多位置参数或断言调用序列。（游标按每个
    ask()
    独立，因此所需的响应数量是“一轮中的LLM调用次数”，而非“对话中的轮次”——详见上文的多轮模拟。）
  • 仅模拟LLM但未模拟工具——工具函数仍会执行（可能调用真实API/访问磁盘）。若要隔离LLM行为，需模拟工具，或覆盖其
    Depends
    依赖以注入测试替身。
  • 设置
    response_schema
    后断言
    reply.body
    ——
    body
    是原始文本。需使用
    await reply.content()
    获取验证后的值。
  • 在异步测试间共享
    Agent
    实例    ——agents包含可变状态（变量、依赖）。为保证隔离性，需为每个测试创建全新的agent实例。
  • 在CI中使用真实服务商客户端——需为每个测试或通过fixture用
    TestConfig
    包装服务商配置；切勿依赖测试环境中存在
    OPENAI_API_KEY
    等密钥。