★ FEATURED ARTICLE 1. 从“五个子智能体”到客服流水线一次对 Spring AI Alibaba 内置编排能力的逆向解剖我第一次看到 Spring AI Alibaba 的Flow和A2AAgent-to-Agent调用文档时满脑子都是问号。官方示例里那几行flow.builder().add(...).add(...)看起来像 DSL 脚本但没人告诉我它到底在调度什么、数据怎么流、失败了往哪抛。直到上个月一个模拟项目X要求我快速搭出能处理“查订单→验身份→查物流→生成话术→转人工”五步闭环的客服机器人我才真正把这堆抽象概念摁在地上一节一节拆开来看。这个标题里的“五个子智能体”不是指五个独立部署的服务而是五个职责明确、可复用、带状态边界的逻辑单元——查单 Agent、身份核验 Agent、物流查询 Agent、话术生成 Agent、人工转接 Agent。它们各自封装了领域知识、API 调用、错误兜底和上下文传递逻辑。而“拼成客服流水线”本质上是在 Spring AI Alibaba 框架内用Flow定义它们的执行顺序、输入输出契约、异常分支与重试策略用A2A实现跨 Agent 的轻量级远程调用不走 HTTP不启新线程却能天然支持异步、超时、熔断和上下文透传。关键词里虽然空着但整件事的核心锚点其实就三个Flow 编排不是简单串行而是带条件跳转、并行合并、失败回滚的有向无环图、A2A 远程调用不是 RPC也不是消息队列是 Spring 生态内基于Function接口 Reactive流的语义化代理、子智能体自治性每个 Agent 必须能独立测试、独立配置、独立降级不能互相污染状态。这三点决定了你搭出来的到底是“能跑的 Demo”还是“可运维、可灰度、可监控的生产级流水线”。很多人卡在第一步为什么非得用 Flow直接写个 Service 方法链不更直白实测下来纯 Java 方法链在真实客服场景里会迅速失控。比如“查订单”成功后要“验身份”但如果用户没登录就得跳过验身份直接进话术生成环节又比如“查物流”超时了不能让整个流程卡死得降级返回“物流信息暂未同步”同时触发告警。这些分支、超时、降级、日志追踪如果全靠 if-else 和 try-catch 堆三个月后连你自己都看不懂调用树。而 Flow 提供的是一套声明式契约你只说“这个节点输入是什么、输出给谁、失败了走哪条路”框架自动帮你织入重试、熔断、链路追踪和可观测埋点。这不是炫技是工程复杂度的必然收敛。提示Flow 不是替代 Spring Integration 或 Apache Camel 的企业集成方案它的定位非常清晰——专为 LLM 驱动的多步骤智能体协作而生。它不处理 MQ 消息路由也不做 ETL 数据清洗但它能把“调用大模型生成摘要”和“调用内部 API 查询数据库”这两个异构操作用同一套语义、同一套错误码、同一套监控指标串起来。2. Flow 编排的底层骨架不是工作流引擎而是函数式管道编译器Spring AI Alibaba 的 Flow表面看是Flow.Builder链式调用但它的本质是一个将声明式节点定义编译为可执行FunctionFluxInput, FluxOutput管道的编译器。这句话听起来绕但理解它是避开后续所有坑的前提。我们先看一个最简 Flow 定义Flow flow Flow.builder() .add(checkOrder, checkOrderAgent) .add(verifyIdentity, verifyIdentityAgent) .add(getLogistics, getLogisticsAgent) .build();这段代码执行后flow对象并不是一个“运行时容器”而是一个被编译好的、可复用的函数式管道。它内部会做三件关键事节点拓扑解析将.add()的顺序转化为有向边构建 DAG有向无环图。注意这里默认是线性依赖但你可以用.add(A, a).add(B, b, A)显式指定 B 依赖 A从而支持扇入多个节点输出汇入一个节点输入输出契约绑定每个 Agent 必须实现FunctionInput, Output接口或其 Reactive 变体FunctionFluxInput, FluxOutput。Flow 在编译时会检查checkOrderAgent的输入类型是否与前一节点的输出类型兼容不兼容则编译期报错而不是运行时报ClassCastException执行上下文注入Flow 会为每个节点创建一个AgentExecutionContext其中包含本次调用的全局correlationId、requestId、timeout、retryConfig以及一个可读写的MapString, Object用于跨节点传递临时状态比如“验身份”成功后把用户等级写进去“话术生成”就能据此调整语气。这才是为什么它叫“内置编排”——它不依赖外部工作流引擎如 Camunda所有调度逻辑都运行在当前 Spring 应用的同一个 JVM 里没有网络开销没有序列化成本也没有额外的运维组件。但代价是它无法跨进程编排。如果你的五个子智能体分别部署在五个微服务里Flow 就不适用你得切到 A2A 或其他 RPC 方案。那么Flow 如何处理分支比如“验身份失败”时跳过“查物流”直接进“话术生成”答案是用ConditionalNode。它不是 if-else 语法糖而是一个实现了FunctionInput, Boolean的判断函数Flow flow Flow.builder() .add(checkOrder, checkOrderAgent) .add(verifyIdentity, verifyIdentityAgent) .add(getLogistics, getLogisticsAgent, ConditionalNode.of(shouldGetLogistics, input - ((Map)input).get(identityVerified) Boolean.TRUE)) .add(generateScript, generateScriptAgent) .build();这里ConditionalNode.of()创建了一个判断节点它的apply()方法返回true时才执行getLogistics。注意这个判断函数接收的是上游节点的完整输出通常是 Map 或自定义 DTO你可以从中提取任意字段做决策。这种设计的好处是判断逻辑可测试、可配置、可热更新不像硬编码的 if-else 那样散落在各处。再来看并行。客服场景中“查订单”和“查用户画像”完全可以并发执行等两者都返回后再合并结果。Flow 用ParallelNode支持ParallelNode parallel ParallelNode.builder() .add(orderInfo, checkOrderAgent) .add(profileInfo, getUserProfileAgent) .mergeStrategy((results) - { Map merged new HashMap(); merged.putAll(results.get(orderInfo)); merged.putAll(results.get(profileInfo)); return merged; }) .build(); Flow flow Flow.builder() .add(parallelFetch, parallel) // 这个节点输出就是合并后的 Map .add(generateScript, generateScriptAgent) .build();ParallelNode的mergeStrategy是关键。它不是一个简单的Map.merge()而是你定义的业务逻辑比如当订单状态是“已发货”且用户等级是“VIP”时话术要加一句“您的专属物流管家已上线”。这个合并策略就是真正的业务胶水。注意ParallelNode 的并发是 Reactive 的基于 Project Reactor 的Mono.zip()或Flux.zip()不是Async线程池。这意味着它天然支持背压backpressure不会因为下游消费慢而导致上游 OOM。但这也意味着如果你的某个子 Agent 是阻塞式 JDBC 查询必须用Mono.fromCallable(() - {...}).subscribeOn(Schedulers.boundedElastic())显式切换线程否则会阻塞整个 Reactor 事件循环线程。最后说失败处理。Flow 默认是“任一节点失败整个流程中断”。但客服场景需要更精细的控制比如“查物流”超时可以降级返回缓存数据不影响整体流程。这时要用RetryableNodeRetryableNode retryableLogistics RetryableNode.builder(getLogisticsAgent) .maxAttempts(3) .backoff(Duration.ofSeconds(1)) .fallback((input, throwable) - { log.warn(getLogistics failed, use cache, throwable); return Mono.just(Map.of(logisticsStatus, CACHE_UNAVAILABLE)); }) .build();fallback函数是核心。它接收原始输入和异常返回一个MonoOutput作为降级结果。这个设计比 Hystrix 的 fallbackMethod 更灵活因为你可以在降级逻辑里调用缓存、发告警、甚至调用另一个备用 Agent。3. A2A 远程调用不是 RPC是带语义的函数代理当五个子智能体分散在不同模块、甚至不同 Spring Boot 应用里时Flow 的本地编排就失效了。这时候Spring AI Alibaba 提供的 A2AAgent-to-Agent调用机制就成了唯一选择。但必须立刻划清界限A2A 不是传统 RPC不是 gRPC也不是 REST Client 封装。它是一种基于 Spring Cloud Function 和 WebClient 的、面向智能体交互场景的语义化代理。它的核心思想是让远程 Agent 像本地 Function 一样被调用同时自动处理序列化、超时、重试、熔断、链路追踪且调用方完全感知不到网络细节。我们以“查订单 Agent”为例。假设它部署在order-service应用中暴露一个标准的 Spring Cloud Function Endpoint// order-service 中 Bean public FunctionOrderQueryRequest, OrderQueryResponse checkOrder() { return request - { // 真实查库逻辑 return orderService.queryByOrderId(request.getOrderId()); }; }在customer-service客服主应用中你不需要写任何 HTTP Client 代码。只需声明一个A2AAgentBeanBean public A2AAgent checkOrderAgent() { return A2AAgent.builder() .functionName(checkOrder) // 对应 order-service 中的 Function 名 .baseUrl(http://order-service:8080) // 目标服务地址 .timeout(Duration.ofSeconds(5)) .retryConfig(RetryConfig.builder() .maxAttempts(2) .backoff(Duration.ofMillis(500)) .build()) .build(); }然后在 Flow 中直接使用Flow flow Flow.builder() .add(checkOrder, checkOrderAgent) // 看起来和本地 Agent 一模一样 .add(verifyIdentity, verifyIdentityAgent) .build();这就是 A2A 的魔力调用方代码零感知网络。checkOrderAgent.apply(input)这一行背后发生了什么请求构造A2A 会将inputOrderQueryRequest序列化为 JSON通过 WebClient 发送到http://order-service:8080/functions/checkOrder超时与重试由timeout和retryConfig控制失败时自动重试重试间有退避backoff响应反序列化收到OrderQueryResponseJSON 后自动反序列化为 Java 对象错误映射如果order-service返回 4xx/5xxA2A 会包装成A2AException并携带原始 HTTP 状态码和错误消息方便 Flow 的fallback处理链路追踪注入自动将当前TraceId和SpanId注入 HTTP Header如X-B3-TraceId确保全链路可观测。所以A2A 的本质是 Spring Cloud Function 的 HTTP 绑定层 智能体交互协议的封装。它不发明新轮子而是把现有生态的能力用智能体协作的语义重新组织。但这里有个极易踩的坑A2A 调用默认是阻塞式的。上面的checkOrderAgent.apply(input)看似是函数调用但底层是WebClient.block()。在高并发客服场景下这会导致线程池耗尽。正确做法是强制使用 Reactive 版本。Bean public A2AAgent checkOrderAgent() { return A2AAgent.builder() .functionName(checkOrder) .baseUrl(http://order-service:8080) .useReactive(true) // 关键启用 Reactive 模式 .build(); }启用useReactive(true)后checkOrderAgent.apply(input)返回的是MonoOrderQueryResponse而不是OrderQueryResponse。这意味着你必须把整个 Flow 升级为 ReactiveFlow flow Flow.builder() .add(checkOrder, checkOrderAgent) // 返回 Mono .add(verifyIdentity, verifyIdentityAgent) // 也必须是 Mono... 类型 .build(); // 执行时 MonoFluxObject result flow.execute(Mono.just(input));这个升级不是可选的是必须的。因为只有 Reactive 模式才能实现真正的非阻塞 I/O才能支撑每秒数千次的客服请求。而本地 Agent 也必须适配verifyIdentityAgent的类型得是FunctionMonoInput, MonoOutput而不是FunctionInput, Output。注意A2A 的baseUrl必须是服务发现就绪的地址。如果你用 Nacos 或 Eureka应该用DiscoveryClient动态获取实例而不是写死 IP。Spring AI Alibaba 提供了DiscoveryA2AAgentBuilder它会自动从注册中心拉取order-service的健康实例列表并做负载均衡默认轮询。这是生产环境的标配否则单点故障会直接击穿整个客服流水线。4. 五个子智能体的实战拆解从单点功能到流水线协同现在我们把前面讲的 Flow 和 A2A真正落到“查订单→验身份→查物流→生成话术→转人工”这五个子智能体上。这不是理论推演而是我在模拟项目X中实际落地的结构。每个 Agent 都是一个独立的 Spring Bean遵循统一契约但内部实现千差万别。4.1 查订单 Agent状态机驱动的容错查询这个 Agent 的输入是OrderQueryRequest含orderId,userId输出是OrderQueryResponse含订单状态、金额、商品列表等。难点在于订单可能不存在、用户无权查看、数据库临时不可用。如果简单 throw Exception整个 Flow 就断了。我的实现是引入轻量级状态机public class CheckOrderAgent implements FunctionMonoOrderQueryRequest, MonoOrderQueryResponse { private final OrderService orderService; private final CacheManager cacheManager; Override public MonoOrderQueryResponse apply(MonoOrderQueryRequest requestMono) { return requestMono .flatMap(request - { // Step 1: 先查缓存毫秒级 String cacheKey order: request.getOrderId(); MonoOrderQueryResponse fromCache cacheManager.get(cacheKey, OrderQueryResponse.class); return fromCache.switchIfEmpty( // Step 2: 缓存未命中查 DB但加熔断 Mono.fromCallable(() - orderService.queryByOrderId(request.getOrderId())) .transform(it - CircuitBreakerOperator.of(circuitBreaker)) // 熔断器 .onErrorResume(throwable - { // Step 3: DB 查询失败降级为“订单状态未知” log.error(DB query failed for order {}, request.getOrderId(), throwable); return Mono.just(OrderQueryResponse.unknown(request.getOrderId())); }) .doOnSuccess(resp - { // Step 4: 查询成功写回缓存异步不阻塞主流程 cacheManager.put(cacheKey, resp, Duration.ofMinutes(10)); }) ); }); } }这个 Agent 的价值不在“查”而在“查的策略”缓存穿透防护、熔断降级、异步回填。它让“查订单”这个动作从一个脆弱的数据库依赖变成了一个高可用的业务能力。4.2 验身份 Agent上下文驱动的动态策略输入是OrderQueryResponse来自上一步输出是IdentityVerificationResult含verified: true/false,riskLevel: LOW/MEDIUM/HIGH。关键点是验证策略不能写死。对普通订单只需校验userId匹配对高风险订单金额 5000需调用风控服务做二次验证。我用AgentExecutionContext实现动态路由public class VerifyIdentityAgent implements FunctionMonoOrderQueryResponse, MonoIdentityVerificationResult { private final RiskAssessmentService riskService; private final UserService userService; Override public MonoIdentityVerificationResult apply(MonoOrderQueryResponse responseMono) { return responseMono .flatMap(response - { // 从上下文中提取风险等级由前序节点或规则引擎注入 String riskLevel (String) AgentExecutionContext.current().get(riskLevel); if (HIGH.equals(riskLevel)) { // 高风险调用风控服务 return riskService.assessRisk(response.getOrderId()) .map(result - IdentityVerificationResult.of(result.isApproved(), result.getRiskLevel())); } else { // 低风险本地校验 return userService.verifyUserOwnsOrder(response.getUserId(), response.getOrderId()) .map(valid - IdentityVerificationResult.of(valid, LOW)); } }); } }AgentExecutionContext.current()是 Flow 提供的线程局部变量它让跨节点的状态传递变得极其自然。你不需要在每个Mono里手动zip()上下文框架自动为你维护。4.3 查物流 AgentA2A 调用的典型范式这是第一个 A2A 调用。目标服务logistics-service提供getLogisticsFunction。我的LogisticsAgent就是前面展示的A2AAgent实例。但生产环境必须加两道保险本地缓存兜底即使logistics-service宕机也能返回 5 分钟前的物流快照异步刷新当 A2A 调用成功后启动一个异步任务把最新物流信息推送到 Redis供下次缓存命中。public class LogisticsAgent implements FunctionMonoOrderQueryResponse, MonoLogisticsResponse { private final A2AAgent logisticsA2A; private final CacheManager cacheManager; Override public MonoLogisticsResponse apply(MonoOrderQueryResponse responseMono) { return responseMono .flatMap(response - { String cacheKey logistics: response.getOrderId(); // 先查缓存 MonoLogisticsResponse fromCache cacheManager.get(cacheKey, LogisticsResponse.class); return fromCache.switchIfEmpty( // 缓存未命中走 A2A logisticsA2A.apply(Mono.just(new LogisticsQueryRequest(response.getOrderId()))) .onErrorResume(throwable - { log.warn(A2A to logistics-service failed, use cache, throwable); return cacheManager.get(cacheKey, LogisticsResponse.class); }) .doOnSuccess(logisticsResp - { // 异步刷新缓存不阻塞主流程 CompletableFuture.runAsync(() - cacheManager.put(cacheKey, logisticsResp, Duration.ofMinutes(5))); }) ); }); } }这个模式我称之为“A2A with Cache-Aside Async Refresh”是微服务间智能体调用的黄金组合。4.4 生成话术 AgentLLM 驱动的上下文融合这是整个流水线的“大脑”。输入是前面所有节点的聚合结果订单、身份、物流输出是ScriptResponse含text: String,tone: FORMAL/CASUAL,nextStep: TRANSFER_TO_AGENT/END_CONVERSATION。我用 Spring AI Alibaba 的ChatClient封装了一个ScriptGeneratorAgentpublic class ScriptGeneratorAgent implements FunctionMonoMapString, Object, MonoScriptResponse { private final ChatClient chatClient; Override public MonoScriptResponse apply(MonoMapString, Object contextMono) { return contextMono .flatMap(context - { // 构建 LLM Prompt注入所有上下文 String prompt buildPrompt(context); return chatClient.call(new Prompt(List.of(new SystemMessage(你是一名专业客服助手...), new UserMessage(prompt)))) .map(chatResponse - { // 解析 LLM 输出的 JSON return parseScriptResponse(chatResponse.getResult().getOutput().getContent()); }); }); } private String buildPrompt(MapString, Object context) { OrderQueryResponse order (OrderQueryResponse) context.get(order); IdentityVerificationResult identity (IdentityVerificationResult) context.get(identity); LogisticsResponse logistics (LogisticsResponse) context.get(logistics); return String.format( 订单号 %s状态 %s用户已验证%s物流状态%s。请生成一段 %s 语气的客服话术告知用户当前情况并建议下一步操作。, order.getOrderId(), order.getStatus(), identity.isVerified(), logistics.getStatus(), identity.getRiskLevel().equals(HIGH) ? 正式 : 亲切 ); } }关键点在于buildPrompt它把结构化数据订单、身份、物流动态组装成自然语言指令让 LLM 真正理解业务上下文。这比直接喂 JSON 给模型有效十倍。4.5 转人工 Agent状态驱动的无缝交接最后一个 Agent 不是调用 API而是根据话术生成的结果决定是否触发人工介入。输入是ScriptResponse输出是TransferInstruction含queue: VIP_QUEUE/STANDARD_QUEUE,priority: URGENT/NORMAL,context: Map。它的逻辑很简单public class TransferToAgentAgent implements FunctionMonoScriptResponse, MonoTransferInstruction { Override public MonoTransferInstruction apply(MonoScriptResponse scriptMono) { return scriptMono .map(script - { if (script.getNextStep() NEXT_STEP.TRANSFER_TO_AGENT) { // 根据用户等级和订单金额分配队列和优先级 String queue script.getUserLevel() ! null script.getUserLevel().equals(VIP) ? VIP_QUEUE : STANDARD_QUEUE; String priority script.getOrderAmount() 10000 ? URGENT : NORMAL; return new TransferInstruction(queue, priority, script.getContext()); } else { return null; // 不转人工返回 null 表示跳过 } }) .filter(Objects::nonNull); // 过滤掉 null保证下游只收到需要转人工的指令 } }这个 Agent 的价值在于把“转人工”这个运营动作变成了一个可编程、可配置、可审计的业务决策点。它不再依赖客服人员的手动判断而是由数据驱动。5. 流水线的可观测性没有监控的智能体就是黑盒搭好五个子智能体和 Flow 编排只是完成了 50%。剩下 50%是让整个流水线“看得见、管得住、调得顺”。Spring AI Alibaba 内置了 Micrometer 集成但默认配置远远不够生产。5.1 Flow 级别的指标不只是成功率我为每个 Flow 实例注册了以下核心指标spring.ai.flow.execution.time.seconds按flow.name和statussuccess/failed打点P95/P99 延迟一目了然spring.ai.flow.node.execution.count按flow.name和node.name统计每个节点的执行次数快速定位瓶颈节点比如“查物流”调用量突增说明前端页面有 bugspring.ai.flow.node.error.count按flow.name,node.name,error.typeTimeoutException, A2AException, RuntimeException统计精准定位故障根因。这些指标配合 Grafana 看板能回答所有关键问题哪个节点最慢哪个节点错误最多错误类型分布如何流量峰值出现在什么时间5.2 A2A 调用的深度追踪A2A 的baseUrl是服务名但实际调用的是具体实例。我扩展了A2AAgent在每次调用前后记录targetInstance: ip:port和responseTimeMs并上报为 Tagged Metric// 扩展 A2AAgent 的 execute 方法 public MonoResponse execute(Request request) { long start System.currentTimeMillis(); return webClient.post() .uri(/functions/{functionName}, functionName) .bodyValue(request) .exchangeToMono(clientResponse - { long duration System.currentTimeMillis() - start; // 上报指标a2a.call.duration.seconds{functioncheckOrder, instance10.0.1.12:8080, status200} meterRegistry.timer(a2a.call.duration.seconds, function, functionName, instance, clientResponse.request().url().getHost() : clientResponse.request().url().getPort(), status, String.valueOf(clientResponse.statusCode().value())) .record(duration, TimeUnit.MILLISECONDS); return clientResponse.bodyToMono(Response.class); }); }这样当logistics-service出现慢查询时你不仅能知道“A2A 调用慢”还能精确到“调用logistics-service的10.0.1.12实例慢”直接定位到具体机器。5.3 子智能体的独立健康检查每个 Agent 必须提供/actuator/health/{agentName}端点。例如Component public class CheckOrderAgentHealthIndicator implements HealthIndicator { private final OrderService orderService; Override public Health health() { try { // 执行一个轻量级探活查询 orderService.ping(); return Health.up().withDetail(ping, success).build(); } catch (Exception e) { return Health.down().withDetail(error, e.getMessage()).build(); } } }然后在网关层聚合所有 Agent 的健康状态生成一个customer-flow的总健康视图。当“查订单”和“验身份”都健康但“话术生成”的 LLM 服务不可用时总健康状态是DOWN但你能一眼看出是哪个环节出了问题。5.4 日志的上下文透传一条 Trace 串起所有节点所有日志必须带上traceId和flowId。我用 MDCMapped Diagnostic Context实现// 在 Flow 执行入口 String traceId MDC.get(traceId); String flowId UUID.randomUUID().toString(); MDC.put(traceId, traceId); MDC.put(flowId, flowId); try { flow.execute(input); } finally { MDC.clear(); // 清理避免线程复用污染 }然后在每个 Agent 的日志里自动带上这些字段log.info(Start processing order {}, order.getOrderId()); // 输出[traceIdabc123, flowIddef456] Start processing order 123456配合 ELK 或 Loki你可以用一个traceId查到这次客服请求在所有五个子智能体中的完整执行日志、耗时、参数和返回值。这才是真正的端到端可观测。提示不要在日志里打印敏感信息如用户手机号、身份证号。我专门写了SensitiveDataMasker工具类在日志输出前自动脱敏。这是合规底线不是可选项。6. 从 Demo 到生产那些文档里不会写的实战经验最后分享几个我在模拟项目X中踩过的、血淋淋的坑以及对应的解决方案。这些经验比任何架构图都珍贵。6.1 坑Flow 编译时类型检查太严格DTO 微小变更导致编译失败现象OrderQueryResponse加了一个deliveryTime字段所有引用它的 Agent 都编译不过因为 Flow 认为输入输出类型不匹配。原因Flow 的类型检查是基于 Java Class 的全限定名。哪怕只是加了个 getterClass 对象就变了。解法用通用 DTO 显式转换。定义一个FlowContext作为所有节点的统一输入输出载体public class FlowContext { private MapString, Object data new HashMap(); public T T get(String key, ClassT type) { ... } public void put(String key, Object value) { ... } }所有 Agent 的输入输出都变成FunctionMonoFlowContext, MonoFlowContext。节点之间通过context.put(order, order)和context.get(order, Order.class)传递数据。这样DTO 的变更只影响具体节点的put/get逻辑不影响 Flow 编译。6.2 坑A2A 调用在 Kubernetes 环境下 DNS 解析失败现象本地开发一切正常部署到 K8s 后A2A 调用http://order-service:8080报UnknownHostException。原因K8s 的 Service DNS 名称是order-service.namespace.svc.cluster.local而order-service:8080是短名称需要 CoreDNS 配置支持。解法永远用 FQDN全限定域名。在application.yml中配置spring: ai: alibaba: a2a: base-url: http://order-service.default.svc.cluster.local:8080或者更优雅地用 Spring Cloud Kubernetes 的DiscoveryClient自动解析Bean public A2AAgent checkOrderAgent(DiscoveryClient discoveryClient) { ServiceInstance instance discoveryClient.getInstances(order-service).get(0); String baseUrl http:// instance.getHost() : instance.getPort(); return A2AAgent.builder().baseUrl(baseUrl).functionName(checkOrder).build(); }6.3 坑LLM 生成的话术包含幻觉导致客服回复错误信息现象“查物流”返回“已签收”但 LLM 在话术里写成了“预计明天送达”。原因LLM 是概率模型它会“脑补”不存在的信息。不能无条件信任它的输出。解法LLM 输出后必须做结构化校验。在ScriptGeneratorAgent中增加一层PostProcessorprivate ScriptResponse postProcess(ScriptResponse raw) { // 规则1如果物流状态是 DELIVERED话术里不能出现 预计、明天 等未来时态词 if (DELIVERED.equals(raw.getLogisticsStatus()) raw.getText().matches(.*预计.*|.*明天.*|.*后天.*)) { raw.setText(raw.getText().replaceFirst(预计.*, 已签收。)); } // 规则2订单金额必须和原始数据一致防止篡改 if (!raw.getOrderAmount().equals(originalOrder.getAmount())) { raw.setOrderAmount(originalOrder.getAmount()); } return raw; }这层规则引擎是 LLM 应用的“安全阀”。它不追求 100% 覆盖但能拦截 90% 的高危幻觉。6.4 坑Flow 执行过程中JVM 内存飙升GC 频繁现象高并发下Flow.execute()调用后老年代内存持续增长最终 OOM。原因Flux和Mono的背压策略不当。比如某个 Agent 返回了Flux.just(...)但下游消费慢数据就在内存里堆积。解法强制所有 Flux 使用onBackpressureBuffer()并设上限Bean public Flow flow() { return Flow.builder() .add(checkOrder, checkOrderAgent) .add(verifyIdentity, verifyIdentityAgent) .bufferSize(1000) // 关键全局设置背压缓冲区大小 .build(); }同时在每个 Agent 的apply()方法里对上游Flux显式调用.onBackpressureBuffer(100)。100 是经验值根据你的平均消息大小和内存预算调整。6.5 坑本地测试 Flow 很快压测时延迟暴增 10 倍现象单线程测试Flow 平均耗时 200msJMeter 100 并发平均耗时 2000ms。原因线程竞争。AgentExecutionContext是ThreadLocal但在 Reactor 的parallel()操作符里线程会切换ThreadLocal丢失。解法用 Reactor 的Context替代ThreadLocal。重写AgentExecutionContextpublic class AgentExecutionContext { private static final ContextView.KeyAgentExecutionContext KEY ContextView.key(agent.execution.context); public static AgentExecutionContext current() { return ReactorContext.current().getOrDefault(KEY, new AgentExecutionContext()); } public static T MonoT withContext(MonoT mono) { return mono.contextWrite(ctx - ctx.put(KEY, current())); } }然后在 Flow 执行入口MonoFluxObject result AgentExecutionContext.withContext( flow.execute(Mono.just(input)) );这样上下文就能随着Mono的流转自动在不同线程间传递彻底解决线程切换导致的上下文丢失和性能
阅读完成 · 觉得有帮助？
