中间件

概述

智能体中间件是在不修改智能体或模型代码的前提下，向智能体执行流程中的关键位置注入自定义逻辑（日志、追踪、输入改写、访问控制等）的机制。 AgentScope 暴露了 6 个 hook 位置外加一个工具提供 hook，覆盖了从外层 reply 流程一路下沉到底层模型 API 调用的全链路：

位置	类型	说明
`on_reply`	Onion	包裹一次完整的 reply 流程，覆盖其中所有 ReAct 轮次、工具执行与最终输出
`on_reasoning`	Onion	包裹一轮 ReAct 中的推理步骤（输入组装 → 模型调用 → 流式解码）
`on_acting`	Onion	包裹一次工具调用的执行
`on_model_call`	Onion	包裹一次底层 `ChatModel` API 调用，最贴近模型
`on_compress_context`	Onion	包裹 `Agent.compress_context()` —— 在每次推理步骤前触发，由智能体决定是否压缩上下文
`on_system_prompt`	Transformer	在每次组装 system prompt 时触发；多个中间件串行接力，每一个把上一个的输出再做一次变换
`list_tools`	Tool source	可选。返回中间件贡献的 `list[ToolBase]`。不会被自动调用 —— 由组装智能体 toolkit 的调用方决定是否调用以及如何合并结果。

以上 hook 均作用于智能体层面。若需要在单个工具实例上挂钩 —— 无论该工具是在智能体内部还是外部被调用 —— 请参见工具中间件。

三种类型的差别：

Onion（洋葱式）—— 中间件包裹下一层 handler，可以在 next_handler() 前后插入逻辑、观察中间事件流。
Transformer（变换式）—— 中间件之间串成流水线，前一个的输出作为后一个的输入，不存在「内层」概念。
Tool source（工具来源）—— 不在运行时路径上的 hook。Agent.__init__ 不会调用 list_tools()；需要显式从中间件中收集工具并自行传入 toolkit。

下图展示这些 hook 在智能体生命周期中的嵌套关系。on_system_prompt 嵌入在 on_reasoning 内部，因为它在 reasoning 步骤组装 system prompt 时被触发；on_compress_context 位于每轮 ReAct 顶部，在 reasoning 之前：

on_reply

ReAct loop（每一轮）

on_compress_context（上下文压缩判定）

on_reasoning

on_system_prompt（组装 system prompt）

on_model_call（模型 API 调用）

on_acting（每个工具调用一次）

当前 on_acting 只包裹智能体运行时内部的工具执行；通过 external execution 在智能体外部执行的工具不会被 on_acting 追踪到。

装备中间件

AgentScope 把一组 hook 装在一个类里 —— 同一个中间件类可以同时实现 6 个 hook 位置（外加可选的 list_tools tool-provider hook）中的任意子集。把实例传入 Agent(middlewares=[...]) 即可装备：

from agentscope.agent import Agent
from agentscope.middleware import TracingMiddleware

agent = Agent(
    name="assistant",
    system_prompt="You are a helpful assistant.",
    model=model,
    toolkit=toolkit,
    middlewares=[TracingMiddleware()],
)

构造时智能体会扫描每个中间件实例，按它实际实现了哪些 hook，把它分配到对应位置的执行列表中。未实现的位置自动跳过，不产生任何调用开销。

内置中间件

AgentScope 目前支持如下的中间件实现：

场景	实现	说明
链路追踪	`TracingMiddleware`	提供智能体全生命周期的 OpenTelemetry 追踪
预算控制	`ReplyBudgetControlMiddleware`	控制单次 reply 的 Token 预算
语音生成	`TTSMiddleware`	拦截文本输出并合成语音
模型路由	`ModelRouterMiddleware`	按用户输入为每次回复选择合适的对话模型
长期记忆	`AgenticMemoryMiddleware`， `Mem0Middleware`	提供跨会话的长期记忆实现
RAG	`RAGMiddleware`	提供访问外部知识库的能力

链路追踪

TracingMiddleware 为智能体全生命周期接入 OpenTelemetry 追踪。它在 on_reply、on_model_call、on_acting 三个位置打点，按层级生成 span。使用前先在进程中注册 TracerProvider 与 OTLP 导出器：

from opentelemetry import trace
from opentelemetry.sdk.trace import TracerProvider
from opentelemetry.sdk.trace.export import BatchSpanProcessor
from opentelemetry.exporter.otlp.proto.http.trace_exporter import OTLPSpanExporter

provider = TracerProvider()
provider.add_span_processor(
    BatchSpanProcessor(OTLPSpanExporter(endpoint="http://localhost:4318/v1/traces")),
)
trace.set_tracer_provider(provider)

随后把 TracingMiddleware 装到智能体上即可：

from agentscope.agent import Agent
from agentscope.middleware import TracingMiddleware

agent = Agent(
    name="assistant",
    system_prompt="You are a helpful assistant.",
    model=model,
    toolkit=toolkit,
    middlewares=[TracingMiddleware()],
)

每次 reply 会产出一棵嵌套 span 树，每个层级捕获的关键属性如下：

Agent Reply Span
Model Call Span
Tool Execution Span

来自 on_reply：

Agent 名称、session ID、reply ID
输入消息与最终输出消息
HITL 等待中的工具调用
External execution 等待中的工具调用

来自 on_model_call：

模型名、模型 API、输入 / 输出 token 数
请求与响应的消息内容
对流式响应做包装，在最终 chunk 上写入属性

来自 on_acting：

工具名、调用 ID、入参
工具执行结果

未配置 TracerProvider 时，所有 hook 会直接短路到 next_handler()，不创建 span 也不计算属性 —— 几乎零开销。

Agent 收到 ExternalExecutionResultEvent（外部执行的工具结果）时，TracingMiddleware 会为每条外部执行结果合成一个补偿 span，让外部系统执行的工具也能保留完整可观测性。

添加自定义 Span

如需在智能体生命周期内追踪自定义操作，可直接使用 OpenTelemetry Python SDK。获取一个作用域为 AgentScope 的 tracer，并将目标代码包裹在 span 中：

from opentelemetry import trace
from agentscope import __version__

tracer = trace.get_tracer("agentscope", __version__)

with tracer.start_as_current_span(
    name="your_span_name",
    attributes={
        # 附加到 span 上的可选键值对，
        # 例如函数名、入参或任何自定义元数据。
    },
    end_on_exit=True,
) as span:
    # 你的代码

这些自定义 span 会与 AgentScope 内置的 span 一并上报，发送到 TracerProvider 中配置的同一个 OTLP collector。

预算控制

ReplyBudgetControlMiddleware 为单次 reply 强制一个加权 token 预算。它累计统计一次 reply 内所有 reasoning 步骤的 token 消耗，一旦预算耗尽，便指示智能体立即收尾、不再调用任何工具。适用于为长链路、工具密集的 ReAct 循环设定成本或时延上限。每次模型调用的加权成本按如下公式计算：

cost = input_token_weight * input_tokens + output_token_weight * output_tokens

当累计成本达到 token_budget 时，middleware 会：

在智能体上下文最后一条 assistant 消息中追加一个 HintBlock（若不存在则新建一条 AssistantMsg），提醒模型给出最终结论性回复。
把下一步 reasoning 的 tool_choice 强制覆写为 ToolChoice(mode="none")，阻止继续调用工具。

像其他 middleware 一样挂载：

from agentscope.agent import Agent
from agentscope.middleware import ReplyBudgetControlMiddleware

agent = Agent(
    name="assistant",
    system_prompt="You are a helpful assistant.",
    model=model,
    toolkit=toolkit,
    middlewares=[
        ReplyBudgetControlMiddleware(
            token_budget=10000,
            input_token_weight=1.0,
            output_token_weight=2.0,
        ),
    ],
)

构造参数如下：

float

必填

单次 reply 允许的最大加权 token 成本。累计成本达到此阈值后，智能体会被指示立刻收尾、不再调用工具。

float

默认值:"1"

计算加权成本时应用于 input tokens 的乘数。

float

默认值:"1"

计算加权成本时应用于 output tokens 的乘数。通常将其设为高于 input_token_weight，以体现 output tokens 实际成本更高。

str

预算超出时注入智能体上下文的提示消息。默认是内置的收尾 prompt，要求模型给出最终结论性回复、不再调用任何工具。

该中间件实例自身无状态 —— 所有运行时状态存放在 agent.state.middle_context 中，以中间件 key 与当前 reply_id 作为索引。这意味着同一个中间件实例可以安全地在多个智能体间共享，且预算状态在 human-in-the-loop（HITL）中断与恢复中保持有效。reply 结束时状态会被自动清理。

预算的作用域是单次 reply，而非智能体生命周期。每次新的 reply 都会从零计数，因此该限制对每次 agent(...) 调用独立生效。

语音生成

TTSMiddleware 拦截智能体的文本输出并合成语音，将 DataBlockStartEvent / DataBlockDeltaEvent / DataBlockEndEvent 音频事件注入到事件流中。它在 on_reply 阶段拦截，观察每个 TextBlockDeltaEvent 和 TextBlockEndEvent。

from agentscope.agent import Agent
from agentscope.middleware import TTSMiddleware
from agentscope.tts import DashScopeTTSModel
from agentscope.credential import DashScopeCredential

agent = Agent(
    name="assistant",
    system_prompt="You are a helpful assistant.",
    model=model,
    toolkit=toolkit,
    middlewares=[
        TTSMiddleware(
            DashScopeTTSModel(
                credential=DashScopeCredential(api_key="..."),
                model="qwen3-tts-flash",
                parameters=DashScopeTTSModel.Parameters(voice="Cherry"),
                stream=True,
            ),
        ),
    ],
)

中间件根据 TTS 模型的模式自动适配：

TTS 模式	行为
标准模式 (`realtime=False`)	累积文本直到 `TextBlockEndEvent`，然后调用 `synthesize(text)` 一次性合成，将完整音频作为一个 data block 输出
实时模式 (`realtime=True`)	每个 `TextBlockDeltaEvent.delta` 通过 `push()` 推送，音频 chunk 即时输出；`TextBlockEndEvent` 时调用 `synthesize()` 刷新剩余音频

输出事件流因模式而异： 标准模式 — 音频在文本块结束后输出：

TextBlockStartEvent
TextBlockDeltaEvent (文本)
TextBlockDeltaEvent (文本)
TextBlockEndEvent
DataBlockStartEvent           ← 文本结束后合成音频
DataBlockDeltaEvent (音频)
DataBlockDeltaEvent (音频)
DataBlockEndEvent

实时模式 — 音频 chunk 与文本交替到达（并发合成）：

TextBlockStartEvent
TextBlockDeltaEvent (文本)
DataBlockStartEvent           ← 文本流期间音频开始
DataBlockDeltaEvent (音频)
TextBlockDeltaEvent (文本)
DataBlockDeltaEvent (音频)
TextBlockEndEvent
DataBlockDeltaEvent (音频)   ← synthesize() 刷新的剩余音频
DataBlockEndEvent

每个 DataBlockDeltaEvent.data 携带增量 base64 编码的音频 chunk；完整音频为所有 delta 解码后的字节拼接，按 block_id 关联。

模型路由

当不同类型的请求适合交给不同模型处理时（例如日常问答交给低成本模型，复杂推理交给更强的模型），ModelRouterMiddleware 可以为每次回复自动选择对话模型。它的工作方式如下：

每次回复开始时，读取输入中最新一条用户消息的文本，交给路由模型从候选模型中选出一个；
本次回复期间把 agent.model 替换为选中的模型，回复结束后恢复为智能体原来的模型；
每次回复只选择一次，回复因用户确认等原因暂停后再恢复时，沿用之前的选择。

路由模型出错、选中不存在的候选，或者输入中没有用户文本时，本次回复沿用智能体自己的模型，不会中断回复。路由模型可以是以下两种之一：

路由模型	选择方式
分类模型（`ClassifierModelBase`）	把候选模型作为一个单选问题交给分类模型，取概率最高的选项
对话模型（`ChatModelBase`）	通过结构化输出让模型从候选名称中选出一个

分类模型来自 agentscope.classifier 模块，它针对给定输入回答类型化的问题（二元判断、单选、评分），并给出每个答案的概率。目前内置的实现是基于 TypeSafe API 的 JevClassifierModel，使用前需要安装可选依赖：

安装 Jev 依赖

pip install "agentscope[classifier-jev]"

下面的示例分别用分类模型和对话模型作为路由模型，在两个候选模型之间选择：

import os
from agentscope.agent import Agent
from agentscope.classifier import JevClassifierModel
from agentscope.credential import DashScopeCredential, TypeSafeCredential
from agentscope.middleware import ChatModelCandidate, ModelRouterMiddleware
from agentscope.model import DashScopeChatModel

credential = DashScopeCredential(api_key=os.environ["DASHSCOPE_API_KEY"])
fast = DashScopeChatModel(credential=credential, model="qwen-plus")
strong = DashScopeChatModel(
    credential=credential,
    model="qwen3-235b-a22b-thinking-2507",
)

router = ModelRouterMiddleware(
    # 路由模型：基于 TypeSafe API 的 Jev 分类模型
    classifier_model=JevClassifierModel(
        credential=TypeSafeCredential(api_key=os.environ["TYPESAFE_API_KEY"]),
    ),
    # 候选模型：name 是分类模型返回的名称，description 说明何时选择它
    candidates=[
        ChatModelCandidate(
            name="fast",
            model=fast,
            description="日常问答与简单任务",
        ),
        ChatModelCandidate(
            name="strong",
            model=strong,
            description="需要多步推理、编程或数学的复杂任务",
        ),
    ],
)

agent = Agent(
    name="assistant",
    system_prompt="You are a helpful assistant.",
    # 路由失败时使用的默认模型
    model=fast,
    middlewares=[router],
)

import os
from agentscope.credential import DashScopeCredential
from agentscope.middleware import ChatModelCandidate, ModelRouterMiddleware
from agentscope.model import DashScopeChatModel

credential = DashScopeCredential(api_key=os.environ["DASHSCOPE_API_KEY"])
fast = DashScopeChatModel(credential=credential, model="qwen-plus")
strong = DashScopeChatModel(
    credential=credential,
    model="qwen3-235b-a22b-thinking-2507",
)

router = ModelRouterMiddleware(
    # 路由模型也可以是任意对话模型，通过结构化输出完成选择
    classifier_model=fast,
    candidates=[
        ChatModelCandidate(name="fast", model=fast, description="日常问答与简单任务"),
        ChatModelCandidate(name="strong", model=strong, description="需要多步推理的复杂任务"),
    ],
    # 可选：自定义路由指令
    instructions="Select the most suitable chat model for the user input.",
)

构造参数如下：

ClassifierModelBase | ChatModelBase

必填

用于选择候选模型的路由模型。

Sequence[ChatModelCandidate]

必填

候选模型列表。每个 ChatModelCandidate 包含 name（路由模型返回的名称，必须唯一）、model（选中时使用的对话模型）和 description（何时选择该模型）。

str

路由问题的指令。默认为 "Select the most suitable chat model for responding to the user input."。

长期记忆

AgentScope 以中间件的形式支持长期记忆，让智能体能够在多个会话间持久保存并调取信息。详见长期记忆章节。

RAG

AgentScope 中知识库同样以中间件形式提供，允许智能体在推理时访问外部知识库。详见 RAG 章节。

自定义中间件

继承基类 MiddlewareBase 并实现所需位置的 hook 即可。下面的示例用一个中间件覆盖了所有位置。每个 onion hook 都会收到一个 input_kwargs 字典，承载流入下一层的字段，通过 next_handler(**input_kwargs) 透传，也可以用关键字参数覆写其中某些字段：

from typing import AsyncGenerator, Awaitable, Callable

from agentscope.agent import Agent
from agentscope.event import AgentEvent
from agentscope.message import Msg
from agentscope.middleware import MiddlewareBase
from agentscope.model import ChatResponse
from agentscope.tool import ToolBase


class FullObservabilityMiddleware(MiddlewareBase):
    """同时观察所有中间件位置，并贡献一个工具。"""

    async def on_reply(
        self,
        agent: Agent,
        # {"inputs": Msg | list[Msg] | UserConfirmResultEvent | ExternalExecutionResultEvent | None}
        input_kwargs: dict,
        next_handler: Callable[..., AsyncGenerator[AgentEvent | Msg, None]],
    ) -> AsyncGenerator[AgentEvent | Msg, None]:
        print(f"[reply] start for {agent.name}")
        async for item in next_handler(**input_kwargs):
            yield item
        print(f"[reply] end for {agent.name}")

    async def on_reasoning(
        self,
        agent: Agent,
        # {"tool_choice": ToolChoice | None}
        input_kwargs: dict,
        next_handler: Callable[..., AsyncGenerator[AgentEvent, None]],
    ) -> AsyncGenerator[AgentEvent, None]:
        print("[reasoning] start")
        async for event in next_handler(**input_kwargs):
            yield event
        print("[reasoning] end")

    async def on_model_call(
        self,
        agent: Agent,
        # {"messages": list[Msg], "tools": list[dict], "tool_choice": ToolChoice | None, "current_model": ChatModelBase}
        input_kwargs: dict,
        next_handler: Callable[
            ..., Awaitable[ChatResponse | AsyncGenerator[ChatResponse, None]]
        ],
    ) -> ChatResponse | AsyncGenerator[ChatResponse, None]:
        print(f"[model_call] {input_kwargs['current_model'].model}")
        result = await next_handler(**input_kwargs)
        print("[model_call] done")
        return result

    async def on_compress_context(
        self,
        agent: Agent,
        # {"context_config": ContextConfig | None}
        input_kwargs: dict,
        next_handler: Callable[..., Awaitable[None]],
    ) -> None:
        print(f"[compress_context] checking context for {agent.name}")
        await next_handler(**input_kwargs)
        print("[compress_context] done")

    async def on_system_prompt(
        self,
        agent: Agent,
        current_prompt: str,
    ) -> str:
        print(f"[system_prompt] length={len(current_prompt)}")
        return current_prompt

    async def list_tools(self) -> list[ToolBase]:
        # 可选 hook。``Agent.__init__`` 不会自动调用；
        # 如要让这些工具可被智能体使用，需要自行从中间件中收集
        # 并传入 toolkit。
        return []

控制智能体流程

on_reply 不只能观察事件，还可以通过「转发、替换或吞掉」事件改变整个 reply 的流程。其中，如果中间件收到 ReplyEndEvent 却不再 yield，Agent 会继续下一轮推理与行动；只有结束事件穿过整条中间件链时，reply 才真正结束。下面的中间件在 Agent 第一次准备结束时追加一条检查指令，并强制再执行一轮：

from collections.abc import AsyncGenerator, Callable

from agentscope.agent import Agent
from agentscope.event import AgentEvent, ReplyEndEvent
from agentscope.message import HintBlock, Msg
from agentscope.middleware import MiddlewareBase
from agentscope.types import ReplyFinishedReason


class ContinueOnceMiddleware(MiddlewareBase):
    async def on_reply(
        self,
        agent: Agent,
        input_kwargs: dict,
        next_handler: Callable[..., AsyncGenerator[AgentEvent | Msg, None]],
    ) -> AsyncGenerator[AgentEvent | Msg, None]:
        continued = False
        async for item in next_handler(**input_kwargs):
            can_continue = (
                isinstance(item, ReplyEndEvent)
                and item.finished_reason
                in {
                    ReplyFinishedReason.COMPLETED,
                    ReplyFinishedReason.EXCEED_MAX_ITERS,
                }
                and not continued
            )
            if can_continue:
                continued = True
                agent.state.cur_iter = min(
                    agent.state.cur_iter,
                    agent.react_config.max_iters - 1,
                )
                agent.state.append_context(
                    agent.name,
                    [HintBlock(hint="结束前请再检查一遍结果。")],
                )
                continue  # 吞掉 ReplyEndEvent，进入下一轮
            yield item

吞掉 EXCEED_MAX_ITERS 前必须释放可用迭代次数（如上例调整 cur_iter），否则 Agent 无法产生新进展。INTERRUPTED 结束事件不能用该机制继续，应原样向外传递。实际中还应设置明确的重试上限，避免中间件导致无限 reply 循环。

执行顺序

洋葱型 hook（on_reply、on_reasoning、on_acting、on_model_call）—— 列表中第一个中间件处于最外层：

middlewares = [mw1, mw2]
# 调用顺序：
# mw1 前 → mw2 前 → 内部逻辑 → mw2 后 → mw1 后

对于流式 / 产出事件的 hook，内层中间件先看到每一个 yield 出的事件：

mw1_pre → mw2_pre → mw2_event → mw1_event → ... → mw2_post → mw1_post

Transformer 类 hook（on_system_prompt）—— 中间件从左到右串行接力：

middlewares = [mw1, mw2]
# original_prompt → mw1.on_system_prompt() → mw2.on_system_prompt() → final

一次 reply 中各 hook 的整体执行顺序遵循 agent 生命周期：

on_reply
  └── 每一轮 ReAct：
        ├── on_compress_context → compress_context()
        │     └── on_system_prompt（压缩前 token 计数）
        ├── on_reasoning
        │     ├── _prepare_model_input() → on_system_prompt
        │     └── on_model_call
        └── on_acting（本轮每个工具调用一次）

list_tools 不在每次 reply 的执行路径上，智能体不会自动调用它 —— 它是一个便捷接口，让中间件可以声明自己的工具。是否收集这些工具由组装 toolkit 的调用方决定。

实用示例

计时中间件

下面的中间件记录每次模型调用的耗时：

import time
from agentscope.middleware import MiddlewareBase

class TimingMiddleware(MiddlewareBase):
    async def on_model_call(self, agent, input_kwargs, next_handler):
        model_name = input_kwargs["current_model"].model
        start = time.time()

        result = await next_handler()

        elapsed = time.time() - start
        print(f"[timing] {agent.name} → {model_name}: {elapsed:.2f}s")
        return result

限速中间件

下面的中间件在两次模型调用之间强制留出最小间隔：

import asyncio
import time
from agentscope.middleware import MiddlewareBase

class RateLimitMiddleware(MiddlewareBase):
    def __init__(self, min_interval: float = 1.0):
        self._last_call = 0.0
        self._min_interval = min_interval

    async def on_model_call(self, agent, input_kwargs, next_handler):
        now = time.time()
        wait = self._min_interval - (now - self._last_call)
        if wait > 0:
            await asyncio.sleep(wait)
        self._last_call = time.time()
        return await next_handler()

动态系统提示中间件

下面的中间件在系统提示中注入实时上下文：

from datetime import datetime
from agentscope.middleware import MiddlewareBase

class DynamicContextMiddleware(MiddlewareBase):
    def __init__(self, context_fn):
        self._context_fn = context_fn

    async def on_system_prompt(self, agent, current_prompt):
        context = self._context_fn()
        return f"{current_prompt}\n\n## Current Context\n{context}"

agent = Agent(
    ...
    middlewares=[
        DynamicContextMiddleware(
            lambda: f"Time: {datetime.now().isoformat()}"
        ),
    ],
)

模型回退中间件

下面的中间件在主模型失败时切换到备用模型：

from agentscope.middleware import MiddlewareBase

class ModelFallbackMiddleware(MiddlewareBase):
    def __init__(self, fallback_model):
        self._fallback = fallback_model

    async def on_model_call(self, agent, input_kwargs, next_handler):
        try:
            return await next_handler()
        except Exception as e:
            print(f"Primary model failed: {e}, switching to fallback")
            return await next_handler(
                current_model=self._fallback,
            )

快速开始

核心模块

智能体即服务

其他

概述

装备中间件

内置中间件

链路追踪

添加自定义 Span

预算控制

语音生成

模型路由

长期记忆

RAG

自定义中间件

控制智能体流程

执行顺序

实用示例

计时中间件

限速中间件

动态系统提示中间件

模型回退中间件

​概述

​装备中间件

​内置中间件

​链路追踪

​添加自定义 Span

​预算控制

​语音生成

​模型路由

​长期记忆

​RAG

​自定义中间件

​控制智能体流程

​执行顺序

​实用示例

​计时中间件

​限速中间件

​动态系统提示中间件

​模型回退中间件

概述

装备中间件

内置中间件

链路追踪

添加自定义 Span

预算控制

语音生成

模型路由

长期记忆

RAG

自定义中间件

控制智能体流程

执行顺序

实用示例

计时中间件

限速中间件

动态系统提示中间件

模型回退中间件