1. 从“打字机效果”说起：为什么流式输出是 AI 应用的刚需
做过 AI 对话类产品的朋友应该都有体会，用户对“等待感”的容忍度极低。你后端调一次大模型接口，哪怕只用了三秒，如果前端一直转圈圈什么都不显示，用户就会怀疑是不是卡死了。而一旦把结果一个字一个字往外“吐”，哪怕总耗时还是三秒，用户的体感也会好很多——这就是所谓的打字机效果。
打字机效果背后依赖的核心技术就是SSE（Server-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 而不是 WebSocket
WebSocket 是双向的，功能更强，但大模型对话绝大多数时候是“客户端发一次请求，服务端持续返回”，并不需要服务端主动向客户端发起通信。用 WebSocket 属于杀鸡用牛刀，还要额外维护心跳、重连、连接状态。
SSE 的优势在于：
- 实现简单：服务端就是往一个 HTTP 响应流里写数据，前端用
EventSource几行代码就能接 - 自动重连：
EventSource内置断线重连机制，配合id字段还能续传 - 走标准 HTTP：天然兼容各种网关、负载均衡、鉴权中间件，不需要额外开端口
- 文本友好：大模型输出本来就是文本，SSE 直接传文本，不需要额外编码
当然 SSE 也有短板：它是单向的，客户端不能通过同一条连接回传数据；另外浏览器对同一域名的 SSE 连接数有限制（HTTP/1.1 下通常是 6 个）。不过对于对话场景，这些限制基本不影响。
2.3 一次完整的 SSE 请求生命周期
我把一次 SSE 请求拆成几个阶段，方便你排查问题时定位：
- 建立连接：客户端发起 GET 请求，带上
Accept: text/event-stream - 服务端响应头：返回
200，Content-Type: text/event-stream，Cache-Control: no-cache，Connection: keep-alive - 持续推送：服务端每产生一段数据就写入响应流，格式遵循 SSE 规范
- 心跳保活：长时间没数据时，服务端定期发送注释行（以
:开头）防止连接被中间层断开 - 结束：服务端发送结束标志（如
data: [DONE]）后关闭连接，或客户端主动断开
这里第 4 步的心跳特别关键。很多网关（比如 Nginx、云厂商的负载均衡）默认 60 秒没有数据传输就会断开连接，如果你的模型思考时间较长，中间没有输出，连接就会被掐断，前端就会看到stream disconnected before completion或者idle timeout waiting for sse这类报错。
3. LangChain 结构化输出：让模型稳定吐出 JSON
3.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 定义一个 schema：
from 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(model="gpt-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(r"```json\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 f"data: {chunk}\n\n" yield "data: [DONE]\n\n" @app.get("/stream") async def stream(prompt: str): return StreamingResponse( event_generator(prompt), media_type="text/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 请求，且不能自定义请求头，很多场景不够用。所以我更推荐用fetch+ReadableStream手动读取：
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: true |
| JSON 解析失败 | 模型输出脏数据 | 打印原始文本 | 清洗 + 降级重试 |
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, timeout=15) yield f"data: {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 对话类产品，强烈建议把流式输出和结构化输出都吃透，这两块基本是绕不开的基本功。
