AS AgentScopeAPI 手册 / API Reference
类class agentscope.middleware

MiddlewareBase

class MiddlewareBase
源码Source agentscope/middleware/_base.py:13–303

MiddlewareBase 提供回复、推理、行动、模型调用、权限检查、上下文压缩和系统提示词等钩子。中间件按“环绕”方式包装下一段逻辑，可在不修改智能体源码的情况下加入日志、审计、缓存或策略。位于 agentscope.middleware 包。核心成员包括 is_implemented、on_reply、on_reasoning、on_acting、on_check_permission、on_model_call。

Middleware provides interception mechanisms at 7 key execution points in the Agent lifecycle:

Onion Pattern Hooks (with before/after logic): - on_reply: Intercepts the entire reply process - on_reasoning: Intercepts the reasoning/model call phase - on_check_permission: Intercepts permission checking for a tool call - on_acting: Intercepts individual tool call execution - on_model_call: Intercepts the raw model API call - on_compress_context: Intercepts context compression

Transformer Pattern Hook (sequential pipeline): - on_system_prompt: Transforms the system prompt string

Each hook is optional - only implement the ones you need. The middleware system will automatically detect which hooks are implemented at runtime.

示例Example

class LoggingMiddleware(MiddlewareBase):
        async def on_reasoning(
            self,
            agent: Agent,
            input_kwargs: dict,
            next_handler: Callable[[], AsyncGenerator],
        ) -> AsyncGenerator:
            print(f"Before reasoning for agent {agent.name}")
            async for event in next_handler():
                yield event
            print(f"After reasoning for agent {agent.name}")

    agent = Agent(
        ...
        middlewares=[LoggingMiddleware()],
        ...
    )

核心精讲Deep dive

MiddlewareBase：无侵入钩子MiddlewareBase: non-invasive hooks

MiddlewareBase 提供回复、推理、行动、模型调用、权限检查、上下文压缩和系统提示词等钩子。中间件按“环绕”方式包装下一段逻辑，可在不修改智能体源码的情况下加入日志、审计、缓存或策略。

MiddlewareBase exposes hooks for reply, reasoning, acting, model calls, permission checks, context compression and system prompts. Middleware wraps the next handler, adding logging, auditing, caching or policy without modifying agent source.

  • 每个钩子都接收 agent、input_kwargs 与 next_handler。Every hook receives the agent, input_kwargs and the next_handler.
  • 调用 next_handler 才会继续执行；不调用即可短路。Calling next_handler continues execution; skipping it short-circuits the pipeline.
  • 多个中间件按注册顺序嵌套，适合观测与策略组合。Multiple middlewares nest in registration order, composing observation and policy.

记录推理阶段耗时Trace the reasoning phase

from collections.abc import AsyncGenerator, Callable

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


class LoggingMiddleware(MiddlewareBase):
    async def on_reasoning(
        self,
        agent: Agent,
        input_kwargs: dict,
        next_handler: Callable[[], AsyncGenerator],
    ) -> AsyncGenerator:
        print(f"Before reasoning for {agent.name}")
        async for event in next_handler():
            yield event
        print(f"After reasoning for {agent.name}")

方法摘要Method Summary

方法Method说明Description
def is_implemented(hook_name: str) -> bool
判断是否 `is_implemented`。
Check if a hook method is implemented in the subclass.
async def on_reply(agent: 'Agent', input_kwargs: dict, next_handler: Callable[..., AsyncGenerator]) -> AsyncGeneratorasync
包裹一次完整回复的执行。
Hook for intercepting the reply process.
async def on_reasoning(agent: 'Agent', input_kwargs: dict, next_handler: Callable[..., AsyncGenerator]) -> AsyncGenerat…async
包裹推理阶段，可观察或改写事件流。
Hook for intercepting the reasoning process.
async def on_acting(agent: 'Agent', input_kwargs: dict, next_handler: Callable[..., AsyncGenerator]) -> AsyncGeneratorasync
包裹工具执行阶段。
Hook for intercepting the raw tool execution.
async def on_check_permission(agent: 'Agent', input_kwargs: dict, next_handler: Callable[..., Awaitable['PermissionDeci…async
在权限检查前介入并返回权限决策。
Hook for intercepting permission checking for one tool call.
async def on_model_call(agent: 'Agent', input_kwargs: dict, next_handler: Callable[..., Awaitable['ChatResponse' | Asyn…async
包裹模型调用，可缓存、路由或记录请求。
Hook for intercepting the model API call.
async def on_compress_context(agent: 'Agent', input_kwargs: dict, next_handler: Callable[..., Awaitable[None]]) -> Noneasync
在上下文压缩前后执行自定义逻辑。
Onion hook for `compress_context` function in `Agent` class
async def on_system_prompt(agent: 'Agent', current_prompt: str) -> strasync
动态修改系统提示词。
Transform the system prompt string.
async def list_tools() -> list[ToolBase]async
列出当前连接可用的工具。
List available tools provided by this middleware. Optional to implement.
async def get_middleware_key() -> strasync
获取中间件键。
Get the unique key for this middleware, used to save middleware states in `AgentState` instances.

方法详情Method Detail

is_implemented

def is_implemented(hook_name: str) -> bool

判断是否 is_implemented。

Check if a hook method is implemented in the subclass.

参数Parameters
名称Name类型Type说明Description
hook_namestr钩子方法到检查的名称。Name of the hook method to check
返回Returns bool
返回布尔值，表示判断结果。True if the hook is implemented (overridden), False otherwise
源码Source agentscope/middleware/_base.py:55–66

on_reply

async def on_reply(agent: 'Agent', input_kwargs: dict, next_handler: Callable[..., AsyncGenerator]) -> AsyncGenerator
async

拦截并扩展回复流程。

Swallowing a completed reply's ReplyEndEvent (receiving it without yielding) forces another reasoning-acting round; the final Msg is only produced once the event escapes the whole chain.

参数Parameters
名称Name类型Type说明Description
agent'Agent'智能体（类型 `'Agent'`）。The Agent instance executing this middleware
input_kwargsdict输入关键字参数（类型 `dict`）。Dictionary containing: - inputs: Msg | list[Msg] | UserConfirmResultEvent | ExternalExecutionResultEvent | None — the unified inputs that trigger this reply (new message(s), a resumption event from a previous outside interaction, or None).
next_handlerCallable[..., AsyncGenerator]下一个处理器（类型 `Callable[..., AsyncGenerator]`）。Callable that executes the next middleware or original method
返回Returns AsyncGenerator
返回AsyncGenerator。AgentEvent | Msg: Events from the reply process
源码Source agentscope/middleware/_base.py:68–99

on_reasoning

async def on_reasoning(agent: 'Agent', input_kwargs: dict, next_handler: Callable[..., AsyncGenerator]) -> AsyncGenerator
async

拦截并扩展推理阶段。

Hook for intercepting the reasoning process.

参数Parameters
名称Name类型Type说明Description
agent'Agent'智能体（类型 `'Agent'`）。The Agent instance executing this middleware
input_kwargsdict输入关键字参数（类型 `dict`）。Dictionary containing: - tool_choice: ToolChoice (default None)
next_handlerCallable[..., AsyncGenerator]下一个处理器（类型 `Callable[..., AsyncGenerator]`）。Callable that executes the next middleware or original method
返回Returns AsyncGenerator
返回AsyncGenerator。Various events from the reasoning process
源码Source agentscope/middleware/_base.py:101–122

on_acting

async def on_acting(agent: 'Agent', input_kwargs: dict, next_handler: Callable[..., AsyncGenerator]) -> AsyncGenerator
async

拦截并扩展工具执行阶段。

This hook wraps only the toolkit.call_tool call — i.e.

说明Notes

  • 注意：该行为受运行时配置或底层实现影响，建议结合对应源码与版本说明确认。Note: Tools with `is_state_injected=True` receive the live `agent.state` object. Offloading such tools to a background task may cause concurrent state mutations — guard against this in your middleware implementation.
参数Parameters
名称Name类型Type说明Description
agent'Agent'智能体（类型 `'Agent'`）。The Agent instance executing this middleware.
input_kwargsdict输入关键字参数（类型 `dict`）。Dictionary containing: - `tool_call` (`ToolCallBlock`): the tool call to execute. By the time this hook is invoked the tool call has already been validated and permitted.
next_handlerCallable[..., AsyncGenerator]下一个处理器（类型 `Callable[..., AsyncGenerator]`）。Callable that executes the next middleware or `_acting_impl`.
返回Returns AsyncGenerator
返回AsyncGenerator。`ToolChunk | ToolResponse`:
源码Source agentscope/middleware/_base.py:124–168

on_check_permission

async def on_check_permission(agent: 'Agent', input_kwargs: dict, next_handler: Callable[..., Awaitable['PermissionDecision']]) -> 'PermissionDecision'
async

处理检查权限事件。

This hook runs after the tool has been resolved and its input has been parsed and validated, but before the resulting decision is consumed by the agent.

说明Notes

  • 注意：该行为受运行时配置或底层实现影响，建议结合对应源码与版本说明确认。Note: Returning without calling `next_handler` bypasses the built-in permission resolution for that call.
参数Parameters
名称Name类型Type说明Description
agent'Agent'智能体（类型 `'Agent'`）。The Agent instance performing the permission check.
input_kwargsdict输入关键字参数（类型 `dict`）。Dictionary containing: - `tool_call` (`ToolCallBlock`): validated call metadata used to correlate the permission decision. - `tool` (`ToolBase`): the resolved tool instance. - `tool_input` (`dict`): the parsed and validated input.
next_handlerCallable[..., Awaitable['PermissionDecision']]下一个处理器（类型 `Callable[..., Awaitable['PermissionDecision']]`）。Callable that executes the next middleware or the built-in permission resolution.
返回Returns 'PermissionDecision'
返回'PermissionDecision'。`PermissionDecision`:
源码Source agentscope/middleware/_base.py:170–211

on_model_call

async def on_model_call(agent: 'Agent', input_kwargs: dict, next_handler: Callable[..., Awaitable['ChatResponse' | AsyncGenerator['ChatResponse', None]]]) -> 'ChatResponse' | AsyncGenerator['ChatResponse', None]
async

拦截并扩展模型调用。

Hook for intercepting the model API call.

参数Parameters
名称Name类型Type说明Description
agent'Agent'智能体（类型 `'Agent'`）。The Agent instance executing this middleware
input_kwargsdict输入关键字参数（类型 `dict`）。Dictionary containing: - messages: list[Msg] - tools: list[dict] - tool_choice: ToolChoice - current_model: The model instance used for this call
next_handlerCallable[..., Awaitable['ChatResponse' | AsyncGenerator['ChatResponse', None]]]下一个处理器（类型 `Callable[..., Awaitable['ChatResponse' | AsyncGenerator['ChatResponse', None]]]`）。Callable that executes the next middleware or original method
返回Returns 'ChatResponse' | AsyncGenerator['ChatResponse', None]
返回'ChatResponse'|AsyncGenerator['ChatResponse',None]。ChatResponse or AsyncGenerator[ChatResponse, None]
源码Source agentscope/middleware/_base.py:213–239

on_compress_context

async def on_compress_context(agent: 'Agent', input_kwargs: dict, next_handler: Callable[..., Awaitable[None]]) -> None
async

处理压缩上下文事件。

Onion hook for compress_context function in Agent class

参数Parameters
名称Name类型Type说明Description
agent'Agent'智能体（类型 `'Agent'`）。The Agent instance executing this middleware
input_kwargsdict输入关键字参数（类型 `dict`）。Dictionary containing: - context_config: ContextConfig | None - instructions: HintBlock | None
next_handlerCallable[..., Awaitable[None]]下一个处理器（类型 `Callable[..., Awaitable[None]]`）。Callable that executes the next middleware or original method
返回Returns None
无返回值。No value is returned.
源码Source agentscope/middleware/_base.py:241–262

on_system_prompt

async def on_system_prompt(agent: 'Agent', current_prompt: str) -> str
async

拦截并修改系统提示词。

This uses a transformer/pipeline pattern rather than onion pattern.

参数Parameters
名称Name类型Type说明Description
agent'Agent'智能体（类型 `'Agent'`）。The Agent instance executing this middleware
current_promptstr当前提示词（类型 `str`）。The current system prompt string
返回Returns str
返回字符串结果。The transformed system prompt
源码Source agentscope/middleware/_base.py:264–284

list_tools

async def list_tools() -> list[ToolBase]
async

列出当前连接可用的工具。

List available tools provided by this middleware.

返回Returns list[ToolBase]
返回ToolBase列表。`list[ToolBase]`:
源码Source agentscope/middleware/_base.py:286–294

get_middleware_key

async def get_middleware_key() -> str
async

获取中间件键。

Optionally, middleware classes can override this method to provide their own unique key.

返回Returns str
返回字符串结果。Returns a string.
源码Source agentscope/middleware/_base.py:296–303
基于官方固定版本源码静态生成Generated from pinned official sources · Python 2.0.11 · Java 2.0.4
签名与文档为静态抽取；如有疑问请以源码为准。Signatures and docs are extracted statically; verify against the source when in doubt.