题解生成的流式输出优化：首字延迟与生成速度的博弈

题解生成的流式输出优化首字延迟与生成速度的博弈一、用户盯着空白页等 10 秒和看着文字逐字冒出来等 15 秒哪个体验更好直觉上10 秒比 15 秒短应该是 10 秒的方案更好。但实际用户感知恰恰相反看空白页等 10 秒焦虑感拉满看文字一个字一个字跳出来等 15 秒感知等待时间反而更短。这是因为流式输出提供了进度感——用户知道系统在干活而且能看到干活的结果正在逐步出现。这种心理效应在用户体验中叫做进度反馈降低等待焦虑。对于 AI 题解生成来说流式输出还有额外的意义。一道题目的分析通常有固定的结构先判断代码是否正确再分析复杂度最后给出优化建议。如果采用非流式一次性返回完整结果用户必须等到所有分析都完成才能看到任何内容。而采用流式输出用户在模型生成第一部分分析的同时就能开始阅读人的阅读速度和模型的生成速度叠加在一起整体感知耗时大幅缩短。sequenceDiagram participant User as 用户 participant Server as 服务端 participant Model as 模型推理 User-Server: 提交判题请求 Server-Model: 发起流式推理请求 (streamtrue) Note over Model: 推理开始 loop Token 逐个生成 Model--Server: 返回单个 Token Server--User: SSE 推送 Token Note over User: 用户看到逐字出现 end Model--Server: 流结束信号 [DONE] Server--User: SSE 关闭连接 Note over User,Model: 用户在第一个 Token 到达时就开始阅读二、TTFB 与生成速度的权衡流式输出有两个核心指标TTFBTime To First Byte首字延迟从请求发起到第一个 token 返回的时间。这个时间越短用户感知的响应越快。TTFB 主要取决于 prompt 的处理时间和模型调度延迟。生成速度后续 token 的产出速率tokens/s。这个速率受模型大小、硬件性能和批处理策略的影响。两者之间存在天然的矛盾。如果追求低 TTFB应该尽量减少 prompt 的长度和复杂度让模型尽快开始生成。但 prompt 太简略可能无法提供足够的上下文导致模型生成的质量下降甚至出现后续的修正和重新生成——整体延迟反而更长。如果追求高生成速度应该使用更大的 batch size 和更激进的并发策略但这可能增加队列等待时间反而让 TTFB 上升。在题解生成场景中一个实用的折中方案是首屏优先把 prompt 拆成核心信息和补充信息两部分。核心信息题目描述、用户代码先发给模型让模型尽快启动推理。补充信息历史判题数据、用户能力画像在模型启动后异步追加。这样在补充上下文的同时不拖慢首字的生成。三、服务端 SSE 流式推送实现服务端通过 SSEServer-Sent Events协议将模型的 token 流逐字推送给前端。SSE 比 WebSocket 更轻量只需要服务端到前端的单向推送不需要全双工通道。/** * 流式题解生成控制器 * * 设计要点 * 1. 使用 SSE (Server-Sent Events) 协议推送 token 流 * 2. 设置合理的超时时间防止连接长时间占用 * 3. 异常时发送错误事件并关闭连接 * 4. 定时发送心跳事件防止代理服务器关闭空闲连接 */ RestController public class StreamingJudgeController { private final ModelStreamClient modelClient; /** * 流式判题接口 * * produces MediaType.TEXT_EVENT_STREAM_VALUE 表示使用 SSE 协议 */ GetMapping(value /judge/stream, produces MediaType.TEXT_EVENT_STREAM_VALUE) public SseEmitter judgeStream( RequestParam String problemId, RequestParam String code) { // 超时时间设置为 60 秒比单次推理的超时稍长 // 使用 SseEmitter 而不是直接写 OutputStream简化连接管理 SseEmitter emitter new SseEmitter(60_000L); // 在独立的线程中处理推理过程避免阻塞主线程 // 主线程需要立刻返回 SseEmitter 对象给 Spring 框架管理 executor.execute(() - { try { // 第一步发送连接建立事件 emitter.send(SseEmitter.event() .name(start) .data(判题开始...)); // 第二步构建 prompt String prompt buildPrompt(problemId, code); // 第三步流式调用模型并转发 token modelClient.streamJudge(prompt, new StreamCallback() { Override public void onToken(String token) { try { // 每个 token 作为一个 SSE data 事件推送 // 使用 token 事件名方便前端区分不同类型的事件 emitter.send(SseEmitter.event() .name(token) .data(token)); } catch (IOException e) { // 连接可能已断开用户关闭了页面 // 此时应中断模型推理释放资源 throw new StreamInterruptedException(e); } } Override public void onComplete() { try { // 发送结束事件附带元数据 emitter.send(SseEmitter.event() .name(done) .data({\status\: \complete\})); emitter.complete(); } catch (IOException e) { emitter.completeWithError(e); } } Override public void onError(Throwable t) { try { // 错误事件携带错误信息前端据此展示异常提示 emitter.send(SseEmitter.event() .name(error) .data(判题失败: t.getMessage())); emitter.complete(); } catch (IOException e) { emitter.completeWithError(e); } } }); } catch (Exception e) { emitter.completeWithError(e); } }); return emitter; } /** * 心跳定时器防止长时间无输出时连接被代理断开 * * 有些反向代理如 Nginx会关闭超过一定时间无数据的连接 * 在模型预热或长时间思考时心跳能维持连接活性 */ private void startHeartbeat(SseEmitter emitter) { ScheduledExecutorService heartbeat Executors .newSingleThreadScheduledExecutor(); heartbeat.scheduleAtFixedRate(() - { try { emitter.send(SseEmitter.event() .name(ping) .data()); } catch (IOException e) { // 连接已关闭停止心跳 heartbeat.shutdown(); } }, 0, 15, TimeUnit.SECONDS); } private String buildPrompt(String problemId, String code) { /* */ return ; } private ExecutorService executor; }心跳定时器在 SSE 实现中容易被遗漏但当 Nginx 的反向代理有proxy_read_timeout配置时长时间无输出的连接会被断开。SSE 连接的占用量也是一个需要考虑的因素——每个 SSE 连接占用一个 HTTP 长连接和一个服务端线程资源并发量较大时需要使用异步非阻塞模型。四、流式输出的后处理挑战流式输出虽然提升了体验但在后处理环节引入了复杂性。格式解析非流式输出拿到的是一个完整的 JSON可以直接解析。流式输出的 token 是分散到达的必须在客户端做拼接和增量解析。如果模型输出的结构是 Markdown JSON 混排比如分析部分是 Markdown结论部分是 JSON增量解析的难度会上升。打断与继续用户可能在生成中途点了停止按钮中断了当前的流式输出。这时候服务端不仅要停止推送还应该请求模型服务取消推理释放 GPU 资源。模型取消不是所有框架都支持需要确认底层推理引擎的取消能力。 前端流式接收的增量解析示例 设计要点 - 维护一个令牌缓冲区累积足够的内容后尝试解析 - 对 Markdown 内容直接追加到显示区 - 对 JSON 代码块等代码块闭合后一次性解析 class StreamParser: def __init__(self): self.buffer # 累积所有收到的 token self.display_text # 已显示给用户的文本 self.in_code_block False # 是否正在 JSON 代码块内 self.code_buffer # JSON 代码块内容缓冲区 def feed(self, token: str): 接收一个 token 并更新显示 self.buffer token if json in self.buffer[len(self.buffer)-10:]: # 检测到 JSON 代码块开始 self.in_code_block True self.code_buffer return {type: text, content: token} if self.in_code_block: if in token: # JSON 代码块结束尝试解析 self.in_code_block False try: parsed json.loads(self.code_buffer) return {type: json, content: parsed} except json.JSONDecodeError: return {type: text, content: token} else: self.code_buffer token return {type: code, content: token} return {type: text, content: token}五、总结流式输出不是简单地把一次性返回改成分批推送。它的核心价值是用进度反馈缓解等待焦虑用 TTFB 和生成速度的协同优化提升整体感知体验。实现上需要处理 SSE 连接管理、心跳保活、前端增量解析、中断取消等一系列工程细节。但对于任何需要长时间 AI 推理的前端交互场景流式输出都是必须支持的基础能力。不做流式再快的模型也会被用户感知成慢的。