责任链复活：Advisor 拦截机制源码解剖 它不是 Filter —— 它是双向环绕的，而且必须能穿过响应式流

一、类比与差异：Filter / Interceptor / WebFilter / Advisor

二、接口层次与两个载体对象

public interface Advisor extends Ordered {          //!hl
    String getName();                                // 用于日志、观测、链路追踪 //!hl
}

public interface CallAroundAdvisor extends Advisor {
    AdvisedResponse aroundCall(AdvisedRequest advisedRequest, CallAroundAdvisorChain chain); //!hl
}

public interface StreamAroundAdvisor extends Advisor {
    Flux<AdvisedResponse> aroundStream(AdvisedRequest request, StreamAroundAdvisorChain chain); //!hl
}

// 模板类：同时实现上面两个，把"环绕"拆成 before / after 两段                //!hl
public interface BaseAdvisor extends CallAroundAdvisor, StreamAroundAdvisor { //!hl
    AdvisedRequest  before(AdvisedRequest request);   //!hl
    AdvisedResponse after(AdvisedResponse response);  //!hl
}

AdvisedRequest 上有哪些字段

三、链是怎么推进的：DefaultAroundAdvisorChain

public class DefaultAroundAdvisorChain
        implements CallAroundAdvisorChain, StreamAroundAdvisorChain {

    private final List<Advisor> advisors;   // 已按 order 升序排好 //!hl
    private final int index;                // 当前推进位置（不可变，靠 new 推进） //!hl

    public DefaultAroundAdvisorChain(List<Advisor> advisors) {
        // 排序：Ordered 语义，值小的在前（= 更外层）
        this.advisors = advisors.stream().sorted(AnnotationAwareOrderComparator.INSTANCE).toList(); //!hl
        this.index = 0;
    }

    private DefaultAroundAdvisorChain(List<Advisor> advisors, int index) {
        this.advisors = advisors;
        this.index = index;
    }

    @Override
    public AdvisedResponse nextAroundCall(AdvisedRequest request) {   //!hl
        if (this.index >= this.advisors.size()) {
            throw new IllegalStateException("No more advisors to call"); //!hl
        }
        var advisor = this.advisors.get(this.index);
        var next = new DefaultAroundAdvisorChain(this.advisors, this.index + 1); //!hl
        return ((CallAroundAdvisor) advisor).aroundCall(request, next); //!hl
    }
}

四、order 语义与分层放置策略

• Advisor extends Ordered，因此可以直接用 @Order 注解或 org.springframework.core.Ordered 常量；排序由 AnnotationAwareOrderComparator 完成。
• 值越小 = 越外层 = 越早执行 before、越晚执行 after。和 Spring AOP 的 @Order 完全一致。

五、内置 Advisor 一览

六、手写生产级 Advisor：CostAuditAdvisor

为什么用 BaseAdvisor 而不是裸 CallAroundAdvisor： 裸接口只覆盖 call()，一旦业务改成 .stream()，你的成本统计静默失效 —— 没有报错，只是账单对不上。BaseAdvisor 已经替你把 before/after 适配到了流式分支（只在最后一个有效 chunk 上触发 after）， 写一份代码两种模式都生效。

public class TokenBudgetExceededException extends RuntimeException {   //!hl
    public TokenBudgetExceededException(String m) { super(m); }
}

public class CostAuditAdvisor implements BaseAdvisor {                 //!hl

    public static final String CTX_TRACE = "costAudit.traceId";
    public static final String CTX_START = "costAudit.startNanos";
    public static final String CTX_TENANT = "costAudit.tenantId";

    private final MeterRegistry meterRegistry;
    private final long maxEstimatedPromptTokens;
    private final int order;

    public CostAuditAdvisor(MeterRegistry r, long maxEstimatedPromptTokens, int order) {
        this.meterRegistry = r; this.order = order;
        this.maxEstimatedPromptTokens = maxEstimatedPromptTokens;
    }

    @Override public int getOrder() { return this.order; }
    @Override public String getName() { return this.getClass().getSimpleName(); }

    @Override
    public AdvisedRequest before(AdvisedRequest request) {             //!hl
        Map<String, Object> ctx = new HashMap<>(request.adviseContext());
        ctx.put(CTX_TRACE, MDC.get("traceId"));
        ctx.put(CTX_START, System.nanoTime());
        ctx.put(CTX_TENANT, request.advisorParams().getOrDefault("tenantId", "unknown")); //!hl

        long estimate = request.messages().stream().mapToLong(m -> estimateTokens(m.getText()))
                .sum() + estimateTokens(request.userText()) + estimateTokens(request.systemText());
        if (estimate > this.maxEstimatedPromptTokens) {                //!hl
            throw new TokenBudgetExceededException(
                    "estimated prompt tokens " + estimate + " > budget " + this.maxEstimatedPromptTokens);
        }
        return AdvisedRequest.from(request).adviseContext(ctx).build();   //!hl
    }

    @Override
    public AdvisedResponse after(AdvisedResponse response) {           //!hl
        long start = (long) response.adviseContext().getOrDefault(CTX_START, System.nanoTime());
        Usage usage = response.response().getMetadata().getUsage();
        long total = usage != null ? usage.getTotalTokens() : 0;
        String tenant = String.valueOf(response.adviseContext().getOrDefault(CTX_TENANT, "unknown"));

        this.meterRegistry.counter("ai.tokens.total", "tenant", tenant).increment(total);      //!hl
        this.meterRegistry.timer("ai.call.latency", "tenant", tenant)
                .record(System.nanoTime() - start, TimeUnit.NANOSECONDS);                      //!hl
        log.info("ai-call traceId={} tenant={} model={} tokens={}",
                response.adviseContext().get(CTX_TRACE), tenant,
                response.response().getMetadata().getModel(), total);
        return response;
    }

    /** 粗估：中文按字符数 / 1.5，英文按 / 4。精确计费请引入 JTokkit。 */
    private static long estimateTokens(String text) {
        if (text == null || text.isEmpty()) return 0;
        long cjk = text.chars().filter(c -> c > 0x2E80).count();
        return Math.round(cjk / 1.5 + (text.length() - cjk) / 4.0) + 4;
    }
}

注册与使用

@Configuration
class AgentConfig {

    @Bean
    ChatClient chatClient(ChatClient.Builder builder,
                          ChatMemory chatMemory,
                          MeterRegistry meterRegistry) {
        return builder
                .defaultAdvisors(
                        MessageChatMemoryAdvisor.builder(chatMemory).build(),          // order 靠外层 //!hl
                        new CostAuditAdvisor(meterRegistry, 8_000, 900))               // 内层 //!hl
                .build();
    }
}

// 调用侧下发运行时参数
chatClient.prompt()
        .user(q)
        .advisors(a -> a.param("tenantId", tenantId).param("userId", userId))   //!hl
        .call()
        .content();

变体：CacheAdvisor —— 以及 BaseAdvisor 唯一的表达力缺口

想做"命中缓存就直接返回、不调用模型"的 CacheAdvisor？ BaseAdvisor 做不到。

流式 Advisor 的三个坑

@Override
default Flux<AdvisedResponse> aroundStream(AdvisedRequest req, StreamAroundAdvisorChain chain) {
    Assert.notNull(req, "advisedRequest cannot be null");

    AdvisedRequest processed = before(req);                 //!hl
    return chain.nextAroundStream(processed).map(res -> {
        if (res.response() != null
                && res.response().getResult() != null
                && res.response().getResult().getOutput() != null
                && StringUtils.hasText(res.response().getResult().getOutput().getText())) {
            return after(res);                              //!hl
        }
        return res;
    });
}