中间件

中间件（Middleware）是 AgentScope 2.x 处理横切关注点的统一机制。追踪、Token 预算、RAG 注入、长期记忆、TTS——这些能力都不侵入 Agent 核心的 ReAct 循环，而是以洋葱层的形式包裹在执行链 外侧。2.x 用中间件替代了 1.x 的 hooks 机制，把"在 reply 的某个阶段插入逻辑"从分散的回调收拢为 统一的、可组合的、有明确生命周期的拦截层。

版本基线：本文以 Python agentscope 2.0.4（PyPI，2026-07-07）为准，要求 Python >=3.11。 导入路径为 from agentscope.middleware import MiddlewareBase, TracingMiddleware, ...（源： src/agentscope/middleware/__init__.py）。2.x 文档站存在版本化陷阱：非版本化的 /en/ URL 会 404，本页所有官方链接均使用 versions/2.0.4/en/ 形式。

为什么需要中间件

Agent 的核心职责是"推理-行动循环"：调用模型、决策工具、执行工具、回灌结果。但真实场景中，你 几乎总需要在循环的外围叠加一层"非业务"逻辑：

• 追踪：每次 reply 花了多久、调用了几次模型、用了多少 token。
• 预算：单次 reply 的 token 开销不能超过上限，超限就强制收尾。
• RAG：在推理前检索知识库，把结果注入系统提示词或上下文。
• 长期记忆：跨会话记住用户偏好与历史事实。
• TTS：把最终回复转成语音。

这些逻辑的共同特征是：不属于 ReAct 循环本身，但需要在循环的特定阶段介入。如果把它们写进 Agent 核心代码，会导致核心膨胀、难以测试、无法按需关闭。中间件把这些逻辑抽到独立的、可插拔的 层，Agent 核心保持纯粹。

前端类比

如果你来自前端，中间件就是 Koa 的洋葱模型或 Redux 的 middleware 链：

• 请求（reply / reply_stream）像 HTTP request 一样，从最外层中间件穿入，经过每一层的"前置 逻辑"，到达 Agent 核心（类似 Redux 的 next(action) 到达 reducer），再从核心穿出，经过每一层 的"后置逻辑"，最终返回。
• next_handler 就是 Koa 里的 await next() 或 Redux 里的 next(action)——调用它把控制权交给 下一层，它返回后再继续执行当前层的后置逻辑。
• 不调 next_handler 就等于 next() 不被调用，链条在此截断，Agent 核心不会执行。

AgentScope 原生语义：中间件不是 React 的 Hooks（useEffect / useMemo）。Hooks 是组件渲染 周期内的副作用挂载点，与渲染阶段绑定；中间件是独立对象，包裹整个 reply 执行链，与渲染无关。 每个中间件实现一个或多个 hook 方法（on_reply / on_reasoning / ...），运行时由 Agent 自动检测 哪些 hook 被实现并按洋葱顺序串联。中间件实例本身无状态——运行时状态存在 agent.state.middle_context 里，因此同一个中间件实例可以被多个 Agent 安全共享。

洋葱模型

下面这张图说明一次 reply / reply_stream 调用如何穿过中间件链。图前先说明：中间件按 middlewares=[...] 列表顺序从外到内包裹，请求逐层穿入到达 Agent 核心，响应再逐层穿出。每一层 可以在前置阶段修改输入、在后置阶段修改输出，也可以直接短路（不调 next_handler）。

图后解释：

• 前置逻辑在 next_handler() 调用之前执行，可以修改 input_kwargs、注入上下文、做前置校验。
• Agent 核心是 next_handler 链的终点——实际的 ReAct 循环（推理、工具决策、工具执行、模型 调用）在这里发生。
• 后置逻辑在 next_handler() 返回之后执行，可以修改产出的事件、记录耗时、做结果改写。
• 洋葱顺序：middlewares=[A, B] 意味着 A 是最外层，B 是次外层。请求经过 A → B → 核心 → B → A。 这和 Koa 完全一致：app.use(A) 在前的是外层。

6 个生命周期 hook

中间件有 6 个 hook 点，覆盖 Agent 执行链的各个阶段。以下方法名全部来自源码 src/agentscope/middleware/_base.py，不要凭直觉拼写——方法名拼错会导致 hook 静默不生效。

5 个采用洋葱模式（有 next_handler，可做前置 + 后置），1 个采用管道模式（无 next_handler，按顺序传递值）：

hook	模式	触发时机	input_kwargs 关键字段	典型用途
on_reply	洋葱	整个 reply / reply_stream	inputs（Msg / 恢复事件 / None）	全链路追踪、Token 预算、TTS
on_reasoning	洋葱	每轮推理（模型调用 + 工具决策）	tool_choice	推理日志、注入额外上下文、预算拦截
on_acting	洋葱	单次工具执行（toolkit.call_tool）	tool_call（ToolCallBlock）	工具执行计时、工具结果改写
on_model_call	洋葱	原始模型 API 调用	messages、tools、tool_choice、current_model	请求/响应日志、模型 fallback
on_compress_context	洋葱	上下文压缩	context_config、instructions	自定义压缩策略、压缩前后审计
on_system_prompt	管道	系统提示词组装	current_prompt: str（直接参数，非 dict）	动态注入提示词、RAG 上下文拼接

关键细节：

• on_acting 只包裹 toolkit.call_tool——即纯 I/O 执行层。权限检查、输入校验、上下文写入 都在 Agent 侧、此 hook 之外完成，因此在这个 hook 里看不到 permission 决策过程。这一隔离使 next_handler 可以安全地卸载到后台 task，它不会自行修改 Agent 上下文。
• on_system_prompt 是管道模式——签名是 async def on_system_prompt(self, agent, current_prompt: str) -> str， 没有 next_handler。多个中间件的 on_system_prompt 按顺序执行，每个接收上一个的输出作为输入， 最终结果成为 Agent 的系统提示词。这和洋葱模式不同：你不能在后置阶段"修改回来"。
• 每个 hook 都是可选的——只实现你需要的那几个。运行时通过 is_implemented() 自动检测子类 覆盖了哪些方法，未实现的 hook 直接跳过。

此外还有两个辅助方法（非生命周期 hook）：

• list_tools() → list[ToolBase]：返回此中间件提供的工具。可选实现，RAG 等中间件用它向 Agent 暴露检索工具。
• get_middleware_key() → str：返回唯一键，用于在 agent.state.middle_context 中隔离保存此 中间件的运行时状态。默认返回类名，子类可覆盖。

内置中间件

agentscope.middleware 开箱提供以下中间件（源：src/agentscope/middleware/__init__.py）：

中间件	作用
TracingMiddleware	OpenTelemetry 追踪，自动记录 reply / reasoning / acting 等阶段跨度
ReplyBudgetControlMiddleware	按 reply 控制 Token 预算，超限后注入 hint 并强制 tool_choice="none"
TTSMiddleware	把回复文本转语音
RAGMiddleware	检索增强生成，在推理前检索知识库并注入相关上下文
AgenticMemoryMiddleware	Agentic 长期记忆（预览能力，详见长期记忆页）
Mem0Middleware	基于 Mem0 后端的长期记忆（预览能力，详见长期记忆页）
ReMeMiddleware	基于 ReMe 后端的长期记忆（预览能力，详见长期记忆页）

⚠️ 源码类名是 ReplyBudgetControlMiddleware，不是 BudgetControlMiddleware。部分 release notes 使用了简写名，但导入时必须用完整的源码名，否则 ImportError。

ReplyBudgetControlMiddleware 详解

这是最常用的内置中间件之一，构造参数如下（源：src/agentscope/middleware/_budget.py）：

参数	类型	默认值	作用
token_budget	float	必填	单次 reply 的最大加权 token 成本上限
input_token_weight	float	1	输入 token 权重乘数
output_token_weight	float	1	输出 token 权重乘数（输出通常更贵，可设 > 1）
hint_message	str	内置	超预算时注入上下文的提示消息

加权成本计算：cost = input_token_weight × input_tokens + output_token_weight × output_tokens。 每次模型调用后累加，一旦累计成本达到 token_budget，中间件会在下一轮推理前注入 HintBlock 提示 Agent 收尾，并把 tool_choice 强制为 ToolChoice(mode="none")，阻止继续调用工具。

预算状态存在 agent.state.middle_context[middleware_key][reply_id] 中，因此跨 HITL 中断与恢复 也能保持累计。reply 结束时通过 ReplyEndEvent 自动清理。

自定义中间件

自定义中间件只需三步：继承 MiddlewareBase、覆盖需要的 hook、在 hook 中调用 next_handler。

下面是一个计时中间件，用 on_reply 包裹整个 reply 过程，在前置阶段记录开始时间，在后置阶段 打印耗时：

import time
from typing import AsyncGenerator, Callable

from agentscope.agent import Agent
from agentscope.middleware import MiddlewareBase


class TimingMiddleware(MiddlewareBase):
    """记录整个 reply 耗时的中间件。"""

    async def on_reply(
        self,
        agent: Agent,
        input_kwargs: dict,
        next_handler: Callable[..., AsyncGenerator],
    ) -> AsyncGenerator:
        start = time.perf_counter()
        print(f"[Timing] {agent.name} reply 开始")

        # 调用 next_handler 把控制权交给下一层 / Agent 核心
        # 产出的每个事件透传给上层
        async for event in next_handler(**input_kwargs):
            yield event

        elapsed = time.perf_counter() - start
        print(f"[Timing] {agent.name} reply 耗时 {elapsed:.3f}s")

要点：

• next_handler 是 async generator——用 async for event in next_handler(...) 消费，每个 event 是 AgentEvent 或 Msg，用 yield event 透传给上层。
• 前置逻辑在 async for 之前，后置逻辑在 async for 之后——这正是洋葱模型的体现。
• input_kwargs 透传——如果你不需要修改输入，直接 next_handler(**input_kwargs) 传下去即可。 需要修改时（如预算中间件把 tool_choice 改为 "none"），改 input_kwargs 再传。

管道模式：on_system_prompt

on_system_prompt 不遵循洋葱模式，它是管道模式——没有 next_handler，直接接收当前提示词 字符串，返回改写后的字符串：

class DateInjectionMiddleware(MiddlewareBase):
    """在系统提示词中注入当前日期。"""

    async def on_system_prompt(
        self,
        agent: Agent,
        current_prompt: str,
    ) -> str:
        from datetime import date

        return current_prompt + f"\n\n当前日期：{date.today().isoformat()}"

多个实现了 on_system_prompt 的中间件按 middlewares 列表顺序依次执行，每个的输出是下一个的 输入。如果列表是 [A, B]，那么 final_prompt = B.on_system_prompt(A.on_system_prompt(original))。

装配

中间件通过 Agent 构造函数的 middlewares 参数装配，传入 MiddlewareBase 实例列表：

import asyncio
import os
from typing import AsyncGenerator

from agentscope.agent import Agent
from agentscope.credential import DashScopeCredential
from agentscope.message import UserMsg
from agentscope.middleware import (
    MiddlewareBase,
    ReplyBudgetControlMiddleware,
    TracingMiddleware,
)
from agentscope.model import DashScopeChatModel


class TimingMiddleware(MiddlewareBase):
    async def on_reply(
        self,
        agent: Agent,
        input_kwargs: dict,
        next_handler,
    ) -> AsyncGenerator:
        import time

        start = time.perf_counter()
        async for event in next_handler(**input_kwargs):
            yield event
        elapsed = time.perf_counter() - start
        print(f"[Timing] {elapsed:.3f}s")


agent = Agent(
    name="my_agent",
    system_prompt="你是一个严谨的中文助手。",
    model=DashScopeChatModel(
        credential=DashScopeCredential(
            api_key=os.environ["DASHSCOPE_API_KEY"],
        ),
        model="qwen-plus",
    ),
    middlewares=[
        TracingMiddleware(),                          # 最外层：全链路追踪
        ReplyBudgetControlMiddleware(                 # 次外层：Token 预算
            token_budget=10000,
            input_token_weight=1.0,
            output_token_weight=2.0,
        ),
        TimingMiddleware(),                           # 最内层：计时
    ],
)

# 正常调用 reply / reply_stream，中间件自动生效


async def main() -> None:
    result = await agent.reply(
        UserMsg(name="user", content="用三句话介绍 AgentScope。"),
    )
    print(result.get_text_content())


asyncio.run(main())

装配顺序即洋葱顺序：列表第一个是最外层，最后一个是最内层（最贴近 Agent 核心）。请求经过 Tracing → Budget → Timing → Agent 核心，响应原路返回 Agent 核心 → Timing → Budget → Tracing。

共享中间件实例：中间件实例本身无状态（运行时状态存在 agent.state.middle_context），因此 同一个中间件实例可以被多个 Agent 安全共享。如果你的中间件在 __init__ 里存了配置（如 token_budget），那只是只读配置，不冲突。

与 1.x hooks 的关系

1.x 提供了一套 hooks 机制（before_reply / after_reply 等），用于在 Agent 执行的特定阶段插入 回调。2.x 把 hooks 完全移除，用中间件替代。

两者的关键区别：

维度	1.x hooks	2.x Middleware
模型	回调函数注册	洋葱 / 管道拦截层
控制流	无法中断或短路	可通过不调 next_handler 短路
后置逻辑	需要单独的 after hook	同一方法内 next_handler 之后
组合性	回调之间无顺序保证	列表顺序即洋葱顺序，可预测
状态	回调内部自管	统一存 agent.state.middle_context

⚠️ 1.x 仅作历史对照：1.x 的 hooks 机制在 2.x 已移除，不能作为 2.x 代码使用。如果你在 迁移 1.x 代码，每个 hook 对应一个中间件 hook 方法，迁移指南见 1.x 到 2.x 迁移。

常见错误

• 忘记调 next_handler：洋葱模式下，不调 next_handler 意味着链条在此截断，Agent 核心不会 执行，reply 会静默返回空结果。这通常是无意的（比如在 async for 前提前 return 而非 yield）。正确做法是确保 next_handler(**input_kwargs) 被调用并 yield 其产出。
• hook 方法名拼错：方法名必须精确匹配源码（on_reply / on_reasoning / on_acting / on_model_call / on_compress_context / on_system_prompt）。运行时通过 is_implemented() 检测子类是否覆盖了方法，拼错名等于没实现，hook 静默不生效——不会报错，但逻辑不会执行。
• 在 hook 里做重 IO 阻塞事件循环：所有 hook 是 async，在 hook 里做同步阻塞 IO（如 requests.get、time.sleep、同步文件读写）会阻塞整个事件循环，影响所有并发的 Agent。应使用 async IO（aiohttp、asyncio.sleep、aiofiles）或把阻塞操作卸载到 asyncio.to_thread。
• on_system_prompt 当洋葱模式用：on_system_prompt 没有 next_handler 参数，签名是 (self, agent, current_prompt: str) -> str。如果你按洋葱模式写了 next_handler 参数，签名 不匹配，hook 不会被执行。
• 误用 BudgetControlMiddleware：源码类名是 ReplyBudgetControlMiddleware（源： _budget.py）。部分 release notes 和旧示例用了 BudgetControlMiddleware 简写名，但导入时 必须用完整源码名，否则 ImportError: cannot import name 'BudgetControlMiddleware'。
• 期望 on_acting 能拦截权限：on_acting 只包裹 toolkit.call_tool 的纯 I/O 执行，权限 检查在这之前已完成。要拦截权限决策，用 Permission 系统（见 权限系统），不要在 on_acting 里做。

先修

• 上下文管理 - 理解上下文压缩与 on_compress_context 的触发条件

下一步

• 权限系统 - 工具执行前的三态决策，与中间件 on_acting 互补

官方参考

• Middleware (2.0.4)
• Middleware 源码 __init__.py
• MiddlewareBase 源码 _base.py
• ReplyBudgetControlMiddleware 源码 _budget.py
• AgentScope GitHub 仓库