系列说明一个 Java 后端视角的 Spring AI 渐进式实战教程载体为开源项目「劳小司 · 智能法律助手」。序章技术栈全景与 AI 学习指南阶段一 · 流式对话内核篇1 SSE 流式·停止生成·思考可见化本文/ 篇2 会话记忆压缩与流式滚动体验阶段二 · 工具调用篇1 Function Calling 与法律计算器 / 篇2 联网搜索与工具预算阶段三 · RAG 知识库篇1 起步与底账化 / 篇2 Agentic RAG 与引用可信度 / 篇3 检索质量与体验阶段四 · 多模型路由篇1 五路级联路由阶段五 · 安全与质量门篇1 安全层与强制检索 / 篇2 质量门与评估门禁 / 篇3 指代消解与阻塞隔离阶段六 · 产品化与用户体系篇1 认证·配额·门禁 / 篇2 前端·移动端·身份 / 篇3 劳动法专精与多模态阶段七 · 存储演进与部署篇1 存储迁移 / 篇2 部署契约本篇涉及controller/ChatController.java、service/ChatService.java、前端stores/chat.js。一、今天要做什么打开任意大模型产品回复是一个字一个字蹦出来的而且生成途中可以随时点停止。本篇把这两件事做扎实再加一件很多教程忽略的——把首 token 之前的黑盒等待变成可见的进度SSE 流式输出模型每生成一个 token 就实时推送而不是等全部生成完再返回真正的停止生成点停止后大模型侧的生成也被中断而不只是前端不显示思考过程可见化路由 / 改写 / 检索 / 生成这些前置步骤实时透出推理模型的reasoning_content也流式展示。二、为什么是 SSEServlet 栈又怎么流式方案适配度轮询延迟高、浪费请求直接排除WebSocket全双工对服务端单向推流过重还要处理心跳重连SSE基于 HTTP 的单向推送天然适配 LLM 逐 token 输出断线自动重连本项目主体跑在ServletTomcat栈上绝大多数 Java 后端更熟悉也方便复用 Spring Security 等 Servlet 生态只在流式接口上引入 WebFlux 的FluxServerSentEvent承载 SSE。不整体切 Netty——这正是一个混合技术栈的务实取舍业务用熟悉的 Servlet流式用响应式各取所长。三、一条流上的 8 类事件哨兵标记设计流式聊天不只是把 token 推出去还要在同一条通道里传引用、思考进度、重试信号等。本项目的SSE 事件契约有 8 类事件含义tokenAI 生成的正文片段reasoning模型真实推理内容reasoning_content 透传stage首 token 前的阶段进度路由/改写/检索/生成citation结构化引用法名/条号/相关度 JSONretry质量门审校不通过通知前端清空重接第二轮quota配额拒绝含剩余量与重置时间done生成完成data 为[DONE]error生成异常难点Spring AI 的ChatClient.stream()只产一条 token 流怎么把 citation/stage/retry 这些异构信息也塞进去做法在 token 流里混入哨兵标记以\u0000空字符为前缀如\u0000CITATION\u0000{json}流末尾统一map转成对应的 SSE 事件类型.map(token-{if(token.startsWith(CITATION_MARKER)){returnServerSentEvent.Stringbuilder(token.substring(CITATION_MARKER.length())).event(EVENT_CITATION).build();}if(token.startsWith(STAGE_MARKER)){/* → stage 事件 */}if(token.startsWith(REASONING_MARKER)){/* → reasoning 事件 */}returnServerSentEvent.Stringbuilder(sseData(token)).event(EVENT_TOKEN).build();})用\u0000前缀是因为正常文本几乎不可能出现空字符天然不会和模型输出撞车。这样一条 Flux 承载全部语义前端按addEventListener(citation, ...)分别处理即可。四、真正的停止生成这是本篇最值钱的部分。假停止只是前端不再渲染模型那边还在一个字一个字烧你的 token。真停止要让大模型侧的生成也中断。关键三招① 停止信号走 Redis跨实例可达。stop 请求和 SSE 流可能落在不同实例上所以用 Redis 键chat:stop:{sessionId}作为信号而不是进程内的标志位。②takeUntilOther触发 Reactor cancel 传播。.takeUntilOther(stopTrigger(sessionId))// stopTrigger 以 400ms 轮询 Redis 停止键takeUntilOther一旦其他源发出信号就终止主流并向上游传播 cancel——cancel 一路传到发起 HTTP 流式请求的客户端连接关闭模型侧生成随之中断。这是 Reactor 响应式流的内置能力比手动dispose优雅得多。③ 订阅时清残留、结束时兜底清理。.doOnSubscribe(s-redissonClient.getBucket(STOP_KEY_PREFIXsessionId).delete()).doFinally(signal-redissonClient.getBucket(STOP_KEY_PREFIXsessionId).delete());防止上一轮遗留的停止标志误伤本轮。前端配合双中断点停止时既es.close()关闭 EventSource又POST /chat/stop通知后端——两端都断体验才干净。一个工程取舍stopTrigger的实现是Flux.interval(400ms).filter(stopFlagSet).take(1)——400ms 轮询Redis 标志而非 Pub/Sub 订阅。停止是低频操作400ms 的感知延迟完全可接受却换来实现极简、无需管理订阅连接的生命周期同时 stop 标志设60s TTL即便某条流异常退出没清理信号也会自动过期、不会永久残留。这是一个够用就好、不为了优雅引入额外复杂度的典型权衡。五、思考过程可见化一次法律问答在首 token 到来前后端其实做了不少事读记忆、路由、指代改写、强制检索……全在阻塞等待。若不透出用户面对的是转圈黑盒。做法是用一个unicast sink 实时发射阶段事件与主生成流拼接Sinks.ManyStringstageSinkSinks.many().unicast().onBackpressureBuffer();MonoChatContextpreparedMono.fromCallable(()-prepare(sessionId,userMessage,stageSink,...)).subscribeOn(Schedulers.boundedElastic())// 阻塞预处理隔离到弹性线程.doFinally(s-stageSink.tryEmitComplete()).cache();// cache 防二次订阅重复执行 prepareprepared.subscribe();// 立即启动 preparestage 实时流入 sinkreturnstageSink.asFlux().concatWith(prepared.flatMapMany(this::pipeline))prepare()每走到一步就emitStage(stages, 检索法律知识库…)前端立刻看到进度条滚动。推理模型的reasoning_content同理以REASONING_MARKER透传前端渲染成可折叠的思考面板并算出已思考 N 秒发送 → 首 token 的耗时。这里顺带用到了阻塞隔离subscribeOn(boundedElastic)cache()让 Tomcat 的服务器线程不被首 token 前的同步 IO 独占——这个主题在阶段五·篇3 会专门展开。六、踩坑备忘① GET 把 token 和问题拼进 URL落进 nginx 访问日志。早期/chat/stream?userMessage...法律咨询原文和 JWT 全进了代理日志。改成POST 请求体后日志里只剩路径——这是 P0 级安全整改。②filter不会取消上游takeUntilOther会。想中断流式用filter只是丢弃元素模型那边还在跑必须用takeUntilOther/take这类会向源传播 cancel 的操作符。③prepare被订阅两次 检索做两遍。stageSink 和 pipeline 分别订阅prepared时若不cache()Mono.fromCallable每次订阅都重跑。加.cache() 提前subscribe()启动既让 stage 实时流出又保证 prepare 只执行一次。④ SSE data 前导空格被协议吞。SSE 规范会剥掉data:后的一个前导空格token 恰好以空格开头时就丢了。解决对 token data 做 JSON 编码sseData()前端JSON.parse还原。七、小结概念一句话SSE单向推送逐 token比 WebSocket 轻哨兵标记一条 Flux 混装 8 类事件map 阶段分流真停止Redis 信号 takeUntilOther cancel 传播到模型思考可见化unicast sink 实时发 stagereasoning 透传八、下篇预告流式跑通了但一个多轮助手如果重启就失忆、聊久了上下文爆 token体验照样崩。下一篇我们做会话记忆的持久化与两级压缩并治理前端的流式滚动体验。源码与体验Gitee国内快https://gitee.com/spaserby/laoxiaosi.git GitHub https://github.com/spaserby/laoxiaosi.git 在线演示https://laoxiaosi.noctisblue.com本系列全套代码皆开源觉得这篇有帮助欢迎顺手点颗 ⭐ 【Spring AI 实战 · 阶段一·篇1】SSE 流式聊天、真正的“停止生成“与思考过程可见化
系列说明一个 Java 后端视角的 Spring AI 渐进式实战教程载体为开源项目「劳小司 · 智能法律助手」。序章技术栈全景与 AI 学习指南阶段一 · 流式对话内核篇1 SSE 流式·停止生成·思考可见化本文/ 篇2 会话记忆压缩与流式滚动体验阶段二 · 工具调用篇1 Function Calling 与法律计算器 / 篇2 联网搜索与工具预算阶段三 · RAG 知识库篇1 起步与底账化 / 篇2 Agentic RAG 与引用可信度 / 篇3 检索质量与体验阶段四 · 多模型路由篇1 五路级联路由阶段五 · 安全与质量门篇1 安全层与强制检索 / 篇2 质量门与评估门禁 / 篇3 指代消解与阻塞隔离阶段六 · 产品化与用户体系篇1 认证·配额·门禁 / 篇2 前端·移动端·身份 / 篇3 劳动法专精与多模态阶段七 · 存储演进与部署篇1 存储迁移 / 篇2 部署契约本篇涉及controller/ChatController.java、service/ChatService.java、前端stores/chat.js。一、今天要做什么打开任意大模型产品回复是一个字一个字蹦出来的而且生成途中可以随时点停止。本篇把这两件事做扎实再加一件很多教程忽略的——把首 token 之前的黑盒等待变成可见的进度SSE 流式输出模型每生成一个 token 就实时推送而不是等全部生成完再返回真正的停止生成点停止后大模型侧的生成也被中断而不只是前端不显示思考过程可见化路由 / 改写 / 检索 / 生成这些前置步骤实时透出推理模型的reasoning_content也流式展示。二、为什么是 SSEServlet 栈又怎么流式方案适配度轮询延迟高、浪费请求直接排除WebSocket全双工对服务端单向推流过重还要处理心跳重连SSE基于 HTTP 的单向推送天然适配 LLM 逐 token 输出断线自动重连本项目主体跑在ServletTomcat栈上绝大多数 Java 后端更熟悉也方便复用 Spring Security 等 Servlet 生态只在流式接口上引入 WebFlux 的FluxServerSentEvent承载 SSE。不整体切 Netty——这正是一个混合技术栈的务实取舍业务用熟悉的 Servlet流式用响应式各取所长。三、一条流上的 8 类事件哨兵标记设计流式聊天不只是把 token 推出去还要在同一条通道里传引用、思考进度、重试信号等。本项目的SSE 事件契约有 8 类事件含义tokenAI 生成的正文片段reasoning模型真实推理内容reasoning_content 透传stage首 token 前的阶段进度路由/改写/检索/生成citation结构化引用法名/条号/相关度 JSONretry质量门审校不通过通知前端清空重接第二轮quota配额拒绝含剩余量与重置时间done生成完成data 为[DONE]error生成异常难点Spring AI 的ChatClient.stream()只产一条 token 流怎么把 citation/stage/retry 这些异构信息也塞进去做法在 token 流里混入哨兵标记以\u0000空字符为前缀如\u0000CITATION\u0000{json}流末尾统一map转成对应的 SSE 事件类型.map(token-{if(token.startsWith(CITATION_MARKER)){returnServerSentEvent.Stringbuilder(token.substring(CITATION_MARKER.length())).event(EVENT_CITATION).build();}if(token.startsWith(STAGE_MARKER)){/* → stage 事件 */}if(token.startsWith(REASONING_MARKER)){/* → reasoning 事件 */}returnServerSentEvent.Stringbuilder(sseData(token)).event(EVENT_TOKEN).build();})用\u0000前缀是因为正常文本几乎不可能出现空字符天然不会和模型输出撞车。这样一条 Flux 承载全部语义前端按addEventListener(citation, ...)分别处理即可。四、真正的停止生成这是本篇最值钱的部分。假停止只是前端不再渲染模型那边还在一个字一个字烧你的 token。真停止要让大模型侧的生成也中断。关键三招① 停止信号走 Redis跨实例可达。stop 请求和 SSE 流可能落在不同实例上所以用 Redis 键chat:stop:{sessionId}作为信号而不是进程内的标志位。②takeUntilOther触发 Reactor cancel 传播。.takeUntilOther(stopTrigger(sessionId))// stopTrigger 以 400ms 轮询 Redis 停止键takeUntilOther一旦其他源发出信号就终止主流并向上游传播 cancel——cancel 一路传到发起 HTTP 流式请求的客户端连接关闭模型侧生成随之中断。这是 Reactor 响应式流的内置能力比手动dispose优雅得多。③ 订阅时清残留、结束时兜底清理。.doOnSubscribe(s-redissonClient.getBucket(STOP_KEY_PREFIXsessionId).delete()).doFinally(signal-redissonClient.getBucket(STOP_KEY_PREFIXsessionId).delete());防止上一轮遗留的停止标志误伤本轮。前端配合双中断点停止时既es.close()关闭 EventSource又POST /chat/stop通知后端——两端都断体验才干净。一个工程取舍stopTrigger的实现是Flux.interval(400ms).filter(stopFlagSet).take(1)——400ms 轮询Redis 标志而非 Pub/Sub 订阅。停止是低频操作400ms 的感知延迟完全可接受却换来实现极简、无需管理订阅连接的生命周期同时 stop 标志设60s TTL即便某条流异常退出没清理信号也会自动过期、不会永久残留。这是一个够用就好、不为了优雅引入额外复杂度的典型权衡。五、思考过程可见化一次法律问答在首 token 到来前后端其实做了不少事读记忆、路由、指代改写、强制检索……全在阻塞等待。若不透出用户面对的是转圈黑盒。做法是用一个unicast sink 实时发射阶段事件与主生成流拼接Sinks.ManyStringstageSinkSinks.many().unicast().onBackpressureBuffer();MonoChatContextpreparedMono.fromCallable(()-prepare(sessionId,userMessage,stageSink,...)).subscribeOn(Schedulers.boundedElastic())// 阻塞预处理隔离到弹性线程.doFinally(s-stageSink.tryEmitComplete()).cache();// cache 防二次订阅重复执行 prepareprepared.subscribe();// 立即启动 preparestage 实时流入 sinkreturnstageSink.asFlux().concatWith(prepared.flatMapMany(this::pipeline))prepare()每走到一步就emitStage(stages, 检索法律知识库…)前端立刻看到进度条滚动。推理模型的reasoning_content同理以REASONING_MARKER透传前端渲染成可折叠的思考面板并算出已思考 N 秒发送 → 首 token 的耗时。这里顺带用到了阻塞隔离subscribeOn(boundedElastic)cache()让 Tomcat 的服务器线程不被首 token 前的同步 IO 独占——这个主题在阶段五·篇3 会专门展开。六、踩坑备忘① GET 把 token 和问题拼进 URL落进 nginx 访问日志。早期/chat/stream?userMessage...法律咨询原文和 JWT 全进了代理日志。改成POST 请求体后日志里只剩路径——这是 P0 级安全整改。②filter不会取消上游takeUntilOther会。想中断流式用filter只是丢弃元素模型那边还在跑必须用takeUntilOther/take这类会向源传播 cancel 的操作符。③prepare被订阅两次 检索做两遍。stageSink 和 pipeline 分别订阅prepared时若不cache()Mono.fromCallable每次订阅都重跑。加.cache() 提前subscribe()启动既让 stage 实时流出又保证 prepare 只执行一次。④ SSE data 前导空格被协议吞。SSE 规范会剥掉data:后的一个前导空格token 恰好以空格开头时就丢了。解决对 token data 做 JSON 编码sseData()前端JSON.parse还原。七、小结概念一句话SSE单向推送逐 token比 WebSocket 轻哨兵标记一条 Flux 混装 8 类事件map 阶段分流真停止Redis 信号 takeUntilOther cancel 传播到模型思考可见化unicast sink 实时发 stagereasoning 透传八、下篇预告流式跑通了但一个多轮助手如果重启就失忆、聊久了上下文爆 token体验照样崩。下一篇我们做会话记忆的持久化与两级压缩并治理前端的流式滚动体验。源码与体验Gitee国内快https://gitee.com/spaserby/laoxiaosi.git GitHub https://github.com/spaserby/laoxiaosi.git 在线演示https://laoxiaosi.noctisblue.com本系列全套代码皆开源觉得这篇有帮助欢迎顺手点颗 ⭐