ag2-middleware

中间件

适用场景

中间件用于处理横切行为，让其在多次运行中一致生效，同时不改变 Agent、模型客户端或工具本身。常见用例：

• 日志记录、追踪、计时
• 瞬时故障时重试
• 在历史记录传给模型前进行精简
• 限制或估算 token 用量
• 重写工具参数 / 结果
• 在工具运行前强制执行策略
• 审计人工输入请求

四个钩子

BaseMiddleware 提供四个异步钩子。只需实现所需的钩子：

钩子	包装范围	用途
on_turn(call_next, event, context) → ModelResponse	整个 Agent 轮次	总延迟、检查请求 / 响应、轮次级策略
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()	审计、重写提示词、自动短路、速率限制

每个实例每轮创建一次，并可在 self 上保存每轮状态。同一实例可以实现多个钩子。

内置中间件

可从 ag2.middleware 导入：

中间件	用途	构造函数
LoggingMiddleware	记录轮次开始 / 结束、每次 LLM 调用和每次工具执行	无参数
RetryMiddleware	重试失败的 LLM 调用	max_retries=N、retry_on=ExceptionClass
HistoryLimiter	在调用 LLM 前限制事件数量	max_events=N
TokenLimiter	在调用 LLM 前按字符数限制 token 预算	max_tokens=N、chars_per_token=4
TelemetryMiddleware	OpenTelemetry GenAI spans（参见 ag2-telemetry）	参见 telemetry 技能

注册 — Agent 级

应用于每一轮：

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),
    ],
)

注册 — 调用级

为单轮添加临时中间件。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)])

调用级中间件会追加在 Agent 的中间件列表之后。

顺序

中间件按注册顺序运行，与嵌套的 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()]）。

工具级与 Agent 级

如果行为只适用于单个工具（验证、编辑该工具的输出、审批关卡），请改用工具中间件——在 @tool、@agent.tool 或 Toolkit 上使用 middleware=[hook]。语法请参见 ag2-add-custom-tool。内置的 approval_required()（参见 ag2-hitl）是一种工具中间件。

Agent 中间件在工具中间件之外运行：BaseMiddleware.on_tool_execution() 会看到包括工具级钩子在内的完整执行过程。

如何选择合适的钩子

• on_turn → 与整个请求 / 响应生命周期相关的行为。
• on_llm_call → 与传入模型或从模型返回的内容相关的行为。
• on_tool_execution → 跨多个工具的工具安全、审计或结果调整。
• 工具级中间件（不是 BaseMiddleware）→ 与单个工具定义相关的行为。
• 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 会记录历史记录，则应注册为 [A, B]，这样 B 看到的就是精简后的视图。
• 单次调用的中间件不会替代 Agent 中间件 — 它会追加到其中。Agent 中间件仍会运行。
• 一个大型中间件包揽五件事 — 让钩子各司其职。在一个类中同时处理日志记录、重试、修改和策略，会很难理清逻辑和执行顺序。请拆分为多个实例。
• 针对单个工具根据 event.name 在 on_tool_execution 中分支处理 — 这是一个不好的信号；单个工具的行为应使用工具级中间件，而 on_tool_execution 应留给横切策略。
• 在自定义代码中加入 OpenTelemetry 检测 — 已有 TelemetryMiddleware 可处理此事；请参阅 ag2-telemetry。