ag2-testing

测试 Agent 和工具

适用场景

为构建 AG2 Agent、自定义 @tool 函数、中间件或响应模式的代码编写测试——适用于任何不想调用真实 LLM API 的场景。

60 秒快速入门 — 模拟 LLM 响应

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() 独立计算；关于它在多轮对话中的含义，请参见下文的“多轮模拟”。）

模拟成功的工具调用

先传入 ToolCallEvent（模型“决定”调用工具），再传入最终答案：

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"

测试工具错误路径

如果工具引发异常，该异常会传播到 ask()：

@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!")

未找到工具

如果 LLM 调用的工具不属于该 Agent，框架会引发 ToolNotFoundError：

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!")

实用测试模式

替换 Depends 依赖项

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")

替换 Inject 依赖项

只需向 agent.ask(...) 传入 dependencies={...}：

await agent.ask("Read", dependencies={"database_pool": fake_pool})

捕获流事件

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"

多轮模拟 — 响应列表按 ask() 计算，而非按对话计算

TestConfig(...) 的响应列表会在单次 ask() 内，跨该轮中重复的 LLM 调用依次消耗——这就是为什么 TestConfig(ToolCallEvent("my_tool"), "final result") 能用于调用工具的轮次（LLM 发出工具调用，工具运行，LLM 再次调用并获得 "final result"）。在内部，TestConfig.create() 会返回一个新的客户端，其迭代器从 responses[0] 开始；这会在每次 ask() 时执行一次——因此，每次新的 ask()（reply.ask(...) 链，或由网络适配器／自动回复 Agent 驱动的每一轮）都会将游标重置到第一个响应。列出更多响应并不能让你指定“对话第 2 轮与第 1 轮不同”。

要让多次 ask() 调用产生不同响应，可以：

• 通过每次 ask() 的 config= 覆盖项，为每一轮传入新的 TestConfig(...)（await agent.ask("…", config=TestConfig("turn-2 reply"))），或
• 使用 ToolCallEvent 模拟模型，并将每轮逻辑放入一个有状态工具中——可以是跟踪自身调用次数并据此返回结果的闭包或类实例。（当由测试代码以外的内容驱动轮次循环时很有用，例如 workflow／discussion 通道；此时无法注入每轮单独的 config=。）

深入了解

• 源文档：website/docs/user-guide/testing.mdx。
• 测试标记／异步配置 — 仓库的 pyproject.toml。使用 @pytest.mark.asyncio（项目使用 pytest-asyncio）。
• 流（用于断言事件）：website/docs/user-guide/advanced/stream.mdx。

常见问题

• 忘记添加 @pytest.mark.asyncio — 测试会跳过或出现异常失败。
• 响应数量不匹配 — 如果 Agent 在一次 ask() 内进行的 LLM 调用次数超出预期，TestConfig 会耗尽响应（例如工具错误后又进行一次 LLM 调用）；StopIteration 会传播。添加更多位置参数或断言调用顺序。（游标按 ask() 计算，因此所需数量是“单轮中的 LLM 调用次数”，而不是“对话中的轮数”——请参见上文的多轮模拟。）
• 模拟了 LLM，却没有模拟工具 — 你的工具函数仍会运行（并可能访问真实 API／磁盘）。如果要隔离 LLM 行为，请模拟工具，或替换其 Depends 以注入测试替身。
• 设置了 response_schema，却断言 reply.body — body 是原始文本。请使用 await reply.content() 获取经过验证的值。
• 在异步测试之间共享 Agent 实例 — Agent 会携带可变状态（变量、依赖项）。每个测试都应构造新的 Agent，以确保隔离。
• 在 CI 中使用真实的提供方客户端 — 在每个测试中或通过 fixture 用 TestConfig 包装提供方配置；切勿依赖测试环境中存在 OPENAI_API_KEY 等变量。