1. 从“打字机效果”说起为什么流式输出是 AI 应用的刚需做过 AI 对话类产品的朋友应该都有体会用户对“等待感”的容忍度极低。你后端调一次大模型接口哪怕只用了三秒如果前端一直转圈圈什么都不显示用户就会怀疑是不是卡死了。而一旦把结果一个字一个字往外“吐”哪怕总耗时还是三秒用户的体感也会好很多——这就是所谓的打字机效果。打字机效果背后依赖的核心技术就是SSEServer-Sent Events。它本质上是一种基于 HTTP 的单向流式推送协议服务端可以持续往客户端推送文本片段客户端通过EventSource或者fetch的流式读取来接收。相比 WebSocket 的双向通信SSE 更轻、更简单天然适合“服务端持续输出、客户端只负责接收”的场景比如大模型的 token 流式返回。但光有流式还不够。真实项目里我们往往还需要模型输出结构化的 JSON比如让它返回一个包含title、summary、tags的对象前端拿到后直接渲染成卡片。这时候问题就来了流式返回的是一段一段的文本碎片JSON 还没拼完你怎么解析如果等全部拼完再解析那打字机效果就没了如果边流边解析又容易在 JSON 不完整时抛异常。这篇文章我就把这条链路完整走一遍从 SSE 的底层原理到 LangChain 的结构化输出再到流式场景下 JSON 的增量解析最后给出前端打字机效果的落地写法。中间会穿插我自己踩过的坑比如idle timeout导致的流中断、stream disconnected before completion这类报错怎么排查。适合正在做 AI 应用、需要把流式输出和结构化数据结合起来的同学。2. SSE 流式原理拆解它到底是怎么把数据“推”过来的2.1 SSE 的协议格式与工作方式SSE 的全称是 Server-Sent Events它复用的还是普通 HTTP 连接只不过响应头里会带上Content-Type: text/event-stream并且服务端不会一次性把响应体写完而是保持连接打开分多次写入数据。客户端收到一段就处理一段直到服务端主动关闭或者连接超时。SSE 的数据格式其实非常朴素就是纯文本每条消息由若干字段组成字段之间用换行分隔消息之间用空行分隔。常见的字段有这几个data消息内容可以有多行多行会被拼接event自定义事件类型客户端可以按类型监听id消息 ID用于断线重连时定位retry重连等待时间单位毫秒一个典型的 SSE 响应长这样data: {token: 你} data: {token: 好} data: {token: } data: [DONE]注意每条data后面跟一个空行这是消息的分隔符。客户端解析时就是按空行切分再把同一消息内的多行data用换行拼起来。提示很多人第一次写 SSE 服务端时忘了在每条消息后加空行导致客户端一直收不到完整消息卡在缓冲区里。这个坑非常常见。2.2 为什么大模型场景偏爱 SSE 而不是 WebSocketWebSocket 是双向的功能更强但大模型对话绝大多数时候是“客户端发一次请求服务端持续返回”并不需要服务端主动向客户端发起通信。用 WebSocket 属于杀鸡用牛刀还要额外维护心跳、重连、连接状态。SSE 的优势在于实现简单服务端就是往一个 HTTP 响应流里写数据前端用EventSource几行代码就能接自动重连EventSource内置断线重连机制配合id字段还能续传走标准 HTTP天然兼容各种网关、负载均衡、鉴权中间件不需要额外开端口文本友好大模型输出本来就是文本SSE 直接传文本不需要额外编码当然 SSE 也有短板它是单向的客户端不能通过同一条连接回传数据另外浏览器对同一域名的 SSE 连接数有限制HTTP/1.1 下通常是 6 个。不过对于对话场景这些限制基本不影响。2.3 一次完整的 SSE 请求生命周期我把一次 SSE 请求拆成几个阶段方便你排查问题时定位建立连接客户端发起 GET 请求带上Accept: text/event-stream服务端响应头返回200Content-Type: text/event-streamCache-Control: no-cacheConnection: keep-alive持续推送服务端每产生一段数据就写入响应流格式遵循 SSE 规范心跳保活长时间没数据时服务端定期发送注释行以:开头防止连接被中间层断开结束服务端发送结束标志如data: [DONE]后关闭连接或客户端主动断开这里第 4 步的心跳特别关键。很多网关比如 Nginx、云厂商的负载均衡默认 60 秒没有数据传输就会断开连接如果你的模型思考时间较长中间没有输出连接就会被掐断前端就会看到stream disconnected before completion或者idle timeout waiting for sse这类报错。3. LangChain 结构化输出让模型稳定吐出 JSON3.1 为什么需要结构化输出大模型默认输出的是自然语言你问它“帮我总结这篇文章”它可能回你一段话也可能回你一个列表格式完全不固定。但真实业务里前端往往需要确定的数据结构比如{ title: 文章标题, summary: 一句话摘要, tags: [标签1, 标签2], sentiment: positive }如果每次都要写正则去抠维护成本极高模型稍微换个措辞就崩了。LangChain 的结构化输出就是来解决这个问题的——它通过约束模型的输出格式让模型直接返回符合 schema 的 JSON。3.2 LangChain 里几种结构化输出的实现方式LangChain 提供了多种让模型输出结构化数据的手段我按可靠性和适用场景排个序方式原理可靠性适用场景with_structured_output利用模型原生 function calling / JSON mode高支持工具调用的模型PydanticOutputParser在 prompt 里注入格式说明解析返回文本中不支持原生结构化输出的模型JsonOutputParser类似上面但用 JSON schema 描述中简单 JSON 结构手动 prompt 约束纯靠提示词要求返回 JSON低兜底方案最推荐的是with_structured_output因为它直接调用模型的原生能力比如 OpenAI 的 function calling、Claude 的 tool use模型在生成时就被约束了格式几乎不会跑偏。用 Pydantic 定义一个 schemafrom pydantic import BaseModel, Field from typing import List class ArticleSummary(BaseModel): title: str Field(description文章标题) summary: str Field(description一句话摘要) tags: List[str] Field(description标签列表) sentiment: str Field(description情感倾向positive/neutral/negative)然后绑定到模型上from langchain_openai import ChatOpenAI llm ChatOpenAI(modelgpt-4o-mini) structured_llm llm.with_structured_output(ArticleSummary) result structured_llm.invoke(帮我总结这段文字...) print(result.title, result.tags)返回的result直接就是ArticleSummary对象不用自己解析 JSON非常省心。3.3 结构化输出与流式的天然矛盾问题来了with_structured_output默认是非流式的。它要等模型把整个 JSON 生成完才能校验并转成 Pydantic 对象。这就意味着你拿不到打字机效果——用户要等好几秒然后 JSON 一次性蹦出来。这就是本文的核心矛盾结构化输出要求完整流式输出要求碎片。怎么调和我的思路是流式拿到的是 JSON 文本碎片自己维护一个缓冲区边收边尝试增量解析。LangChain 其实也提供了stream模式下的结构化输出支持但底层依然是返回文本 chunk需要我们自己处理。4. 流式 JSON 增量解析边收边解析的实战方案4.1 增量解析的核心难点假设模型流式返回这样一段 JSON{title: SSE实战, tags: [流式, 解析], summary: 一篇讲SSE的文章}它可能被切成这样的 chunk{title: SS E实战, tags: [流式, 解析], summary: 一篇讲SSE的文章}你拿到第一个 chunk 时JSON 是不完整的直接json.loads必然报错。所以增量解析要解决两件事判断当前缓冲区是否已经是合法 JSON合法就解析不合法就继续等处理不完整字符串比如SS这种引号还没闭合不能当成完整值4.2 用 json 库的异常做“试探性解析”最简单粗暴的办法是每次收到 chunk 就拼到缓冲区然后尝试json.loads成功就返回失败就继续等。这个思路对大多数场景够用因为 JSON 一旦完整json.loads就能成功。import json class IncrementalJsonParser: def __init__(self): self.buffer self.parsed None def feed(self, chunk: str): self.buffer chunk try: self.parsed json.loads(self.buffer) return self.parsed except json.JSONDecodeError: return None这个方案的问题在于如果 JSON 中间某段恰好是合法 JSON比如数组还没闭合但前面部分合法可能会误判。不过对于对象类型的输出只要最外层大括号没闭合json.loads就会失败所以基本安全。4.3 更稳的方案用 ijson 或 partial-json-parser如果你需要更精细的增量解析比如想在 JSON 还没闭合时就能读到已经完整的字段可以用partial-json-parser这类库。它能解析“部分合法”的 JSON把已经完整的部分返回出来。from partial_json_parser import loads as partial_loads def feed(self, chunk: str): self.buffer chunk try: return partial_loads(self.buffer) except Exception: return None这样即使 JSON 还没闭合你也能拿到{title: SSE实战}这样的部分结果前端可以先把 title 渲染出来tags 等后续 chunk 到了再补。注意增量解析一定要做异常兜底。模型偶尔会输出非法 JSON比如多一个逗号、少一个引号这时候不能直接崩要有降级策略比如记录原始文本、返回错误提示、或者触发一次非流式的重试。4.4 处理模型输出的“脏数据”实际项目里模型返回的流式文本经常带一些“包装”比如前面有json代码块标记后面有结束标记中间夹杂解释性文字这些都会导致 JSON 解析失败。我的处理方式是在喂给解析器之前先做清洗import re def clean_chunk(text: str) - str: text re.sub(rjson\s*, , text) text re.sub(r, , text) return text但清洗要小心不能把 JSON 内部的合法字符也删了。更稳妥的做法是只在流开始时检测并剥离代码块标记流中间的内容原样保留。5. 前后端联调从 FastAPI 到 Vue 的完整链路5.1 后端FastAPI 实现 SSE 接口FastAPI 实现 SSE 非常方便用StreamingResponse配合生成器即可from fastapi import FastAPI from fastapi.responses import StreamingResponse import asyncio app FastAPI() async def event_generator(prompt: str): # 模拟调用 LangChain 流式接口 async for chunk in stream_llm(prompt): yield fdata: {chunk}\n\n yield data: [DONE]\n\n app.get(/stream) async def stream(prompt: str): return StreamingResponse( event_generator(prompt), media_typetext/event-stream, headers{ Cache-Control: no-cache, Connection: keep-alive, X-Accel-Buffering: no, }, )这里有几个关键点media_type必须是text/event-stream每条消息后必须加\n\n这是 SSE 的消息分隔符X-Accel-Buffering: no是给 Nginx 看的告诉它不要缓冲否则数据会被攒着一起发如果用了反向代理记得关闭代理层的缓冲5.2 后端接入 LangChain 流式输出LangChain 的模型对象支持astream方法可以异步逐 chunk 拿到输出async def stream_llm(prompt: str): async for chunk in structured_llm.astream(prompt): if chunk.content: yield chunk.content注意with_structured_output在流式模式下返回的 chunk 可能是 JSON 文本片段需要在前端或后端做增量解析。我一般选择在后端解析把解析好的部分对象推给前端前端只负责渲染逻辑更清晰。5.3 前端Vue 里用 fetch 读取流浏览器原生的EventSource只支持 GET 请求且不能自定义请求头很多场景不够用。所以我更推荐用fetchReadableStream手动读取async function streamChat(prompt) { const response await fetch(/stream?prompt encodeURIComponent(prompt), { headers: { Accept: text/event-stream } }); const reader response.body.getReader(); const decoder new TextDecoder(); let buffer ; while (true) { const { done, value } await reader.read(); if (done) break; buffer decoder.decode(value, { stream: true }); const lines buffer.split(\n\n); buffer lines.pop(); // 最后一段可能不完整留到下次 for (const line of lines) { if (line.startsWith(data: )) { const data line.slice(6); if (data [DONE]) return; handleChunk(data); } } } }这段代码有两个细节值得说decoder.decode(value, { stream: true })里的stream: true很重要它能正确处理跨 chunk 的多字节字符比如中文否则可能出现乱码buffer.split(\n\n)后要把最后一段pop出来留到下次因为一个 SSE 消息可能被 TCP 分包切成两半5.4 前端打字机效果的渲染拿到 chunk 后最简单的打字机效果就是直接往文本后面追加const text ref(); function handleChunk(chunk) { text.value chunk; }Vue 的响应式会自动触发重渲染看起来就是逐字出现。如果想要更平滑的“打字”节奏可以用一个队列 定时器把 chunk 拆成单字慢慢吐const queue []; let timer null; function handleChunk(chunk) { queue.push(...chunk.split()); if (!timer) startTyping(); } function startTyping() { timer setInterval(() { if (queue.length 0) { clearInterval(timer); timer null; return; } text.value queue.shift(); }, 30); }这样即使后端一次推来一大段前端也能均匀地“打”出来视觉上更舒服。6. 常见问题与排查技巧实录6.1 流中断类问题速查表报错信息可能原因排查方向解决方案stream disconnected before completion连接被中间层断开检查 Nginx/网关超时配置增加心跳、调大超时时间idle timeout waiting for sse长时间无数据模型思考时间长服务端定期发注释行保活前端收不到数据代理缓冲检查X-Accel-Buffering关闭代理缓冲中文乱码解码方式错误检查TextDecoder使用stream: trueJSON 解析失败模型输出脏数据打印原始文本清洗 降级重试6.2 心跳保活的正确写法服务端在等待模型输出时可以定期发送注释行async def event_generator(prompt): task asyncio.create_task(collect_chunks(prompt)) while not task.done(): try: chunk await asyncio.wait_for(task, timeout15) yield fdata: {chunk}\n\n except asyncio.TimeoutError: yield : keep-alive\n\n # 注释行客户端会忽略 yield data: [DONE]\n\n注释行以:开头客户端解析时会自动跳过但能保持 TCP 连接活跃防止被网关判定为空闲连接而断开。6.3 结构化输出解析失败的兜底策略模型不是每次都听话偶尔会输出非法 JSON。我的兜底策略分三层第一层增量解析失败时继续累积不立即报错第二层流结束后仍解析失败尝试用正则提取 JSON 片段第三层仍失败则触发一次非流式的结构化输出调用作为最终兜底def fallback_parse(text: str): match re.search(r\{.*\}, text, re.DOTALL) if match: try: return json.loads(match.group()) except json.JSONDecodeError: pass return None提示兜底重试会增加延迟和成本所以只在解析确实失败时触发不要每次都跑。6.4 我踩过的几个坑坑一Nginx 默认缓冲导致流式失效。一开始本地测试好好的部署到服务器后前端一直转圈最后一次性蹦出全部内容。排查半天发现是 Nginx 的proxy_buffering默认开启把流式响应攒起来了。解决办法是在 location 里加proxy_buffering off;或者后端响应头加X-Accel-Buffering: no。坑二EventSource不支持 POST。我一开始想用EventSource传复杂的请求体结果发现它只支持 GET参数只能塞 URL。后来改用fetch手动读流灵活多了。坑三中文被截断成乱码。一个中文字符在 UTF-8 里占 3 个字节如果 TCP 分包正好切在字符中间直接decode就会出乱码。加上{ stream: true }后TextDecoder会自己缓存不完整的字节等下一个 chunk 到了再拼问题解决。坑四with_structured_output和stream不能同时用。我一开始想直接structured_llm.stream()拿到 Pydantic 对象结果发现它返回的还是文本 chunk。后来才明白结构化输出本质是约束生成格式流式拿到的还是原始文本解析得自己做。7. 性能与体验优化让流式输出更丝滑7.1 减少首字节延迟用户感知的“快”很大程度上取决于首字节时间TTFB。如果模型要思考两秒才开始输出用户就会觉得卡。优化方向有几个用更快的模型做首轮响应复杂任务再切换到大模型在 prompt 里明确要求“直接输出不要解释”减少模型的“废话”前缀服务端在等待模型时先发一个空注释行让连接尽快建立7.2 前端渲染的性能考量如果 chunk 来得非常密集比如每秒几十个每次都触发 Vue 重渲染会有性能压力。我的做法是做一个小的节流把 chunk 先塞进队列用requestAnimationFrame批量更新保证每帧最多渲染一次。let pending ; let rafId null; function handleChunk(chunk) { pending chunk; if (!rafId) { rafId requestAnimationFrame(() { text.value pending; pending ; rafId null; }); } }这样既保证了流畅度又避免了频繁重渲染。7.3 断线重连与状态恢复SSE 本身支持断线重连EventSource会自动带上Last-Event-ID请求头。但用fetch手动读流时重连要自己实现。我的做法是记录已接收的内容长度重连时带上偏移量服务端从对应位置继续推送。不过这个方案需要服务端支持断点续传实现成本较高一般场景下直接重新发起请求、清空重来也能接受。8. 一些延伸思考与个人经验这套方案我在几个项目里都跑过整体稳定性不错。有几点体会分享给正在做类似功能的同学。第一流式和结构化输出不要强行统一。有些场景其实不需要流式比如后台批处理任务直接等完整 JSON 更省事。只有面向用户的实时交互场景才值得为打字机效果付出增量解析的复杂度。第二增量解析的粒度要控制好。解析太频繁会浪费 CPU解析太稀疏又失去了流式的意义。我的经验是每收到一个 chunk 就尝试一次因为json.loads对短文本的开销很小实测下来完全不是瓶颈。第三日志一定要打全。流式场景出问题时光看报错很难定位必须把每个 chunk 的原始内容、时间戳、解析结果都记下来。我一般会在开发环境把原始流写到一个文件里出问题直接回放。第四给用户明确的反馈。流式输出过程中如果模型卡住了前端要有个“正在思考”的提示而不是让用户干等。可以在超过一定时间没收到新 chunk 时显示一个加载动画。这套链路涉及的东西不少从协议层到应用层都有坑但一旦跑通用户体验的提升是肉眼可见的。如果你正在做 AI 对话类产品强烈建议把流式输出和结构化输出都吃透这两块基本是绕不开的基本功。