1. 为什么流式输出不是"锦上添花"而是刚需
如果你做过大模型应用，一定遇到过这种场景：用户点下发送按钮，界面卡住十几秒，然后"啪"地一下蹦出一大段完整回答。用户在这十几秒里不知道程序是死是活，体验极差。而换成流式输出之后，文字像打字机一样一个字一个字往外冒，用户立刻就能感知到"它在工作"。这个差别不是视觉上的小修饰，而是产品可用性的分水岭。
流式输出的技术底座是SSE（Server-Sent Events）。它本质上是一种基于 HTTP 的单向推送协议，服务端可以持续往客户端发送文本片段，而不需要客户端反复轮询。相比 WebSocket 的双向通信，SSE 更轻、更简单，天然适合"服务端持续吐字、客户端只负责渲染"这种大模型对话场景。
但光有 SSE 还不够。大模型吐出来的是一段段自然语言文本，而你的业务代码往往需要的是结构化的 JSON 对象——比如从用户提问里提取出"姓名、电话、地址"，或者让模型返回一个可以直接入库的字段集合。这就引出了第二个核心问题：结构化输出。LangChain 提供了with_structured_output这类能力，让模型直接返回符合 Pydantic 模型定义的 JSON，但当你把流式和结构化输出放在一起用的时候，坑就来了——流式吐出来的 JSON 是残缺的、不完整的，你不能每收到一个 chunk 就去JSON.parse，那样必然报错。
这篇内容就是围绕这条链路展开的：从 SSE 的底层原理讲起，到 LangChain 结构化输出的实现方式，再到流式场景下 JSON 的增量解析方案，最后落到打字机效果的前端渲染。适合已经用过大模型 API、想进一步把流式体验做扎实的开发者，也适合刚接触 LangChain、想搞清楚"流式 + 结构化"到底怎么配合的入门者。我会把每一步的"为什么这么做"讲清楚，而不是只丢一段代码让你抄。
2. SSE 流式原理：数据到底是怎么一段段过来的
2.1 SSE 的报文格式与传输机制
很多人以为流式输出是什么高深的技术，其实拆开看非常简单。SSE 的响应体就是一段纯文本，服务端按照固定格式往里面写数据，客户端按行读取并解析。它的 Content-Type 是text/event-stream，每一条消息由若干字段组成，字段之间用换行分隔，消息之间用空行分隔。
一条典型的 SSE 消息长这样：
data: {"content": "你"} data: {"content": "好"} data: {"content": "，世界"}每个data:后面跟的就是一条消息的内容。注意消息之间必须有一个空行，这是 SSE 协议规定的分隔符。客户端读到空行就知道"上一条消息结束了"。除了data字段，还有event（自定义事件类型）、id（消息编号，用于断线重连）、retry（重连间隔）这几个可选字段。大模型流式输出场景里，最常用的就是data，偶尔用event来区分"正常内容"和"结束信号"。
为什么 SSE 能做到"持续推送"？关键在于 HTTP 的chunked transfer encoding（分块传输编码）。普通 HTTP 响应会带一个Content-Length，告诉客户端总共多少字节，收完就结束。而流式响应不带Content-Length，改用Transfer-Encoding: chunked，服务端可以一块一块地写，写完一块客户端就能收到一块，直到服务端主动关闭连接。这就是"打字机效果"的物理基础——不是前端在模拟逐字显示，而是数据真的是一段段到达的。
2.2 为什么大模型场景偏爱 SSE 而不是 WebSocket
这里有个常见的疑问：WebSocket 不是更强大吗，为什么大模型应用普遍用 SSE？我实际做过几个项目之后，总结下来有几个现实原因。
第一，大模型对话是典型的单向流。用户发一次请求，服务端持续返回内容，客户端在这个过程中基本不需要往服务端推数据。WebSocket 的双向能力在这里是浪费的，反而增加了连接管理的复杂度。
第二，SSE 天然走 HTTP，基础设施友好。你不需要额外开端口、不需要处理协议升级（Upgrade）、不需要担心某些网络环境对 WebSocket 的拦截。负载均衡、网关、日志系统对普通 HTTP 的支持都是现成的，SSE 直接复用这一套。
第三，SSE 自带断线重连语义。协议里的id和retry字段就是为断线重连设计的，浏览器端的EventSource会自动处理重连。虽然实际项目里我们经常自己用fetch+ReadableStream来读流（因为EventSource不支持 POST 请求和自定义 header），但协议层面的设计思路是值得借鉴的。
提示：如果你用
EventSource，它只能发 GET 请求，没法带请求体。而大模型对话通常需要 POST 传 prompt，所以生产环境更常见的是用fetch拿到response.body这个ReadableStream，然后自己按 SSE 格式解析。这一点后面会详细讲。
2.3 一个容易踩的坑：idle timeout 与流中断
热词里有个很典型的报错：stream disconnected before completion: idle timeout waiting for sse。这个错误的本质是：连接空闲时间超过了某一层的超时阈值，被强制断开了。
SSE 连接是长连接，如果模型思考时间较长（比如推理模型在"想"的时候可能几十秒不吐字），这段时间连接上没有任何数据流动，中间的任何一层——Nginx、负载均衡、API 网关、甚至客户端自己的超时设置——都可能判定这个连接"僵死"了，然后把它掐掉。表现就是前端收到一半内容突然断了，或者直接报上面那个错。
解决思路分几个层面。服务端层面，可以定期发送心跳注释。SSE 协议里以冒号开头的行是注释，客户端会忽略，但它的作用是保持连接活跃：
: keep-alive每隔 15 到 30 秒发一条这样的注释，就能让中间层知道"这个连接还活着"。Nginx 层面，需要关掉缓冲并调大超时：
location /api/stream { proxy_pass http://backend; proxy_buffering off; proxy_cache off; proxy_read_timeout 300s; proxy_set_header Connection ''; proxy_http_version 1.1; chunked_transfer_encoding off; }proxy_buffering off是关键，否则 Nginx 会把服务端的输出攒起来一起发，流式效果就没了。proxy_read_timeout要设得比模型最长思考时间还大。客户端层面，fetch本身没有超时限制，但如果你用了某些封装库，要检查它有没有默认超时。
我在一个项目里就遇到过：本地测试一切正常，部署到线上后流式输出变成"憋一大段再一次性出来"。排查了半天，最后发现是网关默认开启了响应缓冲。这类问题不看日志、不逐层排查，光看代码是找不到的。
3. LangChain 结构化输出：让模型吐出能直接用的 JSON
3.1 结构化输出解决的是什么问题
大模型默认返回的是自然语言。你问它"帮我提取这段文本里的人名和电话"，它可能回你"好的，这段文本里的人是张三，电话是138xxxx"。这对人来说没问题，但你的代码没法直接用——你总不能写正则去抠"张三"和那串数字吧，模型换个措辞你的正则就废了。
结构化输出要做的就是：约束模型必须返回一个符合预定 schema 的 JSON。比如你定义一个 Pydantic 模型：
from pydantic import BaseModel, Field class PersonInfo(BaseModel): name: str = Field(description="人物姓名") phone: str = Field(description="联系电话") city: str = Field(description="所在城市")然后让模型输出，你期望拿到的是：
{"name": "张三", "phone": "13800000000", "city": "杭州"}这样你的代码就能直接PersonInfo.model_validate_json(...)拿到一个类型安全的对象，字段缺失或类型不对会直接报错，而不是悄悄给你一个错误的字符串。
3.2 LangChain 里实现结构化输出的几种路子
LangChain 提供了with_structured_output这个方法，用法很直接：
from langchain_openai import ChatOpenAI llm = ChatOpenAI(model="gpt-4o-mini") structured_llm = llm.with_structured_output(PersonInfo) result = structured_llm.invoke("张三住在杭州，电话是13800000000") print(result) # PersonInfo(name='张三', phone='13800000000', city='杭州')它底层其实做了两件事：一是把你的 Pydantic 模型转成 JSON Schema，二是通过模型的function calling / tool calling能力或者JSON mode来约束输出格式。不同模型支持的方式不一样，LangChain 会帮你选一个可用的。
这里有个关键选择点：用 tool calling 还是用 JSON mode。tool calling 的约束更强，模型被强制按照工具参数 schema 来输出，格式正确率更高；JSON mode 只是告诉模型"请输出 JSON"，但不保证字段完全符合你的 schema。所以如果你的模型支持 tool calling，优先用它。LangChain 的with_structured_output默认会走 tool calling，如果模型不支持再降级。
还有一种更"土"但更可控的方式：在 prompt 里明确要求输出 JSON，并给出示例，然后自己解析。这种方式灵活，但稳定性依赖模型能力，字段一多就容易漏。我一般只在模型不支持 tool calling 的时候才用这招。
3.3 结构化输出和流式的天然矛盾
现在问题来了。结构化输出要求模型返回完整的、合法的 JSON，而流式输出是一段段吐字符的。这两者放在一起，就产生了一个根本矛盾：你收到的每一个 chunk 都是残缺的 JSON，单独拿出来根本没法解析。
举个例子，模型流式返回的内容可能是这样的：
chunk1: {"na chunk2: me": "张 chunk3: 三", "phone": chunk4: "13800000000" chunk5: , "city": "杭州"}你把 chunk1 拿去JSON.parse，直接报错。你把 chunk1 到 chunk4 拼起来，还是缺个右括号，照样报错。只有拼到 chunk5，才是一个完整的 JSON。
那怎么办？有两个思路。思路一：流式只用来做展示，结构化解析等流结束再做。也就是前端先把原始文本流式显示出来，等流结束后，服务端把完整文本解析成 JSON 返回。这个方案简单，但用户体验上，用户看到的是一堆 JSON 原文在往外冒，不好看。
思路二：增量解析。每收到一个 chunk 就拼到缓冲区，然后尝试解析缓冲区里的内容，能解析出多少算多少。这个方案体验好，但实现复杂，需要一个能处理"不完整 JSON"的解析器。这就是下一节要重点讲的内容。
4. 流式 JSON 的增量解析：怎么在残缺状态下提取数据
4.1 为什么不能直接 JSON.parse
先明确一个事实：标准的JSON.parse是全量解析，它要求输入是一个完整的、合法的 JSON 字符串。少一个括号、多一个逗号、字符串没闭合，都会直接抛异常。而流式场景下，你拿到的缓冲区在绝大多数时刻都是"不完整"的，所以直接 parse 必然失败。
有人会说，那我 try-catch 一下，失败了就等下一个 chunk 呗。这个思路方向是对的，但有个问题：你没法区分"因为不完整而失败"和"因为格式真的错了而失败"。如果模型返回的 JSON 本身就有语法错误，你一直等下去也等不到合法结果，最后就是无限等待。所以需要一个更聪明的解析策略。
4.2 增量解析的两种实用方案
方案一：括号配对 + 部分解析
核心思路是维护一个缓冲区，每次收到新 chunk 就追加进去，然后扫描缓冲区，统计括号（{}和[]）的配对情况，同时处理字符串内的转义。当发现某个对象已经闭合时，就尝试解析这个闭合的部分。
这个方案的关键在于状态机：你需要知道当前是否在字符串内部（因为字符串里的{不算括号），是否在转义状态（\"不算字符串结束）。手写一个这样的扫描器大概几十行代码，但很容易在边界情况上出错，比如 Unicode 转义、嵌套对象、数组里套对象。
方案二：用现成的流式 JSON 解析库
Python 生态里有ijson这类库，但它主要面向超大 JSON 文件的流式读取，对"边收边解析"的支持不算特别顺手。更实用的做法是用partial-json-parser或者自己封装一个"容错解析"函数：先尝试标准解析，失败后尝试补全缺失的括号再解析。
我实际项目里用的是第三种思路，基于 Pydantic 的增量校验。具体做法是：维护一个字符串缓冲区，每次追加后，先尝试用json.loads解析，成功就直接返回；失败则尝试"补全"——统计未闭合的括号数量，在末尾补上对应数量的}或]，再解析。如果补全后能解析成功，说明当前缓冲区是一个"前缀合法"的 JSON，可以提取出已经完整的字段。
import json def try_parse_partial(buffer: str): # 先尝试直接解析 try: return json.loads(buffer) except json.JSONDecodeError: pass # 尝试补全括号 stack = [] in_string = False escape = False for ch in buffer: if escape: escape = False continue if ch == '\\': escape = True continue if ch == '"': in_string = not in_string continue if in_string: continue if ch in '{[': stack.append(ch) elif ch in '}]': if stack: stack.pop() # 补全未闭合的括号 closing = '' for ch in reversed(stack): closing += '}' if ch == '{' else ']' try: return json.loads(buffer + closing) except json.JSONDecodeError: return None这个函数返回None就表示当前缓冲区还不足以解析出任何有效内容，继续等下一个 chunk。返回字典就表示已经能提取出部分字段了。
注意：补全括号解析出来的结果，某些字段可能是"半截"的。比如
{"name": "张补全成{"name": "张"}，你拿到的 name 是"张"，但实际完整值可能是"张三"。所以增量解析的结果只能用于展示，不能用于最终入库。最终数据一定要等流结束后用完整内容重新解析一次。
4.3 结构化输出流式场景下的特殊处理
如果你用的是 LangChain 的with_structured_output配合流式，情况会稍微不同。LangChain 在流式模式下，astream返回的 chunk 可能是AIMessageChunk，里面的tool_call_chunks字段会携带增量 JSON 片段。你需要把这些片段按顺序拼接，才能得到完整的工具调用参数。
from langchain_openai import ChatOpenAI llm = ChatOpenAI(model="gpt-4o-mini") structured_llm = llm.with_structured_output(PersonInfo) buffer = "" async for chunk in structured_llm.astream("张三住在杭州，电话13800000000"): # chunk 是增量对象，需要累积 buffer += str(chunk) partial = try_parse_partial(buffer) if partial: print("当前已解析:", partial)这里有个细节：不同版本的 LangChain 对结构化输出流式的支持程度不一样。有些版本会直接把增量 JSON 片段放在chunk里，有些版本则要求你用astream_events来监听更细粒度的事件。我建议先打印一下 chunk 的实际结构，看清楚它到底吐的是什么，再决定怎么拼。盲目照抄文档很容易踩版本差异的坑。
5. 打字机效果的前端实现：从数据流到视觉呈现
5.1 用 fetch 读取流并逐块渲染
前端要做的第一件事是拿到流。前面说过，EventSource不支持 POST，所以用fetch：
async function streamChat(prompt, onChunk) { const response = await fetch('/api/chat', { method: 'POST', headers: { 'Content-Type': 'application/json' }, body: JSON.stringify({ prompt }) }); const reader = response.body.getReader(); const decoder = new TextDecoder('utf-8'); let buffer = ''; while (true) { const { done, value } = await reader.read(); if (done) break; buffer += decoder.decode(value, { stream: true }); // 按 SSE 格式切分消息 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; onChunk(JSON.parse(data)); } } } }这里有几个关键点。decoder.decode(value, { stream: true })的stream: true很重要，它保证多字节字符（比如中文）在跨 chunk 边界时不会被截断成乱码。buffer.split('\n\n')按 SSE 的消息分隔符切分，lines.pop()把最后一段不完整的消息留在缓冲区，等下一个 chunk 来了再拼。这个"留尾巴"的处理是流式解析的通用套路，少了它就会出现消息被截断的 bug。
5.2 打字机效果的两种实现层次
拿到 chunk 之后，怎么把它变成"打字机"？这里有两个层次。
层次一：chunk 级渲染。模型每吐一个 chunk，你就把 chunk 的内容追加到界面上。这是最简单的做法，但效果取决于模型吐字的粒度。有些模型一次吐好几个字，看起来就是"一跳一跳"的，不够顺滑。
层次二：字符级渲染。把收到的 chunk 先放进一个队列，然后用一个定时器（比如每 20 毫秒）从队列里取一个字符渲染出来。这样无论模型吐字粒度多大，视觉上都是均匀的逐字显示。这个方案体验最好，但要注意队列积压的问题——如果模型吐字速度远快于渲染速度，队列会越积越长，最后用户看到的内容严重滞后。
我的经验是做一个自适应速度：队列短的时候慢点渲染（显得从容），队列长的时候加快渲染速度（追上进度）。具体可以用队列长度动态调整定时器间隔，或者一次渲染多个字符。
let queue = []; let rendering = false; function enqueue(text) { queue.push(...text); if (!rendering) renderLoop(); } function renderLoop() { rendering = true; if (queue.length === 0) { rendering = false; return; } // 队列越长，一次渲染越多字符 const batch = Math.max(1, Math.floor(queue.length / 10)); const chars = queue.splice(0, batch).join(''); appendToDom(chars); const delay = queue.length > 50 ? 5 : 20; setTimeout(renderLoop, delay); }5.3 流式渲染中的 Markdown 与代码块处理
大模型的输出经常包含 Markdown 格式，比如代码块、列表、加粗。如果你直接把原始文本塞进 DOM，用户看到的就是一堆**和 ，体验很差。所以需要实时渲染 Markdown。
但实时渲染 Markdown 有个坑：不完整的 Markdown 语法会导致渲染错乱。比如模型正在输出一个代码块，刚吐了```py，还没吐结束的```，这时候渲染器可能把后面的所有内容都当成代码。解决办法是延迟渲染：对代码块这类需要成对出现的语法，检测到开始标记后先缓存，等结束标记出现再渲染。或者用一个容错性好的 Markdown 渲染器，它对未闭合的语法有自己的处理策略。
另一个细节是代码高亮。流式场景下代码是逐渐出现的，如果每来一个字符就重新高亮整段代码，性能会很差。我的做法是：代码块在流式过程中先用等宽字体纯文本显示，等这个代码块闭合后再做一次高亮。这样既保证了流式过程的流畅，又保证了最终效果的美观。
6. 把整条链路串起来：一个可复现的完整方案
6.1 服务端：FastAPI + LangChain 流式接口
服务端我用 FastAPI 来搭，因为它对异步和流式响应的支持很自然。核心是一个返回StreamingResponse的接口：
from fastapi import FastAPI from fastapi.responses import StreamingResponse from langchain_openai import ChatOpenAI import json app = FastAPI() llm = ChatOpenAI(model="gpt-4o-mini", streaming=True) async def event_generator(prompt: str): try: async for chunk in llm.astream(prompt): if chunk.content: payload = json.dumps({"content": chunk.content}, ensure_ascii=False) yield f"data: {payload}\n\n" yield "data: [DONE]\n\n" except Exception as e: error = json.dumps({"error": str(e)}, ensure_ascii=False) yield f"data: {error}\n\n" @app.post("/api/chat") async def chat(body: dict): return StreamingResponse( event_generator(body["prompt"]), media_type="text/event-stream", headers={ "Cache-Control": "no-cache", "X-Accel-Buffering": "no", } )X-Accel-Buffering: no这个 header 是给 Nginx 看的，告诉它不要缓冲这个响应。ensure_ascii=False保证中文不被转义成\uXXXX，减少传输体积。[DONE]是一个约定俗成的结束标记，前端收到它就停止读取。
6.2 结构化输出与流式展示的分离设计
如果你的场景既要流式展示，又要结构化数据，我建议把两件事分开做。具体来说：流式接口只负责把模型的自然语言输出吐给前端做展示；等流结束后，前端把完整文本回传给一个单独的解析接口，服务端用with_structured_output或者直接调模型做结构化提取，返回干净的 JSON。
这样做的好处是职责清晰。流式接口只管"快"，结构化接口只管"准"。如果硬要把两者揉在一起，你会陷入"既要增量解析又要保证字段完整"的泥潭，代码复杂度飙升，还容易出 bug。
当然，如果你的场景对实时性要求极高，比如用户边看边要看到提取出的字段，那就只能用增量解析方案。这时候记住前面说的原则：增量解析的结果只用于展示，最终数据以流结束后的完整解析为准。
6.3 实测中遇到的几个典型问题
问题一：中文乱码。前面提过，TextDecoder要加stream: true。另外服务端json.dumps要加ensure_ascii=False，否则中文会被转义，虽然不影响正确性，但传输体积会变大。
问题二：流式输出"憋大招"。表现是前端等很久，然后一次性收到全部内容。这几乎都是中间层缓冲导致的。排查顺序：先看 Nginx 有没有proxy_buffering off，再看网关有没有开响应缓冲，最后看服务端框架有没有自己的缓冲机制。FastAPI 的StreamingResponse默认是不缓冲的，但如果你在生成器里做了批量处理，也可能导致这个问题。
问题三：结构化输出字段缺失。用with_structured_output时，偶尔会遇到模型返回的 JSON 缺字段。这通常是因为模型能力不够，或者 schema 太复杂。解决办法：一是换更强的模型，二是把 schema 拆简单点，三是给字段加详细的description帮助模型理解。Pydantic 的Field(description=...)不是摆设，它会被转成 JSON Schema 的一部分传给模型，描述写得越清楚，模型填得越准。
问题四：流中断后无法恢复。SSE 协议本身支持通过Last-Event-ID做断点续传，但大模型场景下实现起来比较麻烦，因为模型不会"从第 N 个字继续生成"。实际项目里更常见的做法是：检测到流中断后，让前端提示用户"网络波动，请重试"，然后重新发起一次完整请求。如果业务允许，也可以把已经生成的部分内容保留在界面上，让用户知道断在哪里。
7. 一些不那么显然的经验之谈
做这类流式 + 结构化的项目，有几个点是我踩过坑之后才真正理解的，文档里通常不会强调。
第一，流式的粒度不由你控制，由模型和网络共同决定。你可能期望每个 chunk 是一个字，但实际可能一次来好几个字，也可能因为网络原因几个 chunk 合并到达。所以前端渲染一定要做缓冲队列，不能假设"来一个 chunk 就渲染一个 chunk"。这个假设一旦成立，遇到网络抖动就会出现内容跳跃。
第二，结构化输出的 schema 要"扁平"。我试过用嵌套很深的 Pydantic 模型（对象里套对象，再套数组），结果模型经常在深层字段上出错。后来把 schema 改成扁平的，字段名用下划线连接表示层级关系，正确率明显提升。模型对深层嵌套结构的处理能力，比我们想象的要弱。
第三，超时设置要分层考虑。客户端、网关、服务端、模型 API 各有各的超时。任何一层的超时小于模型的最长响应时间，都会导致流中断。我的做法是：模型 API 超时设最大（比如 300 秒），网关和服务端设 310 秒，客户端不设超时（或者设一个很大的值）。这样超时永远发生在最外层，方便定位问题。
第四，测试流式一定要用真实网络环境。本地localhost测试流式，几乎不会遇到缓冲和超时问题，因为中间没有任何代理层。一部署到线上，各种问题就冒出来了。所以有条件的话，测试环境要尽量模拟生产环境的网络拓扑，至少要有 Nginx 这一层。
第五，增量解析的"补全"逻辑要处理数组。前面给的try_parse_partial函数处理了对象和数组的括号补全，但实际场景里还有一种情况：数组元素是对象，且对象还没闭合。比如{"items": [{"name": "a"}, {"name": "b，这时候补全逻辑要能正确识别出需要补"}]}。这个逻辑用栈来处理是最稳妥的，但要注意字符串内的括号和转义字符。
最后分享一个调试技巧：在开发流式功能时，我会在服务端把每个 chunk 的内容和到达时间打上日志，前端也把每个 chunk 的接收时间打上日志。这样一旦出现"卡顿"或"跳跃"，对比两端的时间戳就能快速定位是服务端吐得慢，还是网络传输慢，还是前端渲染慢。这个习惯帮我省下了大量排查时间。
