企业官网建设流程全解析

1. 从“五个子智能体”到客服流水线：一次对 Spring AI Alibaba 内置编排能力的逆向解剖

我第一次看到 Spring AI Alibaba 的Flow和A2A（Agent-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链式调用，但它的本质，是一个将声明式节点定义编译为可执行Function<Flux<Input>, Flux<Output>>管道的编译器。这句话听起来绕，但理解它，是避开后续所有坑的前提。

我们先看一个最简 Flow 定义：

Flow flow = Flow.builder() .add("checkOrder", checkOrderAgent) .add("verifyIdentity", verifyIdentityAgent) .add("getLogistics", getLogisticsAgent) .build();

这段代码执行后，flow对象并不是一个“运行时容器”，而是一个被编译好的、可复用的函数式管道。它内部会做三件关键事：

1. 节点拓扑解析：将.add()的顺序转化为有向边，构建 DAG（有向无环图）。注意，这里默认是线性依赖，但你可以用.add("A", a).add("B", b, "A")显式指定 B 依赖 A，从而支持扇入（多个节点输出汇入一个节点）；
2. 输入输出契约绑定：每个 Agent 必须实现Function<Input, Output>接口（或其 Reactive 变体Function<Flux<Input>, Flux<Output>>）。Flow 在编译时会检查checkOrderAgent的输入类型是否与前一节点的输出类型兼容，不兼容则编译期报错，而不是运行时报ClassCastException；
3. 执行上下文注入：Flow 会为每个节点创建一个AgentExecutionContext，其中包含本次调用的全局correlationId、requestId、timeout、retryConfig，以及一个可读写的Map<String, Object>用于跨节点传递临时状态（比如“验身份”成功后把用户等级写进去，“话术生成”就能据此调整语气）。

这才是为什么它叫“内置编排”——它不依赖外部工作流引擎（如 Camunda），所有调度逻辑都运行在当前 Spring 应用的同一个 JVM 里，没有网络开销，没有序列化成本，也没有额外的运维组件。但代价是：它无法跨进程编排。如果你的五个子智能体分别部署在五个微服务里，Flow 就不适用，你得切到 A2A 或其他 RPC 方案。

那么，Flow 如何处理分支？比如“验身份失败”时跳过“查物流”，直接进“话术生成”？答案是：用ConditionalNode。它不是 if-else 语法糖，而是一个实现了Function<Input, 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 默认是“任一节点失败，整个流程中断”。但客服场景需要更精细的控制：比如“查物流”超时，可以降级返回缓存数据，不影响整体流程。这时要用RetryableNode：

RetryableNode 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函数是核心。它接收原始输入和异常，返回一个Mono<Output>作为降级结果。这个设计比 Hystrix 的 fallbackMethod 更灵活，因为你可以在降级逻辑里调用缓存、发告警、甚至调用另一个备用 Agent。

3. A2A 远程调用：不是 RPC，是带语义的函数代理

当五个子智能体分散在不同模块、甚至不同 Spring Boot 应用里时，Flow 的本地编排就失效了。这时候，Spring AI Alibaba 提供的 A2A（Agent-to-Agent）调用机制，就成了唯一选择。但必须立刻划清界限：A2A 不是传统 RPC，不是 gRPC，也不是 REST Client 封装。它是一种基于 Spring Cloud Function 和 WebClient 的、面向智能体交互场景的语义化代理。

它的核心思想是：让远程 Agent 像本地 Function 一样被调用，同时自动处理序列化、超时、重试、熔断、链路追踪，且调用方完全感知不到网络细节。

我们以“查订单 Agent”为例。假设它部署在order-service应用中，暴露一个标准的 Spring Cloud Function Endpoint：

// order-service 中 @Bean public Function<OrderQueryRequest, OrderQueryResponse> checkOrder() { return request -> { // 真实查库逻辑 return orderService.queryByOrderId(request.getOrderId()); }; }

在customer-service（客服主应用）中，你不需要写任何 HTTP Client 代码。只需声明一个A2AAgentBean：

@Bean 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)这一行，背后发生了什么？

1. 请求构造：A2A 会将input（OrderQueryRequest）序列化为 JSON，通过 WebClient 发送到http://order-service:8080/functions/checkOrder；
2. 超时与重试：由timeout和retryConfig控制，失败时自动重试，重试间有退避（backoff）；
3. 响应反序列化：收到OrderQueryResponseJSON 后，自动反序列化为 Java 对象；
4. 错误映射：如果order-service返回 4xx/5xx，A2A 会包装成A2AException，并携带原始 HTTP 状态码和错误消息，方便 Flow 的fallback处理；
5. 链路追踪注入：自动将当前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)返回的是Mono<OrderQueryResponse>，而不是OrderQueryResponse。这意味着你必须把整个 Flow 升级为 Reactive：

Flow flow = Flow.builder() .add("checkOrder", checkOrderAgent) // 返回 Mono .add("verifyIdentity", verifyIdentityAgent) // 也必须是 Mono<...> 类型 .build(); // 执行时 Mono<Flux<Object>> result = flow.execute(Mono.just(input));

这个升级不是可选的，是必须的。因为只有 Reactive 模式才能实现真正的非阻塞 I/O，才能支撑每秒数千次的客服请求。而本地 Agent 也必须适配：verifyIdentityAgent的类型得是Function<Mono<Input>, Mono<Output>>，而不是Function<Input, 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 Function<Mono<OrderQueryRequest>, Mono<OrderQueryResponse>> { private final OrderService orderService; private final CacheManager cacheManager; @Override public Mono<OrderQueryResponse> apply(Mono<OrderQueryRequest> requestMono) { return requestMono .flatMap(request -> { // Step 1: 先查缓存（毫秒级） String cacheKey = "order:" + request.getOrderId(); Mono<OrderQueryResponse> 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 Function<Mono<OrderQueryResponse>, Mono<IdentityVerificationResult>> { private final RiskAssessmentService riskService; private final UserService userService; @Override public Mono<IdentityVerificationResult> apply(Mono<OrderQueryResponse> 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 查物流 Agent：A2A 调用的典型范式

这是第一个 A2A 调用。目标服务logistics-service提供getLogisticsFunction。我的LogisticsAgent就是前面展示的A2AAgent实例。但生产环境必须加两道保险：

1. 本地缓存兜底：即使logistics-service宕机，也能返回 5 分钟前的物流快照；
2. 异步刷新：当 A2A 调用成功后，启动一个异步任务，把最新物流信息推送到 Redis，供下次缓存命中。

public class LogisticsAgent implements Function<Mono<OrderQueryResponse>, Mono<LogisticsResponse>> { private final A2AAgent logisticsA2A; private final CacheManager cacheManager; @Override public Mono<LogisticsResponse> apply(Mono<OrderQueryResponse> responseMono) { return responseMono .flatMap(response -> { String cacheKey = "logistics:" + response.getOrderId(); // 先查缓存 Mono<LogisticsResponse> 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 生成话术 Agent：LLM 驱动的上下文融合

这是整个流水线的“大脑”。输入是前面所有节点的聚合结果（订单、身份、物流），输出是ScriptResponse（含text: String,tone: FORMAL/CASUAL,nextStep: TRANSFER_TO_AGENT/END_CONVERSATION）。

我用 Spring AI Alibaba 的ChatClient封装了一个ScriptGeneratorAgent：

public class ScriptGeneratorAgent implements Function<Mono<Map<String, Object>>, Mono<ScriptResponse>> { private final ChatClient chatClient; @Override public Mono<ScriptResponse> apply(Mono<Map<String, 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(Map<String, 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 Function<Mono<ScriptResponse>, Mono<TransferInstruction>> { @Override public Mono<TransferInstruction> apply(Mono<ScriptResponse> 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和status（success/failed）打点，P95/P99 延迟一目了然；
• spring.ai.flow.node.execution.count：按flow.name和node.name统计每个节点的执行次数，快速定位瓶颈节点（比如“查物流”调用量突增，说明前端页面有 bug）；
• spring.ai.flow.node.error.count：按flow.name,node.name,error.type（TimeoutException, A2AException, RuntimeException）统计，精准定位故障根因。

这些指标，配合 Grafana 看板，能回答所有关键问题：哪个节点最慢？哪个节点错误最多？错误类型分布如何？流量峰值出现在什么时间？

5.2 A2A 调用的深度追踪

A2A 的baseUrl是服务名，但实际调用的是具体实例。我扩展了A2AAgent，在每次调用前后，记录targetInstance: ip:port和responseTimeMs，并上报为 Tagged Metric：

// 扩展 A2AAgent 的 execute 方法 public Mono<Response> 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{function="checkOrder", instance="10.0.1.12:8080", status="200"} 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。我用 MDC（Mapped 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()); // 输出：[traceId=abc123, flowId=def456] Start processing order 123456

配合 ELK 或 Loki，你可以用一个traceId，查到这次客服请求在所有五个子智能体中的完整执行日志、耗时、参数和返回值。这才是真正的端到端可观测。

提示：不要在日志里打印敏感信息（如用户手机号、身份证号）。我专门写了SensitiveDataMasker工具类，在日志输出前自动脱敏。这是合规底线，不是可选项。

6. 从 Demo 到生产：那些文档里不会写的实战经验

最后，分享几个我在模拟项目X中踩过的、血淋淋的坑，以及对应的解决方案。这些经验，比任何架构图都珍贵。

6.1 坑：Flow 编译时类型检查太严格，DTO 微小变更导致编译失败

现象：OrderQueryResponse加了一个deliveryTime字段，所有引用它的 Agent 都编译不过，因为 Flow 认为输入输出类型不匹配。

原因：Flow 的类型检查是基于 Java Class 的全限定名。哪怕只是加了个 getter，Class 对象就变了。

解法：用通用 DTO + 显式转换。定义一个FlowContext作为所有节点的统一输入输出载体：

public class FlowContext { private Map<String, Object> data = new HashMap<>(); public <T> T get(String key, Class<T> type) { ... } public void put(String key, Object value) { ... } }

所有 Agent 的输入输出都变成Function<Mono<FlowContext>, Mono<FlowContext>>。节点之间通过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中，增加一层PostProcessor：

private 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 平均耗时 200ms；JMeter 100 并发，平均耗时 2000ms。

原因：线程竞争。AgentExecutionContext是ThreadLocal，但在 Reactor 的parallel()操作符里，线程会切换，ThreadLocal丢失。

解法：用 Reactor 的Context替代ThreadLocal。重写AgentExecutionContext：

public class AgentExecutionContext { private static final ContextView.Key<AgentExecutionContext> KEY = ContextView.key("agent.execution.context"); public static AgentExecutionContext current() { return ReactorContext.current().getOrDefault(KEY, new AgentExecutionContext()); } public static <T> Mono<T> withContext(Mono<T> mono) { return mono.contextWrite(ctx -> ctx.put(KEY, current())); } }

然后在 Flow 执行入口：

Mono<Flux<Object>> result = AgentExecutionContext.withContext( flow.execute(Mono.just(input)) );

这样，上下文就能随着Mono的流转，自动在不同线程间传递，彻底解决线程切换导致的上下文丢失和性能