ag2-middleware

ミドルウェア

使用するタイミング

ミドルウェアは、エージェント、モデルクライアント、ツール自体を変更せずに、複数の実行に一貫して適用する横断的な処理に使用します。一般的なユースケースは次のとおりです。

• ロギング、トレーシング、時間計測
• 一時的な障害が発生した場合の再試行
• モデルに渡す前の履歴のトリミング
• トークン使用量の上限設定または推定
• ツールの引数や結果の書き換え
• ツール実行前のポリシー適用
• 人間からの入力リクエストの監査

4 つのフック

BaseMiddleware は 4 つの非同期フックを公開します。必要なものだけを実装してください。

フック	対象	用途
on_turn(call_next, event, context) → ModelResponse	エージェントのターン全体	全体のレイテンシー、リクエスト／レスポンスの確認、ターン単位のポリシー
on_llm_call(call_next, events, context) → ModelResponse	各 LLM API 呼び出し	再試行、ロギング、履歴のトリミング、リクエストの変更、キャッシュ
on_tool_execution(call_next, event, context) → ToolResultType	各ツール呼び出し	引数の検証、結果の秘匿化、失敗時のフォールバック、アクセス制御
on_human_input(call_next, event, context) → HumanMessage	各 context.input()	監査、プロンプトの書き換え、自動的なショートサーキット、レート制限

各インスタンスはターンごとに 1 回作成され、self にターン単位の状態を保持できます。同じインスタンスで複数のフックを実装できます。

組み込みミドルウェア

ag2.middleware からインポートできます。

ミドルウェア	目的	コンストラクター
LoggingMiddleware	ターンの開始／終了、各 LLM 呼び出し、各ツール実行を記録します	引数なし
RetryMiddleware	失敗した LLM 呼び出しを再試行します	max_retries=N、retry_on=ExceptionClass
HistoryLimiter	LLM 呼び出し前のイベント数に上限を設定します	max_events=N
TokenLimiter	LLM 呼び出し前に文字数ベースでトークン予算の上限を設定します	max_tokens=N、chars_per_token=4
TelemetryMiddleware	OpenTelemetry GenAI スパン（ag2-telemetry を参照）	telemetry スキルを参照

登録 — エージェント単位

すべてのターンに適用します。

from ag2 import Agent
from ag2.config import OpenAIConfig
from ag2.middleware import LoggingMiddleware, RetryMiddleware

agent = Agent(
    "assistant",
    config=OpenAIConfig(model="gpt-4o-mini"),
    middleware=[
        LoggingMiddleware(),
        RetryMiddleware(max_retries=2),
    ],
)

登録 — 呼び出し単位

1 つのターンに一時的なミドルウェアを追加します。agent.ask(...) と reply.ask(...) のどちらも受け付けます。

from ag2.middleware import TokenLimiter

reply = await agent.ask("Summarise the latest messages.", middleware=[LoggingMiddleware()])
next_turn = await reply.ask("Now answer in one paragraph.", middleware=[TokenLimiter(max_tokens=4000)])

呼び出し単位のミドルウェアは、エージェントのミドルウェアリストの後に追加されます。

実行順序

ミドルウェアは、入れ子になった with ブロックと同様に、登録順に実行されます。[A, B, C] を登録すると、A → B → C の順に入り、C → B → A の順に戻ります。

enter A
  enter B
    enter C
      <LLM call>
    exit C
  exit B
exit A

ロギング、変更、再試行を組み合わせる場合は、この順序が重要です。RetryMiddleware で変更後のリクエストを再試行する場合、変更処理を再試行の内側に配置します。各再試行の試行を個別に記録する場合は、ロギングを再試行の内側に配置します。

独自のミドルウェアを作成する

BaseMiddleware をサブクラス化し、必要なフックを実装します。

import logging
from collections.abc import Sequence
from ag2 import Agent, Context
from ag2.config import OpenAIConfig
from ag2.events import BaseEvent, ModelResponse, ToolCallEvent
from ag2.middleware import BaseMiddleware, LLMCall, Middleware, ToolExecution

class AuditMiddleware(BaseMiddleware):
    def __init__(self, event: BaseEvent, context: Context, logger: logging.Logger) -> None:
        super().__init__(event, context)
        self.logger = logger

    async def on_llm_call(self, call_next: LLMCall, events: Sequence[BaseEvent], context: Context) -> ModelResponse:
        self.logger.info("Calling model with %d events", len(events))
        response = await call_next(events, context)
        self.logger.info("Model returned: %s", response)
        return response

    async def on_tool_execution(self, call_next: ToolExecution, event: ToolCallEvent, context: Context):
        self.logger.info("Executing tool: %s", event.name)
        return await call_next(event, context)

agent = Agent(
    "assistant",
    config=OpenAIConfig(model="gpt-4o-mini"),
    middleware=[
        Middleware(AuditMiddleware, logger=logging.getLogger("ag2.audit")),
    ],
)

ミドルウェアで event と context 以外のコンストラクター引数が必要な場合は、登録時に Middleware(YourClass, ...) でラップしてください。設定不要のミドルウェアは、そのまま渡せます（middleware=[LoggingMiddleware()]）。

ツール単位とエージェント単位

1 つのツールにのみ適用する処理（検証、そのツールの出力の秘匿化、承認ゲート）には、代わりにツールミドルウェアを使用します。@tool、@agent.tool、または Toolkit に middleware=[hook] を指定します。構文については ag2-add-custom-tool を参照してください。組み込みの approval_required()（ag2-hitl を参照）はツールミドルウェアです。

エージェントミドルウェアはツールミドルウェアの外側で実行されます。BaseMiddleware.on_tool_execution() では、ツール単位のフックを含む実行全体を確認できます。

適切なフックの選び方

• on_turn → リクエスト／レスポンスのライフサイクル全体に関する処理。
• on_llm_call → モデルに渡す内容や、モデルから返される内容に関する処理。
• on_tool_execution → 複数のツールにまたがるツールの安全性、監査、結果の整形。
• ツール単位のミドルウェア（BaseMiddleware ではありません）→ 1 つのツール定義に関する処理。
• on_human_input → HITL リクエスト／レスポンスのインターセプト。

さらに詳しく

• references/builtin_middleware.md — すべての組み込みミドルウェアのパラメーター、一般的なユースケースのレシピ、それぞれに適した場面について説明しています。
• website/docs/user-guide/middleware.mdx — 詳細なリファレンス、順序の例、カスタムミドルウェアのガイドラインを掲載しています。
• website/docs/user-guide/tools/tool_middleware.mdx — ツールごとのフック（考え方が異なり、BaseMiddleware ではなく通常の非同期呼び出し可能オブジェクトを使用します）。
• OpenTelemetry の計装について詳しくは、ag2-telemetry を参照してください。

よくある落とし穴

• コンストラクター引数に Middleware(...) を付け忘れる — middleware=[AuditMiddleware]（ラッパーなし）が機能するのは、クラスが event と context だけを必要とする場合に限られます。それ以外の場合は、middleware=[Middleware(AuditMiddleware, logger=...)] のようにラップしてください。
• 変更の適用順序による予想外の挙動 — ミドルウェアは登録順に実行されます。ミドルウェア A が履歴をトリミングし、ミドルウェア B がそれをログに記録する場合は、B がトリミング後の内容を参照できるよう、[A, B] の順に登録してください。
• 呼び出しごとのミドルウェアはエージェントのミドルウェアを置き換えない — 追加されます。エージェントのミドルウェアも引き続き実行されます。
• 1つの大きなミドルウェアで5つの処理を行う — フックの役割を絞ってください。ログ記録、再試行、変更、ポリシーを1つのクラスにまとめると、挙動や順序を把握しにくくなります。複数のインスタンスに分割してください。
• 1つのツールに対して on_tool_execution 内で event.name による分岐を行う — これは設計上の問題を示しています。1つのツールに対する動作にはツール単位のミドルウェアを使い、on_tool_execution は横断的なポリシーのために使用してください。
• OpenTelemetry の計装をカスタムコードに実装する — そのための TelemetryMiddleware があります。ag2-telemetry を参照してください。