ag2-middleware

Compare original and translation side by side

🇺🇸

Original

English

🇨🇳

Translation

Chinese

Middleware

中间件

When to use

适用场景

Middleware is for cross-cutting behaviour that should apply consistently across many runs without changing the agent, model client, or tools themselves. Common use cases:

Logging, tracing, timing
Retry on transient failures
Trim history before it reaches the model
Cap or estimate token usage
Rewrite tool arguments / results
Enforce policies before a tool runs
Audit human-input requests

中间件适用于跨领域行为，无需修改Agent、模型客户端或工具本身，即可在多次运行中一致应用。常见使用场景：

日志记录、追踪、计时
临时故障重试
在历史记录传入模型前进行裁剪
限制或估算令牌使用量
重写工具参数/结果
在工具运行前执行策略校验
审计人工输入请求

Four hooks

四个钩子函数

BaseMiddleware

exposes four async hooks. Implement only the ones you need:

Hook	Wraps	Use for
`on_turn(call_next, event, context) → ModelResponse`	The whole agent turn	Total latency, request/response inspection, turn-level policies
`on_llm_call(call_next, events, context) → ModelResponse`	Each LLM API call	Retry, logging, history trim, request mutation, caching
`on_tool_execution(call_next, event, context) → ToolResultType`	Each tool invocation	Validate args, redact results, fallback on failure, access control
`on_human_input(call_next, event, context) → HumanMessage`	Each `context.input()`	Audit, rewrite prompts, automated short-circuit, rate limit

Each instance is created once per turn and can hold per-turn state on

self

. The same instance can implement multiple hooks.

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

上存储轮次级状态。同一个实例可实现多个钩子函数。

Built-in middleware

内置中间件

Importable from

ag2.middleware

Middleware	Purpose	Constructor
`LoggingMiddleware`	Logs turn start/end, each LLM call, each tool execution	no args
`RetryMiddleware`	Retries failed LLM calls	`max_retries=N` , `retry_on=ExceptionClass`
`HistoryLimiter`	Cap event count before LLM call	`max_events=N`
`TokenLimiter`	Char-based token-budget cap before LLM call	`max_tokens=N` , `chars_per_token=4`
`TelemetryMiddleware`	OpenTelemetry GenAI spans (see `ag2-telemetry` )	see telemetry skill

可从

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` ）	详见遥测技能

Registration — agent-level

注册方式——Agent级

Apply to every turn:

python

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

应用于每一轮次：

python

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

Registration — call-level

注册方式——调用级

Add temporary middleware for one turn. Both

agent.ask(...)

and

reply.ask(...)

accept it:

python

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

Call-level middleware is appended after the agent's middleware list.

为单个轮次添加临时中间件。

agent.ask(...)

和

reply.ask(...)

均支持：

python

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的中间件列表之后。

Ordering

执行顺序

Middleware runs in registration order, like nested

with

blocks. Registering

[A, B, C]

enters

A → B → C

and unwinds

C → B → A

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

This matters when you mix logging, mutation, retry. If

RetryMiddleware

should retry mutated requests, mutation goes inside retry; if you want each retry attempt logged separately, logging goes inside retry.

中间件按注册顺序运行，类似嵌套的

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

需要重试经过修改的请求，修改逻辑应放在重试内部；如果希望单独记录每次重试尝试，日志逻辑应放在重试内部。

Writing your own

自定义中间件

Subclass

BaseMiddleware

, implement the hooks you need:

python

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

If your middleware needs constructor args beyond

event

and

context

, wrap with
Middleware(YourClass, ...)
when registering. Zero-config middleware can be passed bare (

middleware=[LoggingMiddleware()]

继承

BaseMiddleware

，实现所需的钩子函数：

python

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

）。

Tool-scoped vs agent-scoped

工具级与Agent级中间件对比

For behaviour that applies to one tool only (validation, redaction for that tool's output, approval gates), use tool middleware instead —

middleware=[hook]

@tool

@agent.tool

, or

Toolkit

. See

ag2-add-custom-tool

for the syntax. The

approval_required()

built-in (see

ag2-hitl

) is a tool middleware.

Agent middleware runs outside tool middleware:

BaseMiddleware.on_tool_execution()

sees the full execution including tool-scoped hooks.

如果行为仅适用于单个工具（验证、该工具输出脱敏、审批 gate），请使用工具中间件——在

@tool

、

@agent.tool

或

Toolkit

上设置

middleware=[hook]

。语法详见

ag2-add-custom-tool

。内置的

approval_required()

（详见

ag2-hitl

）就是一个工具中间件。

Agent级中间件运行在工具级中间件外部：

BaseMiddleware.on_tool_execution()

会看到包含工具级钩子在内的完整执行过程。

Picking the right hook

选择合适的钩子函数

```
on_turn
```
→ behaviour about the whole request/response lifecycle.
```
on_llm_call
```
→ behaviour about what goes into / comes out of the model.
```
on_tool_execution
```
→ tool safety / auditing / result shaping across many tools.
Tool-scoped middleware (not
```
BaseMiddleware
```
) → behaviour for a single tool's definition.
```
on_human_input
```
→ intercept HITL requests/responses.

```
on_turn
```
→ 针对整个请求/响应生命周期的行为。
```
on_llm_call
```
→ 针对模型输入/输出的行为。
```
on_tool_execution
```
→ 跨多个工具的工具安全/审计/结果处理。
工具级中间件（非
```
BaseMiddleware
```
） → 针对单个工具定义的行为。
```
on_human_input
```
→ 拦截HITL请求/响应。

Going deeper

深入学习

```
references/builtin_middleware.md
```
— every built-in's params, common-case recipes, when each fits.
```
website/docs/user-guide/middleware.mdx
```
— full reference, ordering examples, custom-middleware guidelines.
```
website/docs/user-guide/tools/tool_middleware.mdx
```
— per-tool hooks (different mental model — plain async callables, not
```
BaseMiddleware
```
).
For OpenTelemetry instrumentation specifically, see
```
ag2-telemetry
```
.

```
references/builtin_middleware.md
```
—— 每个内置中间件的参数、常见场景示例及适用情况。
```
website/docs/user-guide/middleware.mdx
```
—— 完整参考、执行顺序示例、自定义中间件指南。
```
website/docs/user-guide/tools/tool_middleware.mdx
```
—— 工具级钩子（不同的思维模型——纯异步可调用函数，而非
```
BaseMiddleware
```
）。
如需OpenTelemetry instrumentation相关内容，详见
```
ag2-telemetry
```
。

Common pitfalls

常见误区

Forgetting
Middleware(...)
for constructor args —

middleware=[AuditMiddleware]

(no wrapper) only works if the class needs only

event

and

context

. Otherwise wrap:

middleware=[Middleware(AuditMiddleware, logger=...)]

Mutation order surprises — middleware runs in registration order. If middleware A trims history and middleware B logs it, register
```
[A, B]
```
so B sees the trimmed view.
Per-call middleware doesn't replace agent middleware — it's appended. Agent middleware still runs.
One big middleware doing five things — keep hooks focused. Logging + retry + mutation + policy in one class is hard to reason about and order. Split into multiple instances.
on_tool_execution
branching on
event.name
for a single tool — that's a smell; use tool-scoped middleware for one-tool behaviour and reserve
```
on_tool_execution
```
for cross-cutting policies.
Putting OpenTelemetry instrumentation in custom code — there's a
```
TelemetryMiddleware
```
for that; see
```
ag2-telemetry
```
.

忘记为构造函数参数使用
Middleware(...)
包裹——如果类仅需要
```
event
```
和
```
context
```
，
```
middleware=[AuditMiddleware]
```
（无包裹）可行。否则需包裹：
```
middleware=[Middleware(AuditMiddleware, logger=...)]
```
。
修改顺序引发意外——中间件按注册顺序运行。如果中间件A裁剪历史记录，中间件B记录历史记录，应注册
```
[A, B]
```
，这样B看到的是裁剪后的内容。
调用级中间件不会替换Agent级中间件——它是追加的，Agent级中间件仍会运行。
单个大中间件处理多项任务——保持钩子函数专注。将日志记录+重试+修改+策略放在一个类中会难以理解和排序，应拆分为多个实例。
on_tool_execution
针对单个工具按
event.name
分支——这是不良实践；针对单个工具的行为应使用工具级中间件，
```
on_tool_execution
```
应保留用于跨领域策略。
在自定义代码中实现OpenTelemetry instrumentation——已有
```
TelemetryMiddleware
```
可实现此功能，详见
```
ag2-telemetry
```
。