第 9 章 中间件

本章导读：当你想给所有 Agent 统一加上计时日志、链路追踪、限速或动态 system prompt，而又不想改动任何一个 agent 或 model 类时，middleware 就是为此准备的扩展机制。本章讲解在六类扩展点注入自定义逻辑的中间件机制——洋葱式（包裹式）、变换式、通知式三类 hook 的语义与执行顺序、用 activePoints() 声明参与哪些扩展点、内置的链路追踪等实现，以及计时、限速、模型回退等实用示例。读完本章，你可以在不修改 agent 代码的前提下横切观察与改写执行流程。

本章导览图

你什么时候需要它

判断标准就一条：这段逻辑是否与具体业务无关，又要作用于每一次执行。满足的话，把它装进 middleware，而不是散落在各个业务方法里。典型例子：

• 可观测性。 模型调用耗时、链路追踪、请求日志——内置的 OpenTelemetry 追踪中间件（本章“内置 Middleware”一节）就是现成实现，装配一行代码。
• 流量与稳定性。 两次模型调用之间强制最小间隔（限速）、主模型失败时切到备用模型——本章“实用示例”给出可直接复用的写法。注意如果只是简单的“主→备”回退，builder 上的 fallbackModel / maxRetries 已经够用，不必写 middleware。
• 输入与流程微调。 往 system prompt 注入实时信息、过滤中间轮次的文本只输出最终答案、全部工具被用户拒绝时让 Agent 停下——这些改流程的动作都挂在对应的扩展点上。

与已学内容的关系和边界：第 4 章主循环图的每个阶段（推理、工具执行、模型调用）都有对应的挂点，本章就是往这些挂点上挂逻辑。“这次工具调用允不允许”不由 middleware 管——那是第 10 章权限系统的判定，middleware 更适合做权限之外的横切增强。

概述

Agent middleware 是在不修改 agent 或 model 代码的前提下，向 agent 执行流程中的关键位置注入自定义逻辑（日志、追踪、输入改写、访问控制等）的机制。

AgentScope Java 中，可以在 6 个位置上设置 hook，覆盖了从外层 reply 流程一路下沉到底层模型 API 调用的全链路：

为什么 hook 不止一种形态？因为往执行流程里挂逻辑有三种不同的需求，各自对应一种类型：

• 想包裹一段流程——在它前后插入逻辑、观察中间事件流。机制是 middleware 包裹下一层 handler，在 next.apply(input) 前后各做各的事。这就是 Onion（洋葱式），onAgent、onReasoning、onActing、onModelCall 四个位置都属于它。
• 想接力变换一个值——上一步的输出作为下一步的输入。机制是 middleware 之间串成流水线，前一个的输出作为后一个的输入，不存在「内层」概念。这就是 Transformer（变换式），onSystemPrompt 是唯一的代表。
• 只想在某个时点被通知一声——不需要改变流程。机制是单向同步通知，没有 next 委托；多个 middleware 在固定的生命周期点位按序执行。这就是 Notification（通知式），onAgentStateReady 属于它。

下图展示这些 hook 在 agent 生命周期中的嵌套关系。onSystemPrompt 嵌入在 onReasoning 内部，因为它在 reasoning 步骤组装 system prompt 时被触发；onAgentStateReady 在本次调用的 AgentState 绑定到 RuntimeContext 之后、输入进入流水线之前触发：

onAgent/
├── onAgentStateReady（call 级 AgentState 就绪，输入尚未进入流水线）
└── ReAct loop（每一轮）/
    ├── onReasoning/
    │   ├── onSystemPrompt（组装 system prompt）
    │   └── onModelCall（模型 API 调用）
    └── onActing（每次工具调用）

注意：当前 onActing 只包裹 agent 运行时内部的工具执行；通过 external execution 在 agent 外部执行的工具不会被 onActing 追踪到。

装备 Middleware

AgentScope 把一组 hook 装在一个 MiddlewareBase 实现里 —— 同一个 middleware 类可以同时实现 6 个位置中任意子集的 hook（未实现的 Onion / Transformer 位置默认 next.apply(input)；通知式位置默认不做任何事）。把实例传给 builder 的 middlewares(...) 即可装备：

import io.agentscope.core.ReActAgent;
import io.agentscope.core.middleware.MiddlewareBase;
import io.agentscope.core.tracing.OtelTracingMiddleware;
import java.util.List;

ReActAgent agent =
        ReActAgent.builder()
                .name("assistant")
                .sysPrompt("你是星帆出海的客服小帆。")
                .model(model)
                .toolkit(toolkit)
                .middlewares(List.of(new OtelTracingMiddleware()))
                .build();

middleware(...)（单数）也可单独添加一个；middlewares(...) 接受 List<? extends MiddlewareBase>，未实现的位置自动跳过，不产生任何调用开销。

内置 Middleware

自己动手之前，先看框架自带的三个实现——链路追踪、任务提醒、最终答案过滤，不少需求可以省掉。

OtelTracingMiddleware

OtelTracingMiddleware（位于 io.agentscope.core.tracing）为 agent 全生命周期接入 OpenTelemetry 追踪。它在 onAgent、onModelCall、onActing 三个位置打点，按层级生成 span（追踪中一段有起止时间的操作记录）：

• invoke_agent <name> —— 包裹整次 reply
• chat <model> —— 包裹每次模型 API 调用
• execute_tool <name> —— 包裹每次工具执行

未配置 OpenTelemetry SDK（只剩默认的 no-op provider）时，所有 hook 会直接短路到 next.apply(input)，几乎零开销。

默认情况下，OtelTracingMiddleware 从进程级 GlobalOpenTelemetry 实例读取配置。无参构造函数在 hook 执行时才惰性查找该实例，因此可以在注册全局 SDK 之前构造 middleware。应用如果自行导出 span，除了 AgentScope 之外还需要引入 OpenTelemetry SDK 和 OTLP exporter。使用 OpenTelemetry BOM 保持二者版本一致（下列版本与 AgentScope 当前使用的版本一致）：

<properties>
    <opentelemetry.version>1.61.0</opentelemetry.version>
</properties>

<dependencyManagement>
    <dependencies>
        <dependency>
            <groupId>io.opentelemetry</groupId>
            <artifactId>opentelemetry-bom</artifactId>
            <version>${opentelemetry.version}</version>
            <type>pom</type>
            <scope>import</scope>
        </dependency>
    </dependencies>
</dependencyManagement>

<dependencies>
    <dependency>
        <groupId>io.opentelemetry</groupId>
        <artifactId>opentelemetry-sdk</artifactId>
    </dependency>
    <dependency>
        <groupId>io.opentelemetry</groupId>
        <artifactId>opentelemetry-exporter-otlp</artifactId>
    </dependency>
</dependencies>

构建 agent 之前，在每个进程中只构建并注册一次 SDK。下例中的可选环境变量可保存 Basic <base64-credentials> 形式的值，供 Langfuse 等要求 Authorization header 的后端使用：

import io.agentscope.core.ReActAgent;
import io.agentscope.core.tracing.OtelTracingMiddleware;
import io.opentelemetry.exporter.otlp.http.trace.OtlpHttpSpanExporter;
import io.opentelemetry.sdk.OpenTelemetrySdk;
import io.opentelemetry.sdk.trace.SdkTracerProvider;
import io.opentelemetry.sdk.trace.export.BatchSpanProcessor;

String endpoint =
        System.getenv().getOrDefault(
                "OTEL_EXPORTER_OTLP_ENDPOINT", "http://localhost:4318/v1/traces");
String authorization = System.getenv("OTEL_EXPORTER_OTLP_AUTHORIZATION");

var exporterBuilder = OtlpHttpSpanExporter.builder().setEndpoint(endpoint);
if (authorization != null && !authorization.isBlank()) {
    exporterBuilder.addHeader("Authorization", authorization);
}

SdkTracerProvider tracerProvider =
        SdkTracerProvider.builder()
                .addSpanProcessor(
                        BatchSpanProcessor.builder(exporterBuilder.build()).build())
                .build();

OpenTelemetrySdk.builder()
        .setTracerProvider(tracerProvider)
        .buildAndRegisterGlobal();
Runtime.getRuntime().addShutdownHook(new Thread(tracerProvider::close));

ReActAgent agent =
        ReActAgent.builder()
                .name("assistant")
                .sysPrompt("你是星帆出海的客服小帆。")
                .model(model)
                .toolkit(toolkit)
                .middleware(new OtelTracingMiddleware())
                .build();

必须在 middleware 开始工作前注册 SDK。如果运行环境（例如 Spring Boot 的 OpenTelemetry 自动配置）已经注册了 GlobalOpenTelemetry，直接复用并只添加 middleware 即可。新配置不再调用已弃用的 TracerRegistry.register(...)。应用关闭时应关闭 SdkTracerProvider，让 batch processor 刷新尚未导出的 span。

如果要改用应用自己持有的 SDK，而不是进程级实例，请用 build()（不要用 buildAndRegisterGlobal()）构建 SDK，把 OTLP HTTP exporter 指向应用的 endpoint，并显式挂到 agent 上。这是一条可选的 middleware 路径：调用方拥有 SDK 的生命周期，并负责关闭 provider。生产环境的接入方式仍然是现有的 ReActAgent.builder().middleware(...)，不需要改 builder。

import io.agentscope.core.ReActAgent;
import io.agentscope.core.tracing.OtelTracingMiddleware;
import io.opentelemetry.api.OpenTelemetry;
import io.opentelemetry.exporter.otlp.http.trace.OtlpHttpSpanExporter;
import io.opentelemetry.sdk.OpenTelemetrySdk;
import io.opentelemetry.sdk.trace.SdkTracerProvider;
import io.opentelemetry.sdk.trace.export.BatchSpanProcessor;

String endpoint =
        System.getenv().getOrDefault(
                "OTEL_EXPORTER_OTLP_ENDPOINT", "http://localhost:4318/v1/traces");

SdkTracerProvider tracerProvider =
        SdkTracerProvider.builder()
                .addSpanProcessor(
                        BatchSpanProcessor.builder(
                                        OtlpHttpSpanExporter.builder()
                                                .setEndpoint(endpoint)
                                                .build())
                                .build())
                .build();

OpenTelemetry appSdk =
        OpenTelemetrySdk.builder().setTracerProvider(tracerProvider).build();
Runtime.getRuntime().addShutdownHook(new Thread(tracerProvider::close));

ReActAgent agent =
        ReActAgent.builder()
                .name("assistant")
                .sysPrompt("你是星帆出海的客服小帆。")
                .model(model)
                .toolkit(toolkit)
                .middleware(new OtelTracingMiddleware(appSdk))
                .build();

传入 appSdk 不会替换 GlobalOpenTelemetry，也不会改变 StudioManager。StudioManager 在 initialize() 时仍会独立安装已弃用的 TracerRegistry 追踪。供应商插桩如果改写了 tracer 查找结果，以及 Studio 调用树缺少展开控件，都需要对照实际部署另行验证。此 middleware 决定 onAgent、onModelCall、onActing 的 span 记到哪一个 OpenTelemetry SDK。传入 OpenTelemetry.noop() 同样合法：下游链路仍会执行，且不会记录 span。

构造 OtelTracingMiddleware（无论是 new OtelTracingMiddleware() 还是 new OtelTracingMiddleware(appSdk)）时，第一次创建实例还会注册一个 JVM 范围的 Reactor hook：ContextPropagationOperator.registerOnEachOperator()。从注册那一刻起，该 hook 会在进程中任何代码组装 Flux 和 Mono operator 时对其进行包装，使父 span 在 publishOn / subscribeOn 跨线程之后仍然成立。它无法作用于注册之前已经组装好的链：提前构建、之后复用的 publisher 不会获得上下文传播，因此请在组装需要传播的管道之前先构造该 middleware。它与 span 记到哪一个 SDK 无关：应用自持的 SDK 只把 tracer 查找从 GlobalOpenTelemetry 隔离开，并不能避免这个全局插桩副作用。该 hook 在每个 JVM 中最多安装一次，关闭 middleware 或 SDK 也不会移除它。

每次 reply 会产出一棵嵌套 span 树，关键属性包括 agent 名称、session ID、模型名、token 数、工具名与入参等。

TaskReminderMiddleware

TaskReminderMiddleware（位于 io.agentscope.core.middleware）与内置 TodoTools 配合使用，在每个 reasoning step 之前把当前 AgentState.tasksContext 渲染成 <system-reminder> 注入上下文，避免长任务期间 agent 偏离计划。

通过 builder 上的 enableTaskList(true) 开关与 TodoTools 一同启用：

import io.agentscope.core.ReActAgent;
import io.agentscope.core.tool.Toolkit;
import io.agentscope.core.tool.builtin.TodoTools;

Toolkit toolkit = new Toolkit();
toolkit.registerTool(new TodoTools());

ReActAgent agent =
        ReActAgent.builder()
                .name("planner")
                .sysPrompt("You plan tasks step by step.")
                .model(model)
                .toolkit(toolkit)
                .enableTaskList(true)
                .build();

FinalAnswerFilterMiddleware

FinalAnswerFilterMiddleware 仅输出 ReAct 最终推理轮次的文本。产生工具调用的中间轮次文本会被过滤，工具事件及其他非文本事件仍会正常流式输出。

import io.agentscope.core.middleware.FinalAnswerFilterMiddleware;

ReActAgent agent =
        ReActAgent.builder()
                .name("assistant")
                .model(model)
                .toolkit(toolkit)
                .middleware(new FinalAnswerFilterMiddleware())
                .build();

由于只有在未观察到工具调用时才能确定当前轮次是最终轮次，该 middleware 会将每轮文本缓冲到模型调用结束。

自定义 Middleware

内置的不够用时，实现 MiddlewareBase 接口（位于 io.agentscope.core.middleware）自己写，只重写需要的 hook 即可，其它的不用管。

每个洋葱 hook 收到一个 next 函数，调用 next.apply(input) 进入内层逻辑；可以在调用前后插入自己的处理，或者通过 Flux<AgentEvent> 算子（doOnNext / flatMap / map 等）观察、改写中间事件流。

import io.agentscope.core.agent.Agent;
import io.agentscope.core.agent.RuntimeContext;
import io.agentscope.core.event.AgentEvent;
import io.agentscope.core.middleware.ActingInput;
import io.agentscope.core.middleware.AgentInput;
import io.agentscope.core.middleware.MiddlewareBase;
import io.agentscope.core.middleware.ModelCallInput;
import io.agentscope.core.middleware.ReasoningInput;
import java.util.EnumSet;
import java.util.Set;
import java.util.function.Function;
import reactor.core.publisher.Flux;
import reactor.core.publisher.Mono;

/** 同时观察 agent / reasoning / model_call / system_prompt 四个位置。 */
public class FullObservabilityMiddleware implements MiddlewareBase {

    @Override
    public Set<ExtensionPoint> activePoints() {
        return EnumSet.of(
                ExtensionPoint.ON_AGENT,
                ExtensionPoint.ON_REASONING,
                ExtensionPoint.ON_MODEL_CALL,
                ExtensionPoint.ON_SYSTEM_PROMPT);
    }

    @Override
    public Flux<AgentEvent> onAgent(
            Agent agent, RuntimeContext ctx, AgentInput input, Function<AgentInput, Flux<AgentEvent>> next) {
        System.out.println("[agent] start for " + agent.getName());
        return next.apply(input)
                .doOnComplete(() -> System.out.println("[agent] end for " + agent.getName()));
    }

    @Override
    public Flux<AgentEvent> onReasoning(
            Agent agent, RuntimeContext ctx, ReasoningInput input, Function<ReasoningInput, Flux<AgentEvent>> next) {
        System.out.println("[reasoning] start");
        return next.apply(input).doOnComplete(() -> System.out.println("[reasoning] end"));
    }

    @Override
    public Flux<AgentEvent> onModelCall(
            Agent agent, RuntimeContext ctx, ModelCallInput input, Function<ModelCallInput, Flux<AgentEvent>> next) {
        System.out.println("[model_call] " + input.model().getClass().getSimpleName());
        return next.apply(input).doOnComplete(() -> System.out.println("[model_call] done"));
    }

    @Override
    public Mono<String> onSystemPrompt(Agent agent, RuntimeContext ctx, String currentPrompt) {
        System.out.println("[system_prompt] length=" + currentPrompt.length());
        return Mono.just(currentPrompt);
    }
}

每个 hook 的 input 类型与字段全表见本章末参考速查一节。需要替换流入下一层的字段时，构造一个新的 input record 后再调用 next.apply(...)。

完整可运行示例：agentscope-examples/documentation/.../middleware/CustomizedMiddlewareExample.java、middleware/ModelCallMiddlewareExample.java、middleware/SystemPromptMiddlewareExample.java。

状态就绪通知

onAgentStateReady 是通知式 hook（没有 next）：每次调用同步触发一次，时点在 call 级 AgentState 绑定到 RuntimeContext 之后、输入进入流水线之前。state 与 ctx.getAgentState() 是同一实例且永不为 null——全新会话拿到的是刚创建、上下文为空的状态。inputMessages 是本次调用输入的私有可变副本：就地修改对本次调用全程生效，调用方原始列表不受影响。

import io.agentscope.core.agent.Agent;
import io.agentscope.core.agent.RuntimeContext;
import io.agentscope.core.message.Msg;
import io.agentscope.core.middleware.MiddlewareBase;
import io.agentscope.core.state.AgentState;
import java.util.EnumSet;
import java.util.List;
import java.util.Set;

/** 全新会话首次调用时在输入头部注入用户画像提示。 */
public class SessionBootstrapMiddleware implements MiddlewareBase {

    @Override
    public Set<ExtensionPoint> activePoints() {
        return EnumSet.of(ExtensionPoint.ON_AGENT_STATE_READY);
    }

    @Override
    public void onAgentStateReady(
            Agent agent, RuntimeContext ctx, AgentState state, List<Msg> inputMessages) {
        if (state.getContext().isEmpty()) {  // 冷启动：全新会话
            inputMessages.add(0, loadUserProfilePrompt(ctx.getUserId()));
        }
    }
}

实现必须非阻塞：通知在本次调用的订阅线程上同步执行（流式场景可能是 Reactor 事件循环线程）。触发以每次生命周期执行（订阅）为单位——冷流重订阅会再次触发。抛出的异常原样使本次调用失败，其后的 middleware 不再执行。未覆写该 hook 的 middleware 只是执行默认空调用；固定的开销仅为每次调用一次输入列表浅拷贝，它同时使调用方列表隔离无条件成立。与其他扩展点一样，参与与否由 activePoints() 声明：未声明 ExtensionPoint.ON_AGENT_STATE_READY 的 middleware 即使覆写了该 hook 也不会被通知。生命周期窗口：通知在本次调用通过准入并完成中断注册之后、追踪包裹与错误事件链就位之前执行——通知期间发出的中断会被兑现，而抛出的异常直接到达调用方，不经过 ErrorEvent 钩子或追踪 span。

读取 RuntimeContext

MiddlewareBase 的所有 hook 都将本次 call / stream 绑定的 RuntimeContext（见第 4 章“RuntimeContext”一节）作为第二个参数直接传入——既能读会话字段，也能按类型 / 按 key 取属性，还能反向写入来给下游 hook 和 tool 传值。

import io.agentscope.core.agent.Agent;
import io.agentscope.core.agent.RuntimeContext;
import io.agentscope.core.event.AgentEvent;
import io.agentscope.core.middleware.AgentInput;
import io.agentscope.core.middleware.MiddlewareBase;
import java.util.function.Function;
import reactor.core.publisher.Flux;

/** 把 user / request id 打到日志，并把 trace id 写回 context 供 tool 读取。 */
public class RequestContextMiddleware implements MiddlewareBase {

    @Override
    public Flux<AgentEvent> onAgent(
            Agent agent, RuntimeContext ctx, AgentInput input, Function<AgentInput, Flux<AgentEvent>> next) {
        System.out.printf(
                "[req] user=%s session=%s reqId=%s%n",
                ctx.getUserId(),
                ctx.getSessionId(),
                ctx.get("request_id"));
        ctx.put("trace_id", java.util.UUID.randomUUID().toString());  // 后续 hook / tool 可读
        return next.apply(input);
    }
}

注意点：

• 同一份 RuntimeContext 在整个 reply 内被各层 hook / tool 共享，使用线程安全的内部 map，可以安全地 put 写入。
• 不要把请求级状态缓存到 middleware 实例字段——一个 middleware 实例通常被多个 agent / call 复用；要么放进 RuntimeContext，要么用 Reactor contextWrite。
• 若 builder 上同时配置了全局 toolExecutionContext，框架在分发给 tool 时会把它合并到 per-call context 之后（per-call 优先级更高）。

执行顺序

Onion 类 hook（onAgent、onReasoning、onActing、onModelCall）按 MiddlewareBase.order() 排序——数值越大越处于最外层。默认值是 1；相同 order 的 middleware 保持其 Builder 注册顺序：

middlewares = [mw1(order=2), mw2(order=1)]
// 调用顺序：
// mw1 前 → mw2 前 → 内部逻辑 → mw2 后 → mw1 后

自定义 middleware 可覆写 order()，改变其相对默认优先级的位置。例如 order 为 0 时，会进入所有仍保持默认 order 1 的 middleware 内层：

MiddlewareBase lowerPriority = new MiddlewareBase() {
    @Override
    public int order() {
        return 0;
    }
};

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

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

Transformer 类 hook（onSystemPrompt）—— middleware 从左到右串行接力：

middlewares = [mw1, mw2]
// originalPrompt → mw1.onSystemPrompt() → mw2.onSystemPrompt() → final

通知式 hook（onAgentStateReady）每次调用按 order() 顺序执行一遍——数值大者先执行，与洋葱链的进入方向一致：

middlewares = [mw1(order=2), mw2(order=1)]
// onAgentStateReady：mw1 → mw2

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

onAgent
  ├── onAgentStateReady（状态绑定到 RuntimeContext，早于输入进入流水线）
  └── 每一轮 ReAct：
        ├── onReasoning
        │     ├── prepare model input → onSystemPrompt
        │     └── onModelCall
        └── onActing（本轮每个工具调用一次）

扩展点参与声明（activePoints()）

MiddlewareBase.activePoints() 声明 middleware 在哪些扩展点处于激活状态。它是参与开关，不是对“覆写了哪些方法”的复述：

关键语义：

• 默认全量激活。 未覆写 activePoints() 的 middleware 保持与现状完全一致的行为，包括未来版本新增的扩展点。
• 覆写即接管。 一旦覆写，激活集合就是返回的集合本身；忘记声明的扩展点即使覆写了方法也不会生效。
• 空集合合法。 EnumSet.noneOf(ExtensionPoint.class) 让 middleware 在所有扩展点失活，同时保留注册——可用作运行开关。
• 构造期固化。 agent 构建时读取一次声明；之后修改返回的集合对已构建的 agent 无效。
• 与 order() 正交。 声明只回答“是否参与”，不回答“什么顺序”——各点参与者保持既有 order() 语义。

import io.agentscope.core.middleware.MiddlewareBase;
import java.util.EnumSet;
import java.util.Set;

MiddlewareBase timingOnly =
        new MiddlewareBase() {
            @Override
            public Set<ExtensionPoint> activePoints() {
                // 只在 onModelCall 激活，其余扩展点全部跳过。
                return EnumSet.of(ExtensionPoint.ON_MODEL_CALL);
            }
        };

推荐实践： 为你实现的每个 middleware 准确定义 activePoints()。精确声明让框架能整体跳过未参与者——某扩展点无任何参与者时不建任何包装层（零包装、不组装 pipeline），链更短，构建期的有效执行计划也更清晰。不声明则保持完全兼容的默认行为（全量激活），因此任何时候补充声明都不会破坏行为，只会收窄参与范围。

继承内置 middleware 时同样适用：内置类声明的正是其挂接的扩展点，子类若覆写了新的 hook（例如给 TaskReminderMiddleware 增加 onModelCall），必须相应扩展继承到的 activePoints()——否则新增的 hook 会被静默跳过。

实用示例

最后是几个可以直接照抄的完整示例。

计时 middleware

下面的 middleware 记录每次模型调用的耗时：

import io.agentscope.core.agent.Agent;
import io.agentscope.core.event.AgentEvent;
import io.agentscope.core.middleware.MiddlewareBase;
import io.agentscope.core.middleware.ModelCallInput;
import java.util.function.Function;
import reactor.core.publisher.Flux;

public class TimingMiddleware implements MiddlewareBase {
    @Override
    public Flux<AgentEvent> onModelCall(
            Agent agent, ModelCallInput input, Function<ModelCallInput, Flux<AgentEvent>> next) {
        long start = System.nanoTime();
        return next.apply(input)
                .doFinally(sig -> {
                    long ms = (System.nanoTime() - start) / 1_000_000;
                    System.out.println(
                            "[timing] " + agent.getName() + ": " + ms + "ms");
                });
    }
}

限速 middleware

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

import io.agentscope.core.agent.Agent;
import io.agentscope.core.event.AgentEvent;
import io.agentscope.core.middleware.MiddlewareBase;
import io.agentscope.core.middleware.ModelCallInput;
import java.time.Duration;
import java.util.concurrent.atomic.AtomicLong;
import java.util.function.Function;
import reactor.core.publisher.Flux;
import reactor.core.publisher.Mono;

public class RateLimitMiddleware implements MiddlewareBase {

    private final long minIntervalMs;
    private final AtomicLong lastCall = new AtomicLong(0);

    public RateLimitMiddleware(Duration minInterval) {
        this.minIntervalMs = minInterval.toMillis();
    }

    @Override
    public Flux<AgentEvent> onModelCall(
            Agent agent, ModelCallInput input, Function<ModelCallInput, Flux<AgentEvent>> next) {
        long now = System.currentTimeMillis();
        long wait = minIntervalMs - (now - lastCall.get());
        Mono<Void> delay = wait > 0 ? Mono.delay(Duration.ofMillis(wait)).then() : Mono.empty();
        return delay.thenMany(next.apply(input))
                .doOnSubscribe(s -> lastCall.set(System.currentTimeMillis()));
    }
}

动态 system prompt middleware

下面的 middleware 在 system prompt 中注入实时上下文。也可以直接复用示例 middleware/SystemPromptMiddlewareExample.java：

import io.agentscope.core.agent.Agent;
import io.agentscope.core.middleware.MiddlewareBase;
import java.time.Instant;
import java.util.function.Supplier;
import reactor.core.publisher.Mono;

public class DynamicContextMiddleware implements MiddlewareBase {

    private final Supplier<String> contextFn;

    public DynamicContextMiddleware(Supplier<String> contextFn) {
        this.contextFn = contextFn;
    }

    @Override
    public Mono<String> onSystemPrompt(Agent agent, String currentPrompt) {
        return Mono.just(currentPrompt + "\n\n## Current Context\n" + contextFn.get());
    }
}

// 装配：
// .middlewares(List.of(new DynamicContextMiddleware(() -> "Time: " + Instant.now())))

模型回退 middleware

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

import io.agentscope.core.agent.Agent;
import io.agentscope.core.event.AgentEvent;
import io.agentscope.core.middleware.MiddlewareBase;
import io.agentscope.core.middleware.ModelCallInput;
import io.agentscope.core.model.Model;
import java.util.function.Function;
import reactor.core.publisher.Flux;

public class ModelFallbackMiddleware implements MiddlewareBase {

    private final Model fallback;

    public ModelFallbackMiddleware(Model fallback) {
        this.fallback = fallback;
    }

    @Override
    public Flux<AgentEvent> onModelCall(
            Agent agent, ModelCallInput input, Function<ModelCallInput, Flux<AgentEvent>> next) {
        return next.apply(input)
                .onErrorResume(err -> {
                    System.err.println("Primary model failed: " + err.getMessage()
                            + ", switching to fallback");
                    return next.apply(
                            new ModelCallInput(
                                    input.messages(),
                                    input.tools(),
                                    input.options(),
                                    fallback));
                });
    }
}

提示：若只是简单的「主→备」回退，ReActAgent.Builder 直接暴露了 fallbackModel(...) 与 maxRetries(...)，无需自己写 middleware。观察切换同理：切换发生在 onModelCall 接缝之下，用 ReActAgent.Builder.failoverListener(...)，而不是写 middleware。

全部工具被拒绝时停止 agent

当用户通过 HITL 拒绝了一轮推理产出的全部工具调用时，agent 默认会继续下一轮推理（向后兼容）。如果希望在这种场景下停止 agent，可以编写一个 onActing middleware 观察 AllToolsDeniedEvent 并发出 RequestStopEvent：

import io.agentscope.core.agent.Agent;
import io.agentscope.core.agent.RuntimeContext;
import io.agentscope.core.event.AgentEvent;
import io.agentscope.core.event.AllToolsDeniedEvent;
import io.agentscope.core.event.RequestStopEvent;
import io.agentscope.core.message.GenerateReason;
import io.agentscope.core.middleware.ActingInput;
import io.agentscope.core.middleware.MiddlewareBase;
import java.util.function.Function;
import reactor.core.publisher.Flux;

public class StopOnAllDeniedMiddleware implements MiddlewareBase {

    @Override
    public Flux<AgentEvent> onActing(
            Agent agent, RuntimeContext ctx, ActingInput input,
            Function<ActingInput, Flux<AgentEvent>> next) {
        return next.apply(input)
                .flatMap(event -> {
                    if (event instanceof AllToolsDeniedEvent) {
                        return Flux.just(
                                event,
                                new RequestStopEvent(
                                        "All tools denied by user",
                                        GenerateReason.ALL_TOOLS_DENIED));
                    }
                    return Flux.just(event);
                });
    }
}

装配后，agent 在所有工具被拒绝时会立即停止，返回 GenerateReason.ALL_TOOLS_DENIED：

ReActAgent agent =
        ReActAgent.builder()
                .name("guarded")
                .sysPrompt("...")
                .model(model)
                .toolkit(toolkit)
                .middlewares(List.of(new StopOnAllDeniedMiddleware()))
                .build();

本章小结

这一章走了一条从观察到改写的路线：

1. 机制：六个扩展点、三种 hook 形态——Onion 包裹（onAgent / onReasoning / onActing / onModelCall）、Transformer 串行接力（onSystemPrompt）、Notification 单向通知（onAgentStateReady）；
2. 装备：实现 MiddlewareBase，把实例交给 middlewares(...)；未实现的位置自动跳过；
3. 参与与顺序：activePoints() 声明参与哪些扩展点——覆写即接管、空集合可作运行开关；order() 决定洋葱层级，数值越大越外层；
4. 内置优先：链路追踪（OtelTracingMiddleware）、任务提醒（TaskReminderMiddleware）、最终答案过滤（FinalAnswerFilterMiddleware），不少需求不必自己写；
5. 实操：计时、限速、动态 system prompt、模型回退、全部工具被拒时停止——五个可直接照抄的示例。

middleware 适合做权限之外的横切增强；每一次工具调用本身准不准执行，是下一章权限系统的判定。

验收清单——读到这里，你应该能：

• ☐ 说出六个扩展点各自覆盖哪段流程；
• ☐ 解释 Onion、Transformer、Notification 三种 hook 形态的机制差别；
• ☐ 装备一个自定义 middleware，并用 activePoints() 只激活需要的扩展点；
• ☐ 预言两个 order() 不同的 middleware 在一次 reply 中的执行顺序；
• ☐ 优先复用内置中间件，而不是重写追踪与过滤逻辑；
• ☐ 照抄实用示例，写出一个 onModelCall 计时 middleware。

参考速查

本节收拢各 hook 的 input 类型与字段，供实现时查阅；正文主线不依赖本节。

Hook input 类型

每个 hook 的 input 类型均位于 io.agentscope.core.middleware：