18-AI聊天流式响应与取消
AI 聊天怎样真正流起来？SSE、取消与半完成响应
模型正在生成，页面却等到最后一次性显示；用户已经停止等待，后端仍继续计费。流式聊天需要同时处理传输、渲染和任务生命周期，不能把它简化为把 String 改成 Flux。
Spring AI 接入片段使用 Java 17、Spring Boot 3.5.3 与 Spring AI 1.1.8；协议与第三方平台示例另行标明适用范围。
1. 先定义增量与终态
网络关闭可能是完成，也可能是失败；事件必须让两者可区分。
流式与普通返回有什么区别？
普通接口等待完整结果后返回；流式接口在生成过程中连续发送片段，降低用户等待第一段内容的时间，但不一定缩短整个任务耗时。
SSE 使用 text/event-stream，事件以约定格式分隔。一个事件不一定等于一个 Token，一次网络读取也不一定对应一个完整事件。前端应按事件协议解析，不能把 TCP 数据块当作业务边界。
Spring AI 的 stream().content() 提供文本流，stream().chatResponse() 可获取更完整响应信息。Spring AI 1.1 流式接口
给流设计明确的终止协议
可以区分 delta、done 和 error：delta 是增量内容；done 表示本次生成正常结束；error 表示失败，不应被当成正常回答的一部分。
event: delta
data: {"text":"示意片段"}
event: done
data: {"text":""}
这是协议示意，不是模型实测输出。若已经发送响应头，再发生异常，通常不能重新把 HTTP 状态改成普通 JSON 错误，因此要设计流内错误事件。
2. Spring 后端怎样输出并停止
完整接入示例限定自己的事件契约，客户端按同一协议解析。
一个 Spring WebFlux 接入示例
沿用Java 服务变慢怎样定位？从 CPU、GC 到内存保留链的 Spring AI 1.1.8 与 Boot 配置，将 Web starter 替换为 spring-boot-starter-webflux，不要同时依赖 MVC 与 WebFlux 后再假定一定使用响应式服务器。保留校验依赖与模型 starter。
以下类放在 demo 包中，保留Java 服务变慢怎样定位？从 CPU、GC 到内存保留链 Main 的启动入口，移除原同步 ChatController，再添加这个流式 Controller。
package demo;
import java.time.Duration;
import jakarta.validation.Valid;
import jakarta.validation.constraints.NotBlank;
import jakarta.validation.constraints.Size;
import org.springframework.ai.chat.client.ChatClient;
import org.springframework.http.MediaType;
import org.springframework.http.codec.ServerSentEvent;
import org.springframework.web.bind.annotation.PostMapping;
import org.springframework.web.bind.annotation.RequestBody;
import org.springframework.web.bind.annotation.RestController;
import reactor.core.publisher.Flux;
import reactor.core.publisher.Mono;
@RestController
public class StreamingController {
public record Input(@NotBlank @Size(max = 4000) String message) {}
public record Payload(String text) {}
private final ChatClient client;
public StreamingController(ChatClient.Builder builder) {
client = builder.build();
}
private ServerSentEvent<Payload> event(String type, String text) {
return ServerSentEvent.<Payload>builder(new Payload(text)).event(type).build();
}
@PostMapping(value = "/chat/stream", produces = MediaType.TEXT_EVENT_STREAM_VALUE)
public Flux<ServerSentEvent<Payload>> stream(@Valid @RequestBody Input input) {
return Flux.defer(() -> client.prompt().user(input.message()).stream().content())
.timeout(Duration.ofSeconds(15))
.map(text -> event("delta", text))
.concatWith(Mono.fromSupplier(() -> event("done", "")))
.onErrorResume(error -> Mono.just(event("error", "生成失败，请按请求记录确认状态")))
.doFinally(signal -> System.out.println("stream terminated: " + signal));
}
}
这个 timeout 控制首次及相邻片段的等待，不是整个生成最多 15 秒。持续有片段到达时，总耗时可以超过它；总时限需要另外设置，并使达到时限明确进入失败或取消状态。
示例把异常转成通用错误事件，生产应记录脱敏后的内部原因和请求 ID。输入字段长度限制也不等于完整 Token 预算。
为什么客户端需要正确取消？
浏览器使用支持流读取的 fetch，并在停止生成时通过 AbortController 中止请求。标准 EventSource 主要用于 GET，不适合直接替代这里的 POST JSON 请求；自动重连也可能再次触发生成，需明确请求幂等与恢复语义。
服务端连接断开后，应让取消信号传到上游。不要在 Controller 内部手动 subscribe，再把另一个 Flux 返回；那样可能出现两个订阅，或者返回链取消了，手动启动的任务却仍在执行。
返回 Flux 也不代表链中所有操作都非阻塞。Spring AI 1.1 的工具调用可能执行同步工作，普通 Redis 仓库也可能阻塞；应核对它们在哪个 Scheduler 执行，使用受控隔离并限制容量，避免阻塞响应式处理线程。
取消客户端订阅不保证提供方立即停止计算或停止计费。外部服务是否支持取消，以及网络中已经发生的工作，都要分别确认。已调用的业务工具更不会因为页面关闭而自动回滚。
半个回答应该怎样保存？
为生成记录保存 RUNNING、COMPLETED、FAILED 或 CANCELLED 等状态，并区分已经显示的部分内容与完整答案。失败时不能把半截 JSON 当成结构化成功结果。
断线重连是继续读取同一任务还是重新生成，要有明确协议。重新执行工具可能产生副作用，不能默认“重新打开连接”无成本。
3. 浏览器怎样避免旧回答串到新会话
网络块、请求代数、历史状态与滚动分别处理。
网络块不等于事件
一次读取可能只有半个中文字符，也可能包含多个事件。需要 TextDecoder 保持跨块 UTF-8 状态，再按空行识别 SSE 帧。Fetch 标准、SSE 标准。
下面的 readChatStream.js 针对约定的 UTF-8、LF 或 CRLF 格式。它不实现通用 SSE 重连、retry 与 id 机制，遇到不完整结束会报错：
export async function readChatStream(response, onEvent) {
if (!response.ok || !response.body) throw new Error('无法建立回答连接')
const reader = response.body.getReader()
const decoder = new TextDecoder()
let buffer = ''
let terminal = false
try {
while (!terminal) {
const { value, done } = await reader.read()
buffer += done ? decoder.decode() : decoder.decode(value, { stream: true })
if (buffer.length > 1024 * 1024) throw new Error('单帧内容过大')
let boundary
while ((boundary = /\r?\n\r?\n/.exec(buffer)) !== null) {
const frame = buffer.slice(0, boundary.index)
buffer = buffer.slice(boundary.index + boundary[0].length)
let type = 'message'
const data = []
for (const line of frame.split(/\r?\n/)) {
if (line.startsWith(':')) continue
const colon = line.indexOf(':')
const field = colon < 0 ? line : line.slice(0, colon)
let content = colon < 0 ? '' : line.slice(colon + 1)
if (content.startsWith(' ')) content = content.slice(1)
if (field === 'event') type = content
if (field === 'data') data.push(content)
}
if (!data.length) continue
const payload = JSON.parse(data.join('\n'))
onEvent(type, payload)
if (type === 'done' || type === 'error') {
terminal = true
break
}
}
if (done) break
}
if (!terminal) throw new Error('回答连接中断，内容未完成')
} finally {
try { await reader.cancel() } finally { reader.releaseLock() }
}
}
实际应用再限制回答总长度、整体时长和空闲时长。单帧限制不能代替整个回复的上限。
取消与历史记录怎样配合？
AbortController 停止浏览器等待，不证明模型或工具已经停止。后端需要感知断线，或提供受授权的停止接口，并核查任务状态。
历史消息由后端持久化，保存完整、失败或取消状态。页面刷新后不能把一段残留文本展示成成功回答。会话列表、消息查询和停止接口都要检查当前用户归属。
同一会话是否允许同时发送两条问题，需要明确策略。顺序会话通常串行生成，否则模型上下文和显示顺序容易不一致。
渲染速度与滚动也需要控制
每个很小片段都立即重算 Markdown，可能增加开销。可以将片段短暂聚合，在动画帧或固定节奏更新视图，同时限制总长度。
用户正在向上阅读时，不要每个片段都强制滚动到底部。只有原本接近底部时自动跟随，并提供回到底部按钮。
静态检查不能证明所有浏览器流行为正确。上线前按实际目标浏览器验证分块、取消、代理超时和会话切换，本文没有运行这些测试。
4. 代理和多端还会改变传输行为
检查每一层的缓冲与超时，不假定手机端拥有全部浏览器能力。
代理缓冲为什么会让流看起来失效？
若 Nginx 将上游内容缓存后再发，浏览器就可能等到很晚才看到内容。下面是单独 SSE 路径的配置片段，需要放入已有 server 配置并核对实际超时预算：
location /chat/stream {
proxy_pass http://127.0.0.1:8080;
proxy_http_version 1.1;
proxy_buffering off;
proxy_cache off;
proxy_read_timeout 75s;
}
proxy_read_timeout 不是整个请求的固定截止时间。还要观察网关、负载均衡、压缩、浏览器读取与展示层。服务端有片段日志，不代表用户已经收到。
心跳事件可以帮助维持连接和发现失联，但心跳不应掩盖模型长时间没有输出的事实。模型空闲超时与 HTTP 连接心跳应分开管理。
一个 API 与流式接口配置
以下内容放在已有 Nginx http 块内。示例监听本地测试端口，不包含正式 HTTPS 证书配置；后端实际路径保留 /api：
upstream assistant_backend {
server 127.0.0.1:8080;
}
server {
listen 8088;
server_name localhost;
client_max_body_size 10m;
location /api/chat/stream {
proxy_pass http://assistant_backend;
proxy_http_version 1.1;
proxy_set_header Host $host;
proxy_set_header X-Forwarded-For $proxy_add_x_forwarded_for;
proxy_set_header X-Forwarded-Proto $scheme;
proxy_set_header Connection "";
proxy_buffering off;
proxy_cache off;
proxy_connect_timeout 5s;
proxy_read_timeout 75s;
}
location /api/ {
proxy_pass http://assistant_backend;
proxy_set_header Host $host;
proxy_set_header X-Forwarded-For $proxy_add_x_forwarded_for;
proxy_set_header X-Forwarded-Proto $scheme;
proxy_connect_timeout 5s;
proxy_read_timeout 30s;
}
}
实际部署在容器中时，上游地址改成应用服务名；127.0.0.1 只适用于 Nginx 与应用处在对应的同机网络环境。
流式响应关闭代理缓冲后，仍要检查应用服务器、压缩和外层 CDN 是否缓存。只改 Nginx 一处，无法证明整条链路都及时转发。
超时为什么不是一个数字？
连接超时负责建立上游连接，读取超时控制两次读取之间允许的等待，并不是整个请求的总时长。proxy_read_timeout。
SSE 心跳可以避免长时间没有数据，但不能替代整个任务截止时间。应用需要独立限制最大执行时长、工具等待和模型费用。
超时从浏览器、网关、Nginx 到后端分别存在。内层超时应形成可解释结果，外层略留余量；否则用户只看到代理截断，后端仍在工作。
同一份代码有哪些不同运行条件？
H5 使用浏览器环境，受到同源策略和 CORS 约束。小程序受到平台的合法域名、HTTPS 等规则约束。App 还受设备网络、证书与系统权限影响。
开发工具中的“忽略域名校验”不能代表发布后的行为。真机上的 localhost 指的是设备自己，不是开发电脑。应提供设备可达且符合平台要求的测试地址。
明确构建目标、平台版本、基础库与 HBuilderX 版本，把问题记录为可复现条件，而不是只说“手机访问不了”。uni.request 文档。
流式响应和组件兼容不能想当然
浏览器 fetch 的 ReadableStream 示例不能直接保证在所有 uni-app 目标端工作。平台是否支持分块、相关回调与取消，需要查目标端能力表并真机验证。
对于 AI 聊天，可以选择目标端支持的分块请求，或通过后端 WebSocket 协议转发，但仍需认证、事件顺序、终态和断线恢复。不要为了统一接口名称而省略这些约束。
维护一张目标端验证表：登录、请求、上传、下载、取消、长连接和组件展示。明确“已支持”“需兼容实现”“尚未验证”，有助于避免上线前才发现差异。
5. 设计判断与适用边界
前后端首先约定唯一事件协议。本文不同示例分别展示原理或独立接入方式，不能把 GET 后端和 POST 客户端当成天然配套。真实项目统一路径、方法、字段和终态，再做集成检查。
取消先阻止继续提交工作，再向下游传播；已经完成的工具副作用另行核查。历史记录保存完整、取消或失败状态，不让半段文本伪装成完整答案。渲染使用安全文本或清理后的 Markdown，避免执行模型生成的 HTML。
6. 问答
问：返回 Flux 就一定能逐段显示吗？
答：还取决于订阅、服务端写出、代理缓冲和客户端解析，必须看整条链路。
问：timeout 15 秒就是总生成时间吗？
答：示例的响应式 timeout 主要限制相邻信号间隔，总时限需单独设计。
问：为什么需要 generation 和 abort 两种机制？
答：取消减少浪费，代数检查阻止迟到回调更新新任务，二者处理不同问题。
问：连接关闭就算回答完成吗？
答：应取得约定的完成事件；否则可能是中断。
问：为什么后端流式输出，前端一次收到？
答：检查代理、应用压缩和外层缓存，缓冲可能聚合数据。
问：把 read_timeout 改得很大就能解决问题吗？
答：不能，可能只是掩盖慢请求，任务仍需要总预算。