AgentScope 2.0源码解析系列第七篇 Middleware模块 把所有定制赶出 Agent 类

本文是 AgentScope 2.0 源码解析系列 第 7 篇（Middleware 模块）。

已发：整体架构 · Agent · Event · Message · Model · tool · permission　｜　下一篇：RAG。

关注追更，后续会持续更新 Workspace、Skill、formatter 等模块。

读完这篇你会带走这些源码层面的结论：

• 整个框架只有 1 个 Agent 类、零继承体系——所有定制（RAG/Tracing/预算/记忆/TTS）都被赶到了 MiddlewareBase 的 6 个 Hook 里，Agent 核心因此能保持稳定
• 6 个 Hook 不是随意切的：5 个洋葱钩子（on_reply/on_reasoning/on_acting/on_model_call/on_compress_context）精确落在 ReAct 循环的 5 个阶段边界，1 个变换器钩子（on_system_prompt）单独处理字符串增强
• is_implemented
   用「基类方法对象 is 子类方法对象」做检测——构造时一次性分桶到 6 条链，运行时只遍历列表，不抛异常、不用 sentinel、零反射
• execute_chain
   递归闭包里有个关键设计：next_handler(**{**input_kwargs, **kwargs}) 允许中间件改写传给下游的参数，预算耗尽时把 tool_choice 改成 none 就靠这一行
• on_acting
   有意只包住 toolkit.call_tool 这一层纯 I/O，权限检查和上下文写入都在钩子之外——这是它能被安全 offload 到后台任务的硬约束
• ReplyBudgetControlMiddleware
   把运行时状态全塞进 agent.state.middle_context[middleware_key][reply_id]，实例本身无状态，所以同一实例能安全跨 Agent 共享、HITL 中断后状态也自然续上

适合谁读：在写或准备改 Agent 框架、想给 Agent 加横切能力（RAG / Tracing / 预算 / 长期记忆 / TTS）、或对"如何让核心类不被业务撑爆"这个框架设计问题感兴趣的开发者。预计阅读：主线约 17 分钟（6 张字段表分散在各小节，深度查阅另需约 6 分钟）。

🎯 如果只记一件事

AgentScope 把"定制 Agent"这件事从「继承并重写」彻底改成了「挂中间件」——Agent 类因此保持单类零继承，所有变化点集中在 6 个拦截点 + 两种包裹模式（洋葱 for 事件流、变换器 for 字符串）。这套「核心循环不可变、变化点全外置到拦截器」的赌注，搬到任何要长期演进的核心框架都成立：宁可多设计几个 hook，也不要让业务把核心类继承成树。

一、这个模块到底在解决什么问题？

读 Agent 篇的时候，有件事会让人愣一下：整个 AgentScope 只有 1 个 Agent 类，没有任何子类。不是抽象基类 + N 个实现那种继承树，就是孤零零一个类。

这在 Agent 框架里很不寻常。多数框架（LangChain 的多种 Agent 类型、AutoGen 的 ConversableAgent 子类化）都靠继承来区分行为：ReActAgent 继承 BaseAgent，ToolCallingAgent 又继承 ReActAgent……定制一个新行为就加一层子类。

AgentScope 选了另一条路：Agent 类做最小闭环（ReAct 循环），把所有"额外能力"赶出去。RAG 检索、链路追踪、Token 预算、长期记忆、语音合成——这些本来很容易写进 Agent 类的逻辑，全部抽成独立的、可插拔的拦截器。Agent 在固定执行节点上调它们，自己不知道它们的存在。

这就是 Middleware 模块解决的问题。它不是"锦上添花的扩展点"，而是让 Agent 类保持单类的根本机制：没有中间件，要么 Agent 类膨胀成上帝类，要么不得不引入继承树。

一句话：Middleware 是 AgentScope 把定制能力从核心类里"搬出去"的载体——挂中间件就改行为，不动 Agent 源码。

二、这个模块在整个框架中的位置

图 1 · 6 个 Hook 如何落在 ReAct 循环的阶段边界上

中间件的输入是 agent 实例 + 当前阶段的 input_kwargs + 下游处理器 next_handler；产出是事件流（yield）或变换后的字符串。它处在 Agent 核心循环和外部能力之间：

• 能读能改
  ：agent.state（含 context、middle_context、reply_id）、input_kwargs
• 能拦截能注入
  ：事件流可以过滤、可以追加新事件
• 能短路
  ：不调 next_handler 就等于拦截整个下游
• 能产工具
  ：list_tools() 把工具塞进 Toolkit，让 Agent 能调用
• 能产提示词
  ：on_system_prompt 往系统提示词里加内容

依赖方向上，中间件依赖 agent、event、message、tool、model（on_model_call 包裹 ChatModelBase）、permission（工具的 check_permissions）；没有任何模块反向依赖中间件——中间件是被 Agent 调用的，不调别人。这个单向依赖是它能保持可插拔的关键。

Agent 在构造时收 middlewares: list[MiddlewareBase]，在 _reply、_reasoning、_acting、_call_model、_get_system_prompt、compress_context 六个方法里把对应钩子的中间件串成链。

三、为什么这样设计？

这是理解整个模块的关键。先讲设计，再讲代码。

3.1 拦截点为什么是 6 个，不是 1 个也不是 30 个

最容易想到的中间件设计是"一个大钩子"——在 reply 入口和出口各切一刀，所有定制都在这一个 on_reply 里做。简单，但粒度太粗：你想在"每次工具调用"时插日志，或者在"每次模型调用"时改参数，on_reply 帮不了你，它只看到整个 reply 的事件流，要从中过滤出 ModelCallEndEvent 再累加 token——能做，但别扭。

另一个极端是每个内部方法都开钩子。灵活，但框架复杂度爆炸，维护成本高，中间件作者也要面对几十个钩子不知该实现哪个。

AgentScope 的选法是跟着横切需求的自然分布走，在 ReAct 循环的 5 个阶段边界各切一刀：

这 5 个点的共同特征是都是洋葱模式——有明确的"调用前"和"调用后"，中间件能层层包裹。

第 6 个钩子 on_system_prompt 不在这个表里，因为它的语义完全不同：不包裹任何调用，就是"给我个字符串，我返回改过的字符串"。它走变换器模式，多个中间件顺序串联，前一个的输出是后一个的输入。为什么不用洋葱？因为洋葱的 next_handler 对"改字符串"这件事是多余的——你不需要在"改之前"和"改之后"各做事，你就是改一下。模式跟着语义走，这是这套设计里很值得学的一点。

3.2 is_implemented：为什么不用 hasattr，不抛异常

MiddlewareBase 的 6 个钩子默认实现都 raise RuntimeError（说明基类不打算被直接调用）。子类按需重写。问题来了：Agent 怎么知道某个中间件实现了哪些钩子，该把它放进哪条链？

最直觉的两种做法都有问题：

• hasattr
  ：每个子类都"有"这 6 个方法（继承自基类），hasattr 永远返回 True，没法区分"继承了默认实现"和"自己重写了"
• try 调用再捕 RuntimeError
  ：每次调用都要真跑一遍才知道，运行时开销大，而且异常本不该做控制流

AgentScope 的做法是比较方法对象身份（_base.py L52-L63）：

def is_implemented(self, hook_name: str) -> bool:    base_method = getattr(MiddlewareBase, hook_name, None)    sub_method = getattr(type(self), hook_name, None)    return base_method is not sub_method

子类没重写时，type(self).on_reply 解析到的就是基类的那个方法对象（MRO 透传），is 比较返回 False；重写了就是子类自己的方法对象，返回 True。不依赖异常，不依赖标记属性，纯靠 Python 方法解析机制的副产物。 构造时一次性过滤（_agent.py L170-188），运行时只遍历已经分好桶的列表。

代价是：如果你用 functools.wraps 之类装饰器包了钩子方法，方法对象身份可能变，需要小心。但框架自带的中间件都没踩这个坑。

3.3 execute_chain：递归闭包，和那个能改参数的小机关

Agent 里 6 个钩子入口用了同一套模板搭洋葱链。以 _reply 为例（_agent.py L570-600）：

async def execute_chain(index=0, inputs=inputs):    if index >= len(self._reply_middlewares):        async for item in self._reply_impl(inputs=inputs):            yield item    else:        mw = self._reply_middlewares[index]        input_kwargs = {"inputs": inputs}        async def next_handler(**kwargs):            async for item in execute_chain(index + 1, **{**input_kwargs, **kwargs}):                yield item        async for item in mw.on_reply(            agent=self, input_kwargs=input_kwargs, next_handler=next_handler,        ):            yield item

递归 + 闭包。next_handler 里递归调 execute_chain(index + 1)，链就串起来了。中间件列表为空时直接进 _reply_impl，零开销。

但这套模板里最值得玩味的是 next_handler 的签名和那一行展开：

async for item in execute_chain(index + 1, **{**input_kwargs, **kwargs}):

**kwargs 是中间件调 next_handler 时传进来的。{**input_kwargs, **kwargs} 让中间件传的参数覆盖原来的 input_kwargs。这意味着中间件不只是"包裹"，还能改写传给下游的参数。

这个能力最直接的用法在 ReplyBudgetControlMiddleware.on_reasoning（_budget.py L204）：

input_kwargs["tool_choice"] = ToolChoice(mode="none")async for event in next_handler(**input_kwargs):    yield event

预算耗尽时，它把 tool_choice 强制改成 none 再交给下游——下游的模型调用因此不会触发任何工具，Agent 只能做最后一轮总结。整个机制不需要 Agent 协作，中间件单方面就能改流程参数。这是洋葱模式比"纯 before/after 钩子"强的地方。

3.4 on_model_call 的返回类型为什么是联合类型

5 个洋葱钩子里，4 个返回 AsyncGenerator（它们在流式产出事件）。唯独 on_model_call 返回 ChatResponse | AsyncGenerator[ChatResponse, None]——因为模型调用本身就有两种形态：非流式一次性返回 ChatResponse，流式持续吐 chunk。

这给中间件作者添了点麻烦：得同时处理两种返回。TracingMiddleware 的处理值得抄——它遇到 AsyncGenerator 时，不急着关 span，而是用 _trace_async_generator_wrapper 包一层，在生成器耗尽（最后一个 chunk 消费完）后才记录响应属性并关 span（_trace.py L85-L108）。过早关 span 会丢掉响应的 token 数、finish_reason 这些属性。

这个细节暴露了一个更普遍的问题：流式场景下，"调用结束"这个时间点被拉长了。任何要在"模型调用后"做事的中间件都得面对这个拉长。

3.5 中间件无状态 vs 状态外置

中间件能被多个 Agent 共享（一个 Tracing 中间件实例挂到 10 个 Agent 上），前提是实例上不能存 per-agent 的运行时状态。但很多中间件确实需要状态（预算计数、检索缓存、检索任务句柄）。AgentScope 的解法是统一外置到 agent.state.middle_context：

agent.state.middle_context[middleware_key][reply_id] = 任意状态

middleware_key 由 get_middleware_key() 给（默认是类名，可重写），reply_id 是当前 reply 的标识。这样状态跟着 Agent 走、跟着 reply 隔离，中间件实例本身干干净净。ReplyBudgetControlMiddleware 是教科书式的实现（_budget.py L128-146）：

• ReplyStartEvent
   → 初始化 middle_context[key][reply_id] = 0
• ModelCallEndEvent
   → += input_weight * input_tokens + output_weight * output_tokens
• ReplyEndEvent
   → pop(reply_id)

状态存进 agent.state 还有个意外好处：HITL 中断后状态天然续上。reply 被人机协同打断、过一会再恢复时，middle_context 里的预算计数还在，不会重置。

但不是所有中间件都遵守这个约束。RAGMiddleware 把 _cached_inputs 存在实例上（_rag.py L592），AgenticMemoryMiddleware 把 _cached_input 和 _retrieval_task 也存实例上（_agentic_memory/_middleware.py L328-332）。源码注释承认这是有意取舍——在多 Agent 并发共享同一中间件实例时会有竞争风险。如果你要复用这两个中间件，记住别让多个 Agent 同时共用同一个实例。

四、跟着我读源码

阅读索引

第一次读只需精读 MiddlewareBase 钩子表 一张，其余字段表（Budget 构造参数、RAG Parameters、AgenticMemory Parameters）按需查阅，跳过不影响理解主线。

MiddlewareBase（核心，必读）

MiddlewareBase（_base.py）是整个模块的入口，定义了 6 个钩子 + 3 个辅助方法。这张表读透了，后面所有中间件都是对它的填空。

5 个洋葱模式钩子（签名统一：agent + input_kwargs + next_handler）：

1 个变换器模式钩子：

3 个辅助方法：

on_acting 的文档注释（_base.py L114-L158）特别强调了一个边界：它只包裹 toolkit.call_tool 这一层纯 I/O 执行，权限检查、输入校验、上下文写入都在 Agent 层、不在这个钩子内。这不是随手写的注释——它是"on_acting 能被安全 offload 到后台任务"的前提：因为这个钩子里不会修改 agent 上下文，把它丢到后台跑不会引发并发污染（除了 is_state_injected=True 的工具，那条注释也警告了）。

ReplyBudgetControlMiddleware（按需查阅）

_budget.py，208 行，是理解洋葱中间件最好的起点——同时实现 on_reply（管状态生命周期）和 on_reasoning（管预算执行），代码量又小。

构造参数：

加权 cost 的算法就一行：input_weight * input_tokens + output_weight * output_tokens。输出通常比输入贵，所以 output_token_weight 默认给个比 1 大的值是常见用法。

执行逻辑分两个钩子：

on_reply（L98-148）只做状态生命周期——监听三种事件，初始化/累加/清除 middle_context 里的计数。它自己不判断预算，只负责把账记对。

on_reasoning（L150-207）才是预算执行的地方。每轮推理前读累计消耗，超了就做两件事：往上下文末尾的 assistant 消息里追加一个 HintBlock（"你该收尾了，别再调工具"），再把 tool_choice 改成 ToolChoice(mode="none")。然后照常调 next_handler。

💡 容易看漏的设计

预算检查为什么放在 on_reasoning 而不是 on_model_call？因为预算耗尽后，目标不是"阻断模型调用"，而是"让 Agent 做最后一轮总结推理"——带着 hint 提示、禁用工具，再调一次模型让它收尾。如果在 on_model_call 层阻断，Agent 连总结的机会都没有，reply 会戛然而止。on_reasoning 这一层刚好能"改参数 + 放行"，正合适。

RAGMiddleware（按需查阅）

_rag.py，766 行，本模块体量最大。一个类同时支持两种检索模式，靠 mode 参数切换。

Parameters 字段表（RAGMiddleware.Parameters，_rag.py L478-568）：

hint_template 用了 SkipJsonSchema（L540），不出现在暴露给前端的配置 schema 里。源码注释直说原因：让前端用户改提示模板会导致 "session-by-session prompt drift"。还配了个 field_validator（L555-568）强校验模板里 {context} 占位符有且仅有 1 个——多了会重复内容，少了会丢内容，_wrap_hint 是按第一次出现做 partition 的。

agentic 模式很直接：list_tools() 返回一个 _SearchKnowledgeTool（L105-299），Agent 自己决定何时检索。这个工具标了 is_read_only=True、is_concurrency_safe=True，check_permissions 直接返回 ALLOW（L217-238）——检索是只读的，没必要走权限栅栏。工具的输入 schema 里 knowledge_bases 字段的 enum 被动态收窄成当前实际挂载的知识库名（L196-215），LLM 没法编造不存在的库名。

static 模式复杂得多，靠 on_reply 和 on_reasoning 两个钩子接力（L620-765）：

1. on_reply
    进来时，把用户原始输入 deepcopy 后缓存到 self._cached_inputs（带说话人名前缀）
2. on_reasoning
    在 agent.state.cur_iter == 0（每轮 reply 的首轮推理）时，用缓存的输入做跨知识库搜索，命中了就 append_context 注入 HintBlock，可选发 HintBlockEvent
3. on_reasoning
    的 finally 里，如果 persist_hint=False，按 block id 精确移除刚注入的 HintBlock

为什么要缓存输入？因为到 on_reasoning 执行时，Agent 已经把用户输入消费进 state.context 了，原始 query 取不回来。必须在外层 on_reply 先存一份。

为什么要按 block id 精确移除？因为 append_context 可能把你的 HintBlock 挂到一条已有的 carrier 消息上，那条消息上可能还有别的中间件加的 block。按 id 删只动自己的，不误伤。

跨知识库搜索（_search_across，L308-360）用 asyncio.gather 并发查所有知识库，拍平后按 score 降序、截 top_k。源码注释坦白了一个已知妥协（L323-327）：不同知识库用不同 embedding 模型时，score 量纲不一致，直接比大小不严谨。需要严谨的应该换 RRF 之类的 rank 融合。

TracingMiddleware

_tracing/_trace.py，349 行。实现了 on_reply/on_model_call/on_acting 三个钩子（注意它没实现on_reasoning，所以不会出现在 reasoning 链上）。把 OpenTelemetry span 挂到 reply、模型调用、工具调用三个层级。

最值得抄的是零开销短路。每个钩子开头都调 _check_tracing_enabled()（L58-69）：检查当前 tracer provider 是不是真的 SDK TracerProvider，如果只是默认的 no-op proxy（即用户没调过 setup_tracing），直接透传 next_handler，不做任何 span 操作。这比"用配置决定挂不挂中间件"灵活——中间件始终挂着，只有真正初始化了 tracing 才生效，省去了条件装配的麻烦。

on_reply（L136-241）对 HITL 场景的处理很细：reply 过程中如果出了 RequireUserConfirmEvent（等用户确认工具）或 RequireExternalExecutionEvent（等外部执行），finally 块把这些待处理工具名记到 span 属性 AGENTSCOPE_HITL_PENDING_TOOLS / AGENTSCOPE_EXTERNAL_EXECUTION_PENDING_TOOLS。入口处如果发现输入是 ExternalExecutionResultEvent，还会为每个外部执行的 tool 补一个合成 execute_tool span，标 AGENTSCOPE_IS_EXTERNAL_EXECUTION=True——让链路图上能看到这些"在框架外跑完"的工具。

on_model_call（L246-294）的流式处理呼应了 3.4 节：返回 AsyncGenerator 时用 _trace_async_generator_wrapper 包裹，耗尽后才关 span。

辅助文件：_extractor.py（609 行）从 AgentScope 内部对象提取 OTel 标准属性；_attributes.py 定义 SpanAttributes 和 OperationNameValues，直接引用 OTel GenAI 语义约定；_converter.py 把 ContentBlock 转 OTel GenAI parts 格式。除非要改 Tracing 行为，这三个文件可以跳过。

TTSMiddleware

_tts_middleware.py，180 行，只实现 on_reply。把推理产出的文本事件转成语音，再以 DATA_BLOCK_* 事件注入流里。

两种模式差在推送时机（L63-117）：

• 非实时
  （tts.realtime=False）：攒文本到 TextBlockEndEvent，把整段一次喂给 tts.synthesize(text)，拿到完整音频再注入
• 实时
  （tts.realtime=True）：每来一个 TextBlockDeltaEvent 就 tts.push(delta)，模型边收文本边吐音频，立即注入

注入的事件序列统一是 DataBlockStartEvent → N × DataBlockDeltaEvent → DataBlockEndEvent，每个 delta 携带增量 base64 PCM，拼起来才是完整音频。async with self.tts 包住整个 reply，管 TTS 模型的会话生命周期。

长期记忆三件套

_longterm_memory/ 下三个中间件，共享同一套控制模式抽象（static_control / agent_control / both），但记忆后端各异。

AgenticMemoryMiddleware（803 行）走文件系统路线，思路最特别：Agent 用 Write 工具直接在 Memory/ 目录下写 Markdown 文件，每个文件带 frontmatter（name/description/type，type 分 user/feedback/project/reference 四类）。MEMORY.md 是索引文件，被注入系统提示词。检索时用一个 LLM 从 frontmatter 列表里挑相关文件（最多 5 个），再读文件内容注入。它用了三个钩子联动：on_system_prompt 注入记忆指令 + 截断后的 MEMORY.md；on_reply 缓存用户输入 + 启动异步检索任务；on_reasoning 轮询检索任务、完成则注入 HintBlock。

Parameters 关键字段（_agentic_memory/_middleware.py L219-297）：memory_max_tokens=4000（MEMORY.md 注入上限）、retrieval_async=True（是否异步检索）、retrieval_model=None（检索用 LLM，None 则用 Agent 的模型）、retrieval_max_tokens_per_md=2000（单文件读取上限，防淹没上下文）。

它还有个值得注意的防幻觉设计（L606-609）：LLM 选出来的文件名要用 {h.filename for h in headers} 过一遍，丢掉 LLM 编造的不存在的文件名，再截前 5 个。

Mem0Middleware（741 行）对接 mem0（开源 AsyncMemory 或托管 AsyncMemoryClient）。两种构造路径：传 AgentScope 的 chat_model+embedding_model 让中间件内部自建客户端，或直接传预构建的 client。mode="static_control" 时在 ReplyStartEvent 后注入检索到的记忆、reply 结束后异步写回对话；mode="agent_control" 时暴露 search_memory/add_memory 工具让 Agent 自主调用；"both" 两者都开。user_id 必填，scope_search_by_agent=True 时记忆按 user_id + agent_id 隔离。

ReMeMiddleware（564 行）嵌入 ReMe 应用（in-process，无独立服务）。记忆写入由 ReMe 的 auto_memory job 自动完成，Agent 不参与写入，只有检索工具。它的 chat_model 在构造时固定（不从 agent 取），所以一个中间件实例被多个 Agent 共享时，底层用哪个 LLM 是明确的；per-conversation 的 session_id 则在钩子执行时实时从 agent 读、不存实例上，避免共享实例时串扰。

五、一条消息的完整调用链

假设 Agent 挂了 RAGMiddleware（static 模式）和 ReplyBudgetControlMiddleware，用户发一条消息，跟着它走完全程。选这两个是因为它们恰好覆盖了 on_reply + on_reasoning 联动和参数改写两个最有意思的机制。

图 2 · 5 条链合一的调用时序：玫红色高亮 Budget 改写 tool_choice 那层，橙色虚线是 ModelCallEndEvent 跨层冒泡回外层 on_reply 记账

上图把这次调用经过的 5 条链（on_reply / on_reasoning / on_model_call / on_acting / on_system_prompt）合在一张时序图里，玫红色高亮的是 Budget 改写 tool_choice 的那一层，橙色虚线是 ModelCallEndEvent 跨层冒泡回外层 on_reply 记账的路径。下面逐段说明每一层做了什么。

入口：on_reply 洋葱链

用户调 agent.reply_stream(inputs=Msg(...))，进 self._reply(inputs=inputs)（_agent.py L560-603）。两个中间件都实现了 on_reply，按注册顺序串成链：

RAGMiddleware.on_reply        （先缓存输入）  → ReplyBudgetControlMiddleware.on_reply  （准备记账）    → Agent._reply_impl       （真正跑 ReAct）

RAGMiddleware.on_reply（_rag.py L620-680）：从 input_kwargs["inputs"] 取出用户消息，deepcopy 后给每个消息的文本块前缀加上说话人名，缓存到 self._cached_inputs。然后 try 块里 async for evt in next_handler(**input_kwargs) 把事件流原样向上传。finally 里清掉缓存。

ReplyBudgetControlMiddleware.on_reply（_budget.py L98-148）：调 next_handler 进入下一层，同时监听事件流——ReplyStartEvent 到了就在 middle_context[key][reply_id] 初始化 0，ModelCallEndEvent 到了就累加 cost，ReplyEndEvent 到了就清掉这条 reply 的账。事件本身照样 yield 给上游，它只"顺手记账"。

Agent._reply_impl（_agent.py L664-914）：处理输入消息 → 发 ReplyStartEvent → 进入 ReAct 循环。

注意一个跨层通信：ModelCallEndEvent 是在最内层 _reasoning_impl → _call_model 之后 yield 出来的，它一路冒泡穿过 Budget 中间件的 on_reply。Budget 的 on_reasoning 不需要知道这个事件——账是在外层 on_reply 记的。洋葱链各层通过事件流隐式通信，互不感知。

每轮推理：on_reasoning 洋葱链

ReAct 循环每轮调 self._reasoning()（_agent.py L935-971）。两个中间件都实现了 on_reasoning：

RAGMiddleware.on_reasoning    （首轮注入检索结果）  → ReplyBudgetControlMiddleware.on_reasoning  （检查预算）    → Agent._reasoning_impl   （准备输入、调模型）

RAGMiddleware.on_reasoning（_rag.py L682-765）：只在 cur_iter == 0（首轮）干活。用之前缓存的 _cached_inputs 调 _search_across 跨知识库搜索，命中后 append_context 注入 HintBlock，可选发 HintBlockEvent。finally 里若 persist_hint=False，反向扫 state.context 找到刚注入的 block 按 id 移除。

ReplyBudgetControlMiddleware.on_reasoning（_budget.py L150-207）：读 middle_context[key][reply_id] 的累计 cost。超预算了就往末尾 assistant 消息追加 HintBlock（"该收尾了"），把 input_kwargs["tool_choice"] 改成 ToolChoice(mode="none")。然后 async for event in next_handler(**input_kwargs) 把改过的参数传下去。

这里能清楚看到 3.3 节那个"改参数机关"的实际效果：Budget 改了 tool_choice，下游 _reasoning_impl → _call_model 收到的就是 none，模型这一轮不会产工具调用，Agent 只能输出总结文本。

Agent._reasoning_impl（_agent.py L973-1010+）：_prepare_model_input 组装 messages + tools，调 _call_model。

模型调用：on_model_call 洋葱链

这个例子里没挂实现 on_model_call 的中间件（Tracing 没挂），所以 _call_model（_agent.py L2310-2402）检测到 _model_call_middlewares 为空，直接调model(...)，跳过整个 execute_chain。这就是"空列表零开销"的体现。

如果挂了 Tracing，链会是 TracingMiddleware.on_model_call → model(...)，Tracing 开 span、调 next_handler、根据返回是 ChatResponse 还是 AsyncGenerator 决定立即关 span 还是包一层延迟关。

工具执行：on_acting 洋葱链

模型这一轮若产了工具调用（预算没超的情况下），Agent 对每个工具调 self._acting(tool_call)（_agent.py L1820-1868）。这个例子里也没挂 on_acting 中间件，直接进 _acting_impl → toolkit.call_tool。挂了 Tracing 的话，会在外面包一层 execute_tool {tool_name} span。

注意：权限检查、ToolResultBlock 写回上下文，这些都在 _execute_tool_call（on_acting 钩子的外面）做。on_acting 拿到的 tool_call 是已经过权限和校验的。这是 3.1 节那个"边界设计"的具体体现。

系统提示词：on_system_prompt 变换器链

每次推理前 _get_system_prompt（_agent.py L2274-2276）组装提示词时，顺序跑变换器链：

for mw in self._system_prompt_middlewares:    result = await mw.on_system_prompt(self, result)

没有 next_handler，就是 for 循环。AgenticMemoryMiddleware 会在这里把"记忆使用说明 + 截断后的 MEMORY.md"追加到提示词末尾。

六、如何开始调试源码

第一断点：Agent.__init__ 的分桶处（_agent.py L170-188）。

观察 6 个 _*_middlewares 列表分别装了哪些中间件实例。这是确认"我以为挂了的中间件到底进了哪条链"的最快方式——很多时候中间件不生效，就是因为 is_implemented 判定它没实现那个钩子（比如方法名拼错、装饰器干扰了方法对象身份比较）。

第二断点：_reply 里 execute_chain 的首次调用（_agent.py L602 附近）。

观察 self._reply_middlewares 的顺序。洋葱链的顺序就是列表顺序，最外层先执行——这直接影响"谁包谁"。Budget 和 RAG 谁在外层，决定了 RAG 缓存输入时看到的是不是原始 inputs。

第三断点：ReplyBudgetControlMiddleware.on_reasoning 的预算判断（_budget.py L187，if used >= self.token_budget）。

观察 used、self.token_budget、agent.state.middle_context 的内容。能看到预算是否触发、HintBlock 是否正确注入、tool_choice 是否被改写。

第四断点：RAGMiddleware.on_reasoning 的注入处（_rag.py L738，if blocks:）。

观察 blocks（格式化后的检索结果）、hint 对象、agent.state.context 在 append_context 前后的变化。若 persist_hint=False，单步进 finally（L756-765）确认 block 被按 id 精确移除，没误删别的。

七、如何扩展这个模块

应该改（推荐路径）

加一个新中间件：继承 MiddlewareBase，只实现你需要的钩子。is_implemented 自动检测，Agent 自动把你挂到对应链。不改基类、不改 Agent。这是 99% 的场景。

改预算触发行为：重写 ReplyBudgetControlMiddleware.on_reasoning 的 if used >= self.token_budget 分支。比如做渐进式（先警告 80%、再硬停 100%），只动这个分支。

加新的 RAG 模式：在 RAGMiddleware.Parameters.mode 的 Literal 加值，在 on_reply/on_reasoning 加分支。比如加个"只在用户问问题时检索"的模式。

改 Tracing 属性：改 _extractor.py 的提取函数，或扩 _attributes.py 的 SpanAttributes 常量。

不应该改（有更优替代）

不要改 MiddlewareBase 的钩子签名。所有中间件都依赖它。要传新参数，走 input_kwargs。

不要在中间件里直接改 agent.state.context 里已有的消息内容（除非明确知道后果）。多个中间件可能同时操作上下文。RAGMiddleware 的 persist_hint=False 按 block id 精确移除自己的产物，这个模式可以抄——只动自己注入的 block，不碰别人的。

不要改 execute_chain 模板。_agent.py 里 6 处入口的 execute_chain 是高度对称的模板，改一个漏另外五个会埋一致性 bug。

千万不要改

⚠️ 动了会破坏不变量

is_implemented 的检测逻辑（_base.py L52-63）。它依赖方法对象身份比较，改成异常检测或标记属性，所有中间件的注册都会崩。

on_acting 的边界（_base.py L114-158 注释明确声明）。它只包 toolkit.call_tool，权限和上下文写入在外面。在这个钩子里做权限逻辑，会破坏"可安全 offload 到后台任务"这个不变量——工具执行被设计成无副作用的纯 I/O 层，正是为了能并发。

八、本模块最值得学习的设计

把定制赶出核心类的整体赌注。这是最值得带走的。AgentScope 押注"Agent 类保持单类零继承、所有变化外置到中间件"，换来的是核心 ReAct 循环的长期稳定——加新能力永远不改 Agent 源码，只加中间件。这个赌注的代价是 6 个钩子要选得准（选少了不够用，选多了框架变重），但收益是核心类不被业务撑爆。任何要长期演进的核心框架都面临这个选择。

6 个拦截点的粒度选择。不是 1 个万能钩子，也不是每个方法一个钩子。选点依据是横切需求的自然分布：reply 管生命周期，reasoning 管推理，model_call 管模型 IO，acting 管工具执行，compress_context 管压缩，system_prompt 管提示词增强。这组切分可以原样搬到任何带固定主循环的系统。

两种模式覆盖两种语义。需要"调用前 + 调用后"的用洋葱模式，需要"输入 → 输出"的用变换器模式。on_system_prompt 没硬套 next_handler，是对"别为了统一而统一"的好示范。

is_implemented 的检测技巧。用方法对象身份比较做"是否重写"判定，构造时分桶、运行时零反射。比异常控制流和装饰器注册都干净。注意它的脆弱点（装饰器包装会改身份），但框架自带中间件都没踩。

状态外置到 middle_context。中间件实例无状态 → 可跨 Agent 共享；状态存在 agent.state → HITL 中断后自然续上。一个设计同时解决两个问题。

execute_chain 里的参数改写机关。next_handler(**{**input_kwargs, **kwargs}) 让中间件能改写下游参数。这让洋葱模式不只是"包裹"，还能"改流程"。预算控制改 tool_choice 就是靠它。

on_acting 的边界自觉。只包纯 I/O，把权限和上下文写入挡在外面，换来了可后台 offload 的并发安全。这种"为了一个特性，刻意收窄钩子职责"的自觉，比"钩子包得越多越好"的设计成熟得多。

如果重新设计，有两点可以再打磨。一是 RAGMiddleware 和 AgenticMemoryMiddleware 把状态（_cached_inputs、_retrieval_task）存在实例上而非 middle_context，违反了自己定的"实例无状态"约定，多 Agent 共享时有竞争风险——要么补一份"这些中间件不可跨 Agent 共享"的明确文档，要么把状态也外置。二是洋葱链的错误处理目前是直接冒泡，辅助性中间件（如 Tracing）抛异常会打断整个 reply，框架层面或许该提供"辅助中间件静默降级"的选项。

九、阅读建议

1. _base.py（251 行）
   。6 个钩子的签名、文档、is_implemented，全在这。是读其它一切的前提。
2. _budget.py（208 行）
   。跟着 on_reply 和 on_reasoning 走一遍，状态外置 + 事件拦截 + 参数改写一次看全。最小的完整范例。
3. _agent.py 的链构建
   （_reply L560-603、_reasoning L935-971、_acting L1820-1868、_call_model L2310-2402、_get_system_prompt L2274-2276、compress_context L288-325）。看 6 处入口怎么用同一个 execute_chain 模板搭链，以及 on_system_prompt 为什么只用 for 循环。
4. _rag.py（766 行，选读）
   。重点看 static 模式的 on_reply+on_reasoning 联动、_cached_inputs 为何必要、persist_hint=False 的按 id 清理。
5. _tracing/_trace.py（349 行，选读）
   。重点看零开销短路、流式 span 的延迟关闭、HITL 属性记录。
6. _longterm_memory/（按需）
   。要做长期记忆相关功能，挑一个最接近你场景的深入读。三件套的控制模式抽象（static/agent/both）是共享的，读懂一个另两个就好办。

可以跳过：_tracing/_extractor.py、_converter.py、_utils.py 是 OTel 属性提取和序列化的实现细节，除非改 Tracing 否则不用读。

十、读完之后

回到开头那个问题：为什么 AgentScope 只有一个 Agent 类？因为所有定制都被赶到了 Middleware 里。Agent 核心做最小闭环（ReAct），中间件在 6 个拦截点上叠加能力。两种模式（洋葱 + 变换器）覆盖了"包裹事件流"和"转换字符串"两类需求。is_implemented 做自动发现，execute_chain 做链式调用，middle_context 做状态外置——三件事撑起了整个机制。

读完这个模块，真正值得带走的不是"AgentScope 有哪些中间件"，而是这套核心不可变、变化全外置的框架设计范式。它对 Agent 框架有效，对任何要长期演进、要扛住业务定制的核心系统都有效：先把主循环的横切需求分布摸清楚，在自然边界开拦截点；需要包裹的用洋葱，需要转换的用变换器；状态外置、实例无状态；宁可多设计几个 hook，也不要让业务把核心类继承成树。

留给你的问题

低门槛：你写 Agent 应用时，有没有遇到过"想加个日志/预算/检索，结果不得不动框架核心代码"的时候？你当时是忍着改了，还是想办法绕过去了？

进阶（可选）：如果同时挂 RAG（static 模式）+ Budget + Tracing 三个中间件，你会怎么排它们的注册顺序？换个顺序，行为会不一样吗？（提示：想想 on_reasoning 链上谁先执行——Budget 改了 tool_choice=none 之后，RAG 的检索结果注入还有意义吗？顺序错了会浪费一次 embedding。）

觉得有用？点个「在看」或转发给同样在搞 Agent 框架的朋友。系列持续更新，关注不迷路。

附录：源码元信息

Repository：https://github.com/agentscope-ai/agentscope.git

Branch：v2.0.4