ag2-testing

用途

ag2-testing 用于测试构建 AG2 智能体的代码、自定义 @tool 函数、中间件以及响应模式，适用于任何你不想发起真实 LLM API 调用的测试场景。它通过 ag2.testing 提供的 TestConfig 来替换模型客户端，从而对 LLM 行为进行模拟。

主要能力

模拟 LLM 响应

将 TestConfig 作为智能体的 config 传入，或在单次 ask() 时传入。每个位置参数对应本次 ask() 内下一次 LLM 调用的模拟响应：字符串表示文本回复，ToolCallEvent 表示工具调用分发。示例：

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

模拟成功的工具调用

先传入 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 调用了智能体没有的工具，框架会抛出 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"

使用方式与多轮模拟说明

TestConfig 的响应列表是在单次 ask() 内被消耗的，跨该轮中多次重复的 LLM 调用，而不是跨整段对话。因此 TestConfig(ToolCallEvent("my_tool"), "final result") 可以用于带工具调用的一轮：LLM 发出工具调用，工具执行，再次调用 LLM 并得到 "final result"。内部实现上，TestConfig.create() 返回一个全新的客户端，其迭代器从 responses[0] 开始，且每次 ask() 只执行一次；因此每个新的 ask()（无论是 reply.ask(...) 链，还是由网络适配器或自动回复智能体驱动的每一轮）都会把游标重置到第一个响应。列出更多响应并不能表达“对话第 2 轮与第 1 轮不同”。

若希望在多次 ask() 之间产生变化，可以：

通过每次 ask() 的 config= 覆盖传入新的 TestConfig(...)，例如 await agent.ask("…", config=TestConfig("turn-2 reply"))；或
用 ToolCallEvent 模拟模型，并把每轮逻辑放在有状态的工具中——用闭包或类实例记录被调用次数并据此返回。当轮次循环由测试代码之外的东西驱动时（例如工作流/讨论频道），无法注入每次轮次的 config=，这种做法很有用。

进一步参考

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

常见陷阱

忘记 @pytest.mark.asyncio——测试会被跳过或以异常方式失败。
响应数量不匹配——如果智能体在单次 ask() 内发起的 LLM 调用比预期多（例如工具出错后再次调用 LLM），TestConfig 会用尽响应，并传播 StopIteration。需要增加位置参数或断言调用顺序。游标是按 ask() 的，因此所需数量是“一轮内的 LLM 调用次数”，而不是“对话的轮次数”。
只模拟 LLM 而未模拟工具——工具函数仍会运行（可能访问真实 API 或磁盘）。若要隔离 LLM 行为，请模拟工具，或覆写其 Depends 注入测试替身。
在设置了 response_schema 时断言 reply.body——body 是原始文本。应使用 await reply.content() 获取经过校验的值。
在异步测试间共享 Agent 实例——智能体带有可变状态（变量、依赖）。每个测试应构造全新的智能体以保证隔离。
在 CI 中使用真实提供商客户端——按测试或通过 fixture 用 TestConfig 包装提供商配置；不要依赖 OPENAI_API_KEY 等在测试环境中可用。