ag2-testing

エージェントとツールのテスト

使用する場面

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 がエージェントにないツールを呼び出すと、フレームワークは 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(...) のレスポンスリストは、1 回の ask() 内で、そのラウンド中に繰り返される LLM 呼び出しに対して消費されます。ツールを使うターンで TestConfig(ToolCallEvent("my_tool"), "final result") が機能するのはこのためです（LLM がツール呼び出しを出力し、ツールが実行され、LLM が再度呼び出されて "final result" を受け取ります）。内部では、TestConfig.create() がイテレーターを responses[0] から開始する新しいクライアントを返します。これは ask() ごとに 1 回 行われるため、新しい ask()（reply.ask(...) チェーン、またはネットワークアダプターや自動応答エージェントが実行する各ターン）ごとに、カーソルは 最初のレスポンスから再開します。レスポンスを追加しても、「会話の 2 ターン目は 1 ターン目と異なる」と指定することは できません。

複数の ask() 呼び出し間で変化をつけるには、次のいずれかを行います。

• ターンごとに ask() の config= オーバーライドを使って、新しい TestConfig(...) を渡す（await agent.ask("…", config=TestConfig("turn-2 reply"))）、または
• LLM を 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 を忘れる — テストがスキップされたり、予期しない形で失敗したりします。
• レスポンス数が一致しない — エージェントが 1 回の ask() 内で想定より多くの LLM 呼び出しを行うと、TestConfig のレスポンスが不足します（たとえば、ツールエラーの後に別の LLM 呼び出しが行われる場合など）。StopIteration が伝播します。位置引数を追加するか、呼び出し順序をアサートしてください。（カーソルは ask() ごとなので、必要な数は「会話のターン数」ではなく「1 ラウンド内の LLM 呼び出し数」です。上記の 複数ターンのモック を参照してください。）
• LLM をモックしてツールをモックしない — ツール関数は引き続き実行されます（実際の API やディスクにアクセスする場合があります）。LLM の動作だけを分離する場合は ツール をモックするか、Depends をオーバーライドしてテスト用の代替オブジェクトを注入してください。
• response_schema を設定しているのに reply.body をアサートする — body は生のテキストです。検証済みの値には await reply.content() を使用してください。
• 非同期テスト間で Agent インスタンスを共有する — エージェントは変更可能な状態（変数、依存関係）を保持します。テストを分離するため、テストごとに新しいエージェントを構築してください。
• CI で実際のプロバイダークライアントを使う — プロバイダーの設定をテストごと、またはフィクスチャ経由で TestConfig でラップしてください。テスト環境で OPENAI_API_KEY などが利用できることを前提にしないでください。