数据截至 2026-08-14(上游 commit 3a4e2ae3eec0)

第 4 章 · 中间件、模型与格式化

这一章讲：怎么不改源码改 agent 行为、模型调用失败了怎么办、以及统一的 Msg 怎么变成八家 API 各自认的 JSON。

4.1 中间件：七个挂载点

挂在哪

MiddlewareBase（src/agentscope/middleware/_base.py:13）定义七个钩子，其中六个是洋葱型（能写前置和后置逻辑），一个是变换型（顺序管道）：

钩子	包住什么	类型
`on_reply`	整个 reply	洋葱
`on_reasoning`	一次推理（含模型调用与事件转换）	洋葱
`on_check_permission`	一次权限判定	洋葱
`on_acting`	仅 `toolkit.call_tool` 这一层原始执行	洋葱
`on_model_call`	最内层的模型 API 调用	洋葱
`on_compress_context`	上下文压缩	洋葱
`on_system_prompt`	系统提示词字符串	变换（串行）

嵌套关系：

 on_reply
   └── on_reasoning
         └── on_model_call ──► 真实 API
   └── (行动阶段)
         └── on_check_permission ──► 权限引擎
         └── on_acting ──────────► toolkit.call_tool

`on_acting` 的边界画得很干净

它只包住 toolkit.call_tool，不包括权限检查、入参校验和上下文写入（src/agentscope/agent/_agent.py:2580 的 _acting_impl，函数体只有两行）。docstring 解释了动机：

This separation makes it safe to offload the next_handler coroutine to a background task: it will never mutate agent context on its own.

翻译：因为这一层不碰状态，所以中间件可以放心地把它扔进后台任务——这正是「长时间工具转后台跑，跑完再唤醒 agent」这个功能的地基。

同一段 docstring 里也诚实标了 TODO：注入了 agent.state 的工具（is_state_injected=True）扔后台会有并发改状态的风险，目前还没拦（:2443-2447）。

只在启动时过滤一次

Agent.__init__ 里按钩子分了七个列表（src/agentscope/agent/_agent.py:190-213）：

self._reply_middlewares = [
    _ for _ in middlewares if _.is_implemented("on_reply")
]

is_implemented（src/agentscope/middleware/_base.py:55）就是比较子类方法和基类方法是不是同一个对象。分类只做一次，运行时零开销；没有中间件时还有专门的短路分支，直接调 _impl。

洋葱链怎么写

六个洋葱钩子用的是同一个模板（以 _reasoning 为例，src/agentscope/agent/_agent.py:1466-1494）：

# 示意，非源码：洋葱模板的骨架
async def execute_chain(index=0, **kwargs):
    if index >= len(middlewares):
        async for item in impl(**kwargs):      # 最内层：真实实现
            yield item
    else:
        mw = middlewares[index]
        async def next_handler(**kw):          # 交给中间件的"下一层"
            async for item in execute_chain(index + 1, **{**kwargs, **kw}):
                yield item
        async for item in mw.on_reasoning(agent, kwargs, next_handler):
            yield item

重点看 {**input_kwargs, **kwargs} 这个合并：中间件调 next_handler() 时可以只传想改的参数，没传的沿用原值。所以「只想改 tool_choice」写成 next_handler(tool_choice=ToolChoice(mode="none")) 就行。

权限钩子多做一件事

_check_permission（:1998）在进链前深拷贝了 tool_call 和 tool_input（:2031-2033）：

# Copy so middleware cannot mutate what the agent consumes later.
tool_call = deepcopy(tool_call)
tool_input = deepcopy(tool_input)

即中间件看到的是副本，改了不影响真正执行的那次调用。要改入参得走 PermissionDecision.updated_input。

内置中间件清单

中间件	挂在哪	干什么
`ReplyBudgetControlMiddleware`	`on_reply` + `on_reasoning`	加权 token 预算，超了就注入「收尾」提示并强制 `tool_choice="none"`
`RAGMiddleware`	`on_reply` + `on_reasoning`	注入检索工具与检索结果
`AgenticMemoryMiddleware` / `Mem0Middleware` / `ReMeMiddleware`	长期记忆	三种可切换后端
`TracingMiddleware`	多点	OpenTelemetry 埋点
`TTSMiddleware`	—	文本转语音

ReplyBudgetControlMiddleware（src/agentscope/middleware/_budget.py:21）有个值得学的细节：它自己完全无状态，所有运行时数据放在 agent.state.middle_context 里。这样一个实例能被多个 agent 共享，而且预算状态能跟着 agent 状态一起存盘、跨人工确认的挂起恢复。

4.2 模型层：两级重试

两个 max_retries 不是一回事

容易混：ChatModelBase 和 ModelConfig 都有 max_retries。

层	字段	默认	重试什么
模型内	`ChatModelBase.max_retries`	3	只重试 `_get_retryable_exceptions()` 声明的异常，中间 sleep
agent 级	`ModelConfig.max_retries`	0	重试任何异常，用完就换 `fallback_model`

agent 级默认 0 是刻意的，字段描述写明了理由："Defaults to 0 to avoid compounding with the model's own inner retry loop."（src/agentscope/agent/_config.py:357-367）——两层都重试会指数级放大等待时间。

降级流程

_call_model（src/agentscope/agent/_agent.py:3029）：

 models = [主模型] + ([备用模型] 若配置了)
   │
   for model in models:
       for attempt in range(max_retries + 1):
           试着调用（经过 on_model_call 中间件链）
           成功 ─► 直接返回
           失败 ─► 记下异常，继续
   │
   全挂 ─► 抛最后一个异常

流式响应的累加器

ChatModelBase.__call__（src/agentscope/model/_base.py:182）包了一层 _stream()，做三件事：

累加所有增量块，最后补一个 is_last=True 的完整响应（向后兼容）；
吞掉空内容的"载体块"——OpenAI 兼容 API 会在末尾发一个只带 usage、没有 choices 的块，累加器吸收它的元数据但不往外吐（:246-255，注释里管这类块叫 carrier chunk）；
被取消时把累加结果标成 INTERRUPTED 再吐出去，而不是直接抛。

第 2 点是那种「不做也能跑，做了下游干净很多」的打磨。

消费侧的对称处理

_reasoning_impl（src/agentscope/agent/_agent.py:1497）用 inspect.isasyncgen(res) 区分流式和非流式，走同一条事件转换函数。流式跑完但没收到 is_last=True 的块会抛错（:1446-1453），错误信息把可能原因都列了：网络断、超时、模型 bug。

只有 thinking 的响应不算结束

一个细腻的判断（:1472-1478）：

has_only_thinking_blocks = bool(completed_response.content) and all(
    isinstance(block, ThinkingBlock)
    for block in completed_response.content
)

推理模型有时只吐思考、不吐正文。如果按「没有 tool_call 就是最终答案」判定，会把思考当成回答返回给用户。加上这个条件后，循环继续转，等模型真的说话。

4.3 Formatter：一份 Msg，八种方言

位置

Formatter 挂在模型对象上，不是 agent 上。OpenAIChatModel.__init__ 默认 formatter or OpenAIChatFormatter()（src/agentscope/model/_openai_chat/_model.py:163），调用时 formatted_messages = await self.formatter.format(messages)（:216）。

两种 Formatter

每家 API 都有两个类：

类型	场景	做法
`XxxChatFormatter`	只有用户和一个 agent	逐条按 role 翻译，用 `name` 字段区分身份
`XxxMultiAgentFormatter`	多个 agent 在同一个会话	把非工具消息压成一段带 `<history>` 标签的文本

多智能体格式化的思路

问题：OpenAI 的 messages 只有 user / assistant / system 三种 role。三个 agent 说话，谁是 assistant？

OpenAIMultiAgentFormatter（src/agentscope/formatter/_openai_formatter.py:428）的答案是放弃用 role 表达身份，改用文本：

# Conversation History
The content between <history></history> tags contains your conversation history
<history>
Alice: 我觉得应该先读一遍代码
Bob: 同意，我去看 parser 那块
</history>

整段作为一条 user 消息。

但工具序列不能这么压

工具调用/结果必须保持 API 原生结构（tool_calls + role: "tool"），否则模型认不出。所以先分组：

_group_messages（src/agentscope/formatter/_formatter_base.py:176）把消息切成交替的两类段：

消息序列:  A说  B说  [调用工具 结果]  A说  [调用工具 结果]
分组:      └─agent_message─┘ └tool_seq┘ └agent┘ └tool_seq┘
              压成 history      原样        压     原样

_format_tool_sequence（src/agentscope/formatter/_openai_formatter.py:490）直接委托给单 agent 版的 formatter 处理。同一份逻辑复用，不重写。

还有个小细节：conversation_history_prompt 那段说明只在第一个 agent_message 段前加一次（_format_agent_message 的 is_first 参数），避免重复刷屏。

单 agent formatter 的"冲刷"逻辑

OpenAIChatFormatter.format（:203）里有两处「遇到某类块就先把攒的内容发出去」的冲刷（flush）：碰到 HintBlock 和碰到 ToolResultBlock 时。

原因是这两类块要变成独立的消息（hint → user 消息，tool_result → tool 消息），而 AgentScope 把一次 reply 的所有块塞在同一条 Msg 里。所以格式化时要按块类型把一条消息拆成多条 API 消息。

各家的差异点

差异	处理
OpenAI 不吃 thinking	静默跳过（`:337-340`）
图片可以给远程 URL	直接透传；`file://` 才读盘转 base64（`:96-108`）
音频必须 base64	远程 URL 也要下载后编码，且只支持 wav/mp3（`:117-186`）
工具结果里的多模态	转成紧随其后的一条 `name="system-reminder"` 的 user 消息（`:313-330`）

最后一条是通用难题的通用解：tool 消息装不下图片，只能把图片挪到下一条 user 消息里。

4.4 凭据分离

模型不直接吃 api_key，而是吃 CredentialBase（src/agentscope/credential/）：

DashScopeChatModel(
    credential=DashScopeCredential(api_key=os.environ["DASHSCOPE_API_KEY"]),
    model="qwen3.6-plus",
)

多一层看着啰嗦，但线上服务要按租户注入不同凭据、要轮换、要从密钥管理服务动态取，这一层就是必要的。

4.5 代码地图

主题	文件路径	符号名
中间件协议	`src/agentscope/middleware/_base.py`	`MiddlewareBase`、`is_implemented`
洋葱链模板	`src/agentscope/agent/_agent.py`	`_reply`、`_reasoning`、`_check_permission`、`_acting`
最内层实现	`src/agentscope/agent/_agent.py`	`_acting_impl`、`_reasoning_impl`
中间件分类	`src/agentscope/agent/_agent.py`	`Agent.__init__`（`_reply_middlewares` 等七个列表）
预算控制	`src/agentscope/middleware/_budget.py`	`ReplyBudgetControlMiddleware`
RAG 注入	`src/agentscope/middleware/_rag.py`	`RAGMiddleware`、`_SearchKnowledgeTool`
模型基类与重试	`src/agentscope/model/_base.py`	`ChatModelBase.__call__`、`_get_retryable_exceptions`
agent 级降级	`src/agentscope/agent/_agent.py`	`_call_model`
模型配置	`src/agentscope/agent/_config.py`	`ModelConfig`
OpenAI 实现	`src/agentscope/model/_openai_chat/_model.py`	`OpenAIChatModel`
格式化基类与分组	`src/agentscope/formatter/_formatter_base.py`	`FormatterBase`、`_group_messages`、`convert_tool_result_to_string`
单/多 agent 格式化	`src/agentscope/formatter/_openai_formatter.py`	`OpenAIChatFormatter`、`OpenAIMultiAgentFormatter`、`_format_tool_sequence`、`_format_agent_message`
工具中间件	`src/agentscope/tool/_base.py`	`ToolMiddlewareBase`、`ToolBase.__call__`
相关测试	`tests/`	`formatter__test.py`、`middleware__test.py`、`model_base_test.py`

4.1 中间件：七个挂载点​

挂在哪​

on_acting 的边界画得很干净​

只在启动时过滤一次​

洋葱链怎么写​

权限钩子多做一件事​

内置中间件清单​

4.2 模型层：两级重试​

两个 max_retries 不是一回事​

降级流程​

流式响应的累加器​

消费侧的对称处理​

只有 thinking 的响应不算结束​

4.3 Formatter：一份 Msg，八种方言​

位置​

两种 Formatter​

多智能体格式化的思路​

但工具序列不能这么压​

单 agent formatter 的"冲刷"逻辑​

各家的差异点​

4.4 凭据分离​

4.5 代码地图​