今年年初我们团队在做一个内部任务自动化平台的时候被大语言模型调用的延迟和线程占用折磨得够呛。每次让模型补全一段内容一个线程就要干等两三秒再叠加连续的工具调用线程池压力直线上升最后干脆下决心用 Java 25 和 Spring Boot 4 从零写了一个响应式 AI Agent 框架并且已经在开源社区把代码放了出来。这个框架的代号叫 AgentFlux思路其实很朴素把 Agent 的推理、工具调用、记忆读写全部放到非阻塞的流式链路上借助响应式背压和 SSE 输出让一次包含多步工具调用的交互不再依赖大量阻塞线程。整个框架不绑定特定模型厂商也不强制你用某种 Agent 套路核心就是一个可编排的事件流Thinking、ToolCall、ToolResult、TokenChunk、FinalAnswer。适合正在纠结 Agent 怎么上生产环境的 Java 团队也适合想把 WebFlux 和 LLM 真正揉到一起的同学参考。1. 为什么还要自己写一个 Agent 框架1.1 现有方案的堵点先泼一盆冷水市面上的 Agent 框架很多但 Java 生态里面的选择真的少。大部分让人眼前一亮的产品都集中在 Python 生态Java 团队想直接搬过来要么要起一堆子进程要么要自己搭一套调用协议维护成本比写业务代码还高。更深层的痛点在于阻塞式 HTTP 调用。老写法很简单就是RestTemplate或者HttpClient同步请求模型接口模型输出几百个 token线程就干等着。Agent 场景和普通接口还不一样经常需要多次推理加工具调用一次请求可能要分五六个来回。并发稍高一点线程池就膨胀再强的机器也顶不住这种“一个会话占一个线程”的模型。还有一个很小的痛点但是特别影响体验SSE 流式输出。不少框架所谓的“流式”其实是先同步等到完整结果再一股脑返回给前端。真正做过生成式 AI 应用的都懂用户要的是第一个字尽快出来而不是等三秒看到一整段话。很多存量框架在设计时没有把流式作为一等公民后面补流式接口就补成了一座豆腐渣工程。1.2 Java 25 和 Spring Boot 4 能带来什么选择 Java 25 不是追新而是它确实解决了几个基础问题。GC 暂停和虚拟线程调度的成熟度比早期版本好很多像 ZGC 配合低延迟场景已经算是生产可用了。对响应式框架来说JVM 只是底座更关键的是库生态。Spring Boot 4 相比 3.x 最大的变化是响应式栈的地位明显提升WebFlux 不再是“另一个选择”而是很多默认场景的一等公民。它的自动配置对 Reactor 友好得多WebClient 的连接池、Netty 的启动调优都做得更顺手。我们在做选型对比时还发现Spring Boot 4 对java.util.concurrent和虚拟线程的桥接也更自然允许你在非阻塞链路里通过适配器包一层同步阻塞调用而不至于把整个事件循环阻塞掉。当然如果你只想要“不阻塞”用虚拟线程写同步代码也是一种解法。这点我在下面单独讲。1.3 虚拟线程与响应式的边界到底怎么划这是我在团队内部被问过最多的问题Java 21 开始就有虚拟线程了Java 25 虚拟线程更加成熟直接用同步写法不好吗为什么还要用 Project Reactor 那套被吐槽学不会的东西我的答案一直很明确虚拟线程解决的是“线程阻塞导致资源浪费”的问题但解决不了“流式传输、背压、多路复用长连接”的问题。Agent 场景里模型输出是一个持续几秒甚至几十秒的数据流前端需要 SSE 一点一点接收这个过程天然需要异步流水线。如果每个连接都挂一个虚拟线程等读写大量连接的情况下文件描述符、调度切换、内存占用仍然会快速上升虽然有改善但不是最优雅的解法。更重要的是Spring WebFlux 已经沉淀了完善的背压机制。当模型输出太快而下游数据库写入来不及时Reactor 的Subscription会主动控制请求量不至于把内存打爆。这个能力在虚拟线程模型里需要自己重新实现。所以我们的原则是IO 密集型流式链路走响应式偶尔碰到的阻塞型 SDK 才用虚拟线程包装后放到调度器里两者各自在擅长的地方干活。AgentFlux 的底层运行时因此是 Project Reactor虚拟线程只在适配层做补充。2. AgentFlux 的架构设计把 Agent 拆成可编排的响应式节点2.1 核心抽象一切皆是事件流框架最开始的设计只定了一个接口没有急着设计一堆类继承体系。核心接口非常简单public interface Agent { FluxAgentEvent run(AgentRequest request); }AgentEvent是一个密封接口主要事件类型包括Thinking、ToolCall、ToolResult、MessageChunk和FinalAnswer。为什么一定要用事件流而不是直接返回AgentResult因为 Agent 的本质不是一次函数调用而是一个持续交互的过程。事件流允许调用方随时看到当前 Agent 在想什么、调了什么工具、正在流式输出哪些 token也为后面做审计和重放留下了空间。用户在框架里写一个 Agent实际就是在组合这些事件。这个设计很像把 Agent 当成了“可观测的流水线”而不是黑盒。我很喜欢这种设计因为它天然适配前端的 SSE 场景后端把事件流一个接一个推到浏览器前端可以在转圈的同时实时展示当前进度。2.2 会话状态与事件溯源多轮对话是 Agent 的基本要求但状态管理经常被写成一坨垃圾代码。AgentFlux 没有采用传统的“会话对象存 Map”的方式而是把状态看成事件的投影只要保留完整事件流就能还原任意时刻的对话快照。实现上非常简单每次请求会生成一个sessionId所有的事件都追加到持久化事件表里。重建上下文时只需要按sessionId读取事件流用 reducer 方法逐个应用即可。这样既解决了多实例部署时的会话共享问题又天然支持“回放排查”用户说效果不对直接把事件流拉出来看就能定位是模型推理错了还是工具结果解析错了。事件流本身也是响应式的。读写侧用 R2DBC 对接 PostgreSQL追加事件是一个异步写操作不会阻塞模型输出链路。当然事件表写多了性能会有压力所以我们在框架里加了缓冲区当事件攒到一定量或时间窗口到了再批量落库。这样单条交互的额外写延迟几乎可以忽略。2.3 工具注册中心从注解到 JSON Schema 的自动转换工具调用是 Agent 和外部系统交互的触手但工具定义的体验必须足够简单。AgentFlux 提供了一套注解式 API开发者只需要写一个普通方法框架负责把方法签名转换成模型需要的 JSON Schema。AgentTool(name query_weather, description 查询某个城市的实时天气) public WeatherResult queryWeather(AgentArg(city) String city) { return weatherService.query(city); }在启动阶段框架会扫描所有AgentTool注解方法利用反射提取方法名、参数名、参数描述然后生成 OpenAPI 风格的 JSON Schema。整个过程在 Netty 线程之外完成不会影响响应式启动流程。工具返回对象会统一序列化成 JSON作为ToolResult事件推回到模型链路中。这里有个实现陷阱反射拿不到默认参数名因为 Java 编译后参数名会丢失。所以框架提供了-parameters编译选项的检测如果检测不到就尝试从注解里取。我们自己的 Starter 里直接带了 Maven 插件配置避免用户漏配。2.4 大模型网关适配层很多团队迷信用一个厂商的 SDK结果几个月后想换模型SDK 一换整个 Agent 逻辑跟着重写。AgentFlux 从一开始就把模型网关抽象成了独立接口模型调用和 Agent 编排完全解耦。public interface ReactiveChatGateway { FluxChatChunk chatStream(ChatRequest request); }外部只需要实现这个接口把厂商 SDK 适配成响应式流。框架内置了几个适配器但用户完全可以自己写一个私有适配器把内部的自研推理服务接进来。统一网关还有个额外好处可以方便地在中间插一层日志、限流、计量逻辑不用担心污染 Agent 编排代码。不同模型服务商的接口差异很大有的事流式 SSE有的是普通 JSON有的支持工具调用有的只支持纯文本。适配层的核心工作就是把这些差异消化掉向上层输出一致的ChatChunk。工具调用协议也在网关层统一模型返回的 tool_calls 结构无论如何变化最终都被映射成 AgentFlux 的ToolCall事件。2.5 规划器与回退策略Agent 的能力上限很大程度上取决于规划器。我们做了三个可插拔的实现ReActPlanner让模型循环思考、工具调用、观察结果PlanAndExecutePlanner先让模型生成完整计划再逐步执行FixedChainPlanner适合一些稳定的固定流程比如先查库存再下单。规划器不是越复杂越好。我们内部复盘时发现90% 的生产场景用 ReAct 就够了PlanAndExecute 在需要提前声明步骤的场合更合适。框架允许用户在运行前通过PlannerSelector指定策略也允许在Thinking事件中根据模型输出动态切换。为了不让规划器成为新的故障点AgentFlux 还支持回退策略当模型连续多次返回错误格式或工具调用超时规划器会退回到一个简单的“重试-降级-最终答复”链路而不是僵死在那里。这在生产里太重要了因为大模型偶尔就是会说胡话。3. 实操从零构建一个响应式 AI Agent 示例3.1 环境准备与依赖引入先准备好环境JDK 25Maven 3.9PostgreSQL 14用来存事件流和会话。如果你不想接数据库也可以先跑内存模式把 Fast 启动跑通再加持久化。Maven 依赖只需引入两个包dependency groupIdio.agentflux/groupId artifactIdagentflux-spring-boot-starter/artifactId version0.1.0/version /dependency dependency groupIdorg.springframework.boot/groupId artifactIdspring-boot-starter-webflux/artifactId /dependency启动类和普通 Spring Boot 项目完全一致用的是SpringApplication.run加EnableAgentFlux开启框架能力。它会自动装配事件存储、工具扫描和大模型网关。3.2 定义一个带工具的 Agent工具就沿用刚才的天气查询再补一个“写入日程”的工具。这些方法不需要继承任何父类写在任意 Spring Bean 里即可。Service public class ScheduleTools { AgentTool(name add_schedule, description 把一段文字内容添加到用户的日程表里) public MonoScheduleEntry addSchedule(AgentArg(content) String content, AgentArg(dateTime) LocalDateTime dateTime) { return Mono.fromCallable(() - scheduleRepo.save(new ScheduleEntry(content, dateTime))); } }工具方法返回Mono或普通对象都可以框架会自动把普通对象包装成Mono这样不会阻塞事件循环。返回值序列化为 JSON 后作为ToolResult注入模型上下文。然后创建一个 AgentBean public Agent scheduleAgent(ReactiveChatGateway gateway) { return AgentFlux.newAgentBuilder() .model(default-llm) .systemPrompt(你是一个日程助手当用户提到天气和安排时优先使用工具。) .addTools(ScheduleTools.class) .planner(new ReActPlanner()) .build(); }3.3 通过 SSE 提供流式接口写一个 Controller把 Agent 的事件流直接映射为 SSE。前端每秒都能看到模型吐字体验和原生的 ChatGPT 很像。RestController public class ChatController { PostMapping(value /chat, produces MediaType.TEXT_EVENT_STREAM_VALUE) public FluxServerSentEventAgentEvent chat(RequestBody ChatRequest request) { AgentRequest agentRequest AgentRequest.builder() .sessionId(request.sessionId()) .input(request.message()) .build(); return agent.run(agentRequest) .map(event - ServerSentEvent.builder(event) .event(event.type().name()) .build()) .onErrorResume(ex - Mono.just( ServerSentEvent.builder( new AgentError(ex.getMessage())) .event(Error) .build() )); } }这段代码看起来不长但内含一个容易犯的错误如果映射阶段抛异常整个流会中断。所以我在最后加了onErrorResume把异常转换成Error事件保证前端能收到结束信号而不是一直转圈等待。3.4 用 StepVerifier 测试工具调用链路响应式代码没法像传统 Controller 那样直接 mock 后拿返回值好在 Spring 生态里有StepVerifier。我们测试一个“查询天气再加日程”的完整流程Test void shouldCallToolAndSchedule() { Agent agent buildTestAgent(); StepVerifier.create(agent.run(AgentRequest.of(北京明天适合跑步吗如果适合就帮我写一条日程))) .expectNextMatches(ev - ev.type() AgentEventType.THINKING) .expectNextMatches(ev - ev.type() AgentEventType.TOOL_CALL) .expectNextMatches(ev - ev.type() AgentEventType.TOOL_RESULT) .expectNextMatches(ev - ev.type() AgentEventType.MESSAGE_CHUNK) .expectComplete() .verify(); }这个测试最核心的价值是锁定了事件顺序。模型输出可能不稳定但工具调用流程必须符合预期。如果哪天 LLM 供应商调整了接口导致事件序列变了测试会第一时间报警。3.5 连接真实模型时的参数配置最后一个见真章的地方配置文件。下面是一个最简单的示例agentflux: gateway: endpoint: ${AI_ENDPOINT} api-key: ${AI_API_KEY} default-model: default-llm timeout: 30s session: store: memory这里的endpoint指向你自己统一封装过的模型服务地址可以是开源部署的推理服务也可以是云厂商的 API。AgentFlux 不直接绑定厂商只认这个地址。超时建议设置为 30s 以上因为带工具调用的多轮交互可能需要接近 20s。这个配置看起来简单但我见过不少团队把模型超时时间设成 3s然后天天收到超时报错浪费时间。4. 性能数据与开源后的运营心得4.1 内部压测对比框架从原型阶段就安排了一轮压测目标是看看到底省了多少线程。测试环境是 4C8G 的虚拟机和内存模式存储模型服务用一个静态 Mock 服务模拟每个请求返回约 300 个 token工具调用固定两步。方案并发用户数线程池/调度线程P95响应时间是否稳定传统同步Spring MVC 阻塞HTTP5005005.6s线程池频繁满虚拟线程同步方案500500虚拟线程5.8s内存占用偏高AgentFlux 响应式方案500约32个线程5.2s稳定无拒绝结果并不意外响应式方案并没有让单次请求变快但它用极少的线程扛住了同等的并发。对生产环境来说省下来的线程资源可以用来跑更多计算密集型任务或者直接把应用容器规格降一级省钱又降碳。再说一个有意思的现象虚拟线程方案在压测时表现不差但是当连接数达到 2000 时内存比响应式方案高出近 30%。原因在于每个连接都是一个栈模型流式输出期间的暂存数据积累会推高堆内存。响应式方案通过背压主动消费反而把每个连接的中间态控制在很小的范围内。4.2 发布前要做的三件事开源不是把代码推到仓库就算完事我们在发版前补了很多东西总结出最重要的三个第一完整的示例工程。框架好不好用用户第一眼看的不是文档里的 UML 图而是一个能跑起来的 demo。我们额外做了一个examples/weather-schedule目录里面包含完整的 Controller、工具类和测试用例。这个示例工程在发布后成为了访问量最大的路径。第二明确的 issue 模板和贡献指南。人均经历过的“报 bug 不附上下文”在开源里太常见了。我们提供了模板要求必须贴出事件流日志、Java 版本和框架版本。靠这个模板几乎每个 issue 都能在 24 小时内定位问题。第三许可协议。我们选了宽松的Apache-2.0并明确写了“可以商用改造成邪教项目也没问题但别来问我们为什么不兼容你的私有版本”。选宽松协议的核心目的是让企业用户少一个不引入的理由。4.3 社区反馈中最有价值的几次修正框架开源前我们自觉设计得很完善结果第一个周就被社区用户指出了两个问题。第一个问题是事件表写入的背压未生效。如果我们直接把Flux.merge(addEvent, modelStream)合并事件写入慢的时候模型输出流不会自动降速。社区用户建议每个事件在落库时不直接subscribe而是通过Flux.create配合Disposable来控制请求量。我们采纳后事件积压少了模型输出也没有断流。第二个问题是 SSE 连接断开后Agent 线程不会自动取消。当用户刷页面或关掉浏览器下游的模型调用可能还在继续跑浪费 token 和资源。社区用户提了一个很实在的建议在.doOnCancel里调用session.cancel()并同步把状态标记为中断。我们一开始担心取消会污染事件流后来用了finally保证计数器和审计事件正常关闭才算彻底解决。这两个修正让我体会到开源项目的代码质量不一定全靠自己高质量的使用者就是最好的 QA 团队。5. 常见问题与排查技巧实录5.1 “为什么我的背压没生效”这是新手最容易踩的坑。很多人以为只要用了Flux背压就自动无损。其实不然只有在上游和下游之间建立了明确的Subscription请求量协调时背压才会起作用。常见错误是直接调用.buffer()把整个流攒成 List再一次性消费。一旦用了聚合操作前面的缓存在内存里直接爆掉。我们的建议不要用blockLast()也不要在map里做阻塞 IO。如果某个步骤确实慢请显式声明.limitRate(10)让上游每秒只推 10 个元素。排查方法也简单给流加上.log(agent)观察请求量和下游是否延迟。如果日志里出现大量request(1)紧接着onNext说明背压链路是通的如果request(unbounded)连续出现恭喜你断头路了。5.2 ThreadLocal 在响应式链路里失效Java 里随手写个TraceIdUtil.set(traceId)在 WebFlux 里不会生效因为同一条业务链路会被多个线程交替执行。最直观的现象是日志里的 traceId 一会儿有一会儿没有。AgentFlux 的处理是引入 Reactor 的Context把 TenantId、TraceId、SessionId 都放到Context里。具体工具方法如果需要上下文不直接访问静态类而是通过reactor.util.context.ContextView读取。我们在框架层做了一个AgentContextHolder把Context暴露成自定义的协程风格的存取接口。这里要注意Context是链路私有且不可变的子流程必须显式传递。如果你在flatMap里启动了新的子流子流默认看不到父流的Context需要.contextWrite或者concatMap保持上下文。这个特性写框架时最容易被忽略但用框架的人几乎必踩。5.3 工具参数反序列化失败大模型返回的工具调用参数并不总是标准 JSON。模型偶尔会输出{ city: 北京, dateTime: 明天 }这种含单引号、缺双引号的非法 JSON。如果用默认的 Jackson 解析直接抛异常Agent 就会中断。我们的解法是配置一个宽松的 JSON 反序列化器开启ALLOW_SINGLE_QUOTES和ALLOW_UNQUOTED_FIELD_NAMES并加一层容错如果解析失败把原始字符串发给网关让它重新生成一次 tool_call。这个重试逻辑只允许一次避免模型陷入死循环。经过这套处理实际线上非法 JSON 的比例从 5% 降到了 0.1% 以下。另外时间参数的解析也要宽容。模型喜欢把“明天上午10点”直接输出为2025-12-03 10:00但系统里 DateTime 格式可能是2025-12-03T10:00:00。我们在工具注解里加了一个dateTimeFormat属性允许声明多种解析格式避免每次都被这类边界问题卡住。5.4 SSE 断开后如何优雅终止前面社区反馈提到的doOnCancel只是其中一环。实践中发现在断开瞬间模型服务商仍在产生数据直接断开会导致它的重试风暴。所以框架增加了一个 grace period收到 cancel 信号后不立刻终止而是等当前 token 块输出完再发送一个Interrupted事件并释放资源。这个缓冲窗口默认是 2 秒可通过配置调整。还有一个要注意的是错误码管理。SSE 协议里网络错误导致前端难分辨是模型异常还是连接断开。我们在事件流里增加了一个AgentEndCause字段取值可能是COMPLETED、CANCELLED、ERROR。前端根据这个字段决定是否展示“网络中断”或“生成失败”用户体验会好很多。5.5 调试思路让 Agent 的每一步都可观测响应式代码最怕黑盒。AgentFlux 为了解决这个问题把每一个关键事件都写入结构化日志格式大概是agentIdschedule-agent sessionIdxxx eventTOOL_CALL toolquery_weather args...。这让我们在排查线上问题时几乎可以像看电影一样回放每一次对话。如果还不满足框架内置了一个/actuator/agent/{id}/events的调试端点通过 SSE 输出指定会话的实时事件流。这个端点默认只在本地环境和测试环境打开因为生产开着等于给所有人看用户对话内容了。调试时我也会用一个小技巧在StepVerifier里把expectNextMatches换成.recordWith(ArrayList::new)然后把收集到的所有事件打出来。这样能很快看出模型是不是在某个工具调用环节卡住了还是事件流的顺序本身就错了。省下的时间足够多喝两杯咖啡。最后再分享一点个人经验这个框架做下来我最大的体会是响应式不是银弹但在 Agent 这种天然流式的场景里它确实比传统阻塞模型好写了太多。虚拟线程可以解决线程阻塞却解决不了流式传输、背压和连接的优雅释放两者不应该是对立关系而是各管一段。开源也不只是把代码扔出去。我见过很多项目作者被 issue 一催就崩溃后来我们养成了每周固定时间看 issue 的习惯而不是被实时通知追着跑。框架的设计也因此变得更有韧性早期那些被社区发现的“背压失效”、“SSE 断开不终止”问题恰恰是只靠自己不会踩到的盲区。如果你们团队也准备在 Java 生态里搞 Agent 框架或者正在纠结响应式和虚拟线程怎么选欢迎直接拿 AgentFlux 当成参考能抄就抄不能抄的部分——比如内部的上下文传播实现、工具注册中心的事件流结构也都值得单独拆出来看看。代码是好代码还是坏代码只有跑起来才知道。
