★ FEATURED ARTICLE 干过 AI 应用接入的都知道最磨人的不是模型效果差而是流式输出和结构化解析这两件事碰在一起。你兴致勃勃把 LangChain 的流式结果一端一接前端想做出打字机效果结果不是缓冲区被吞、SSE 断在 idle timeout就是模型吐出来的 JSON 七零八落根本没法解析。其实这些问题背后都是同一根链条SSE 协议怎么传输、后端怎么用 LangChain 组织流、前端怎么消费流、以及结构化输出在流式场景下到底该怎么解析。这篇文章把我踩过的坑和最终跑通的方案完整写出来适合正在接大模型应用的后端、前端同学参考也适合那些刚接触 LangChain 想一步到位实现打字机效果 JSON 结构化输出的人。1. SSE 流的底子为什么它比普通 HTTP 响应更适合 LLM 场景很多教程直接给一段fetchgetReader的代码就完事但一旦让你自己排查为什么拿到乱码、为什么流半路断了不懂协议细节就寸步难行。SSE 全称 Server-Sent Events是 HTML5 时代就定下的标准核心思路很简单服务器把一块数据分多次通过同一个 HTTP 连接推给客户端内容类型是text/event-stream。1.1 SSE 帧格式其实只有几个字段SSE 的每个事件由若干字段组成常见的字段就这几个data:实际内容可以多行多行会被拼接成一个完整数据段换行符拼接为\n。event:自定义事件名默认是message。id:事件 ID客户端断线重连时可以带上Last-Event-ID。retry:重连间隔单位毫秒。注释行以:开头通常是: keep-alive这样的心跳。一个最简 SSE 流长这样data: {delta:你} data: {delta:好}注意每个事件必须用空行也就是两个\n分隔。客户端解析时会读取到data: xxx然后遇到空行就触发一个message事件。这个格式看着简单但实际生产里最容易踩的坑有三个一是编码问题data里的中文必须让服务端保证以 UTF-8 输出否则前端按 UTF-8 解码就会乱码二是流式传输时一个data字段可能被 TCP 切片前端如果不做缓冲合并就会把一个 event 拆成半个解析三是缓冲区问题Nginx 或网关如果开了 buffering内容会被堵在代理层前端拿到的是攒了好几秒后的一次性大块数据打字机效果直接变成 卡顿机。1.2 SSE 和 WebSocket 的取舍什么时候不该选 WS提到流式有人第一反应是 WebSocket。但 LLM 场景我建议默认 SSE除非你确实需要双向实时通信比如用户在线协作编辑。理由很实在SSE 基于普通 HTTP天然穿过大部分代理和负载均衡器服务端不需要单独维护连接状态。EventSource 或 fetch 都支持自动重连WebSocket 断线后重连逻辑得自己写。SSE 是单向的正好匹配 用户发一句话服务器持续推结果 的对话模式。LangChain 官方异步流接口astream本身就是异步生成器用 FastAPI 的StreamingResponse包一层就是 SSE成本极低。当然 SSE 也有局限浏览器对 EventSource 不支持 POST 和自定义 Header如果你要带鉴权 Token就得用fetch自己解析IE 系列完全不支持。所以我的方案里前端统一用fetchReadableStream实现不依赖EventSource。2. 后端接入点用 FastAPI LangChain 构造规范的 SSE 接口先明确一个原则后端流式接口输出的数据格式必须跟协议层的业务语义分离。我建议所有通过 SSE 传输的内容都用 JSON 包一层不要直接吐纯 token 字符串。这样前端解析时能区分事件类型增量、结束、错误、元数据后面做结构化输出也顺理成章。2.1 为什么把 LangChain 的astream包成data:事件LangChain 的model.astream()返回一个异步生成器每次吐出一个 token 或 chunk。如果用 FastAPI 的StreamingResponse直接返回生成器默认 media type 是text/plain完全不是 SSE。我们需要自己包装成上面的帧格式。下面是一个最小可用的后端接口骨架import json from fastapi import FastAPI from fastapi.responses import StreamingResponse from langchain_openai import ChatOpenAI from langchain_core.messages import HumanMessage app FastAPI() model ChatOpenAI(modelgpt-4o-mini, temperature0) app.get(/chat/stream) async def chat_stream(q: str): async def event_generator(): # 开始事件标记一次请求的开始携带 request_id 方便排查 yield fevent: start\ndata: {json.dumps({request_id: req-123}, ensure_asciiFalse)}\n\n # 增量事件默认事件名 message客户端监听 message async for chunk in model.astream([HumanMessage(contentq)]): text chunk.content if hasattr(chunk, content) else str(chunk) payload json.dumps({delta: text}, ensure_asciiFalse) yield fevent: delta\ndata: {payload}\n\n # 结束事件明确告诉前端流已经结束 yield fevent: done\ndata: {json.dumps({status: ok})}\n\n response StreamingResponse(event_generator(), media_typetext/event-stream) # 禁止代理缓冲防止 SSE 被攒批 response.headers[X-Accel-Buffering] no response.headers[Cache-Control] no-cache response.headers[Connection] keep-alive return response这里有几个关键选择我逐个说明。event: start/event: delta/event: done这套自定义事件名是配合前端EventSource-like解析用的。如果你全部用默认message前端就只能靠data里的字段自己判断类型多一层重复逻辑。分离成本几乎为零建议一开始就设计好。X-Accel-Buffering: no这个头是针对 Nginx 的。Nginx 默认会对 FastCGI 和代理响应做缓冲不关掉的话SSE 的data会在 Nginx 层积压前端收到的是聚集后的一次性传输打字机效果彻底失效。如果你用 Go 的net/http或 Node 的 stream同样建议显式禁止缓冲。Connection: keep-alive是 HTTP/1.1 的常规操作但注意 FastAPI 用的 Uvicorn 在 HTTP/2 下这个头会被忽略问题不大。真正影响寿命的是后面章节里说的 idle timeout。2.2 处理认知流content、tool_calls 和 metadata 三类增量LangChain 生态里不同模型的astream吐出来的 chunk 结构不一样。OpenAI 这类工具调用模型最终回复可能是纯文本也可能是 tool_call 片段直接chunk.content取文本没问题但碰到 Claude、Qwen 这类既有文本又有工具调用的模型chunk 里还会有tool_call_chunks。如果你要的是面向最终用户的打字机文本只取content即可如果要结构化输出那要看下一节的方案。建议在后端做一次归一化把上游模型的 chunk 统一转成{ type: text | json | tool_call, content: ... }再包进 SSE。这样前端不需要感知具体用的哪家模型未来替换模型也不用动前端。我实际生产里还做了一层片段累积器。因为某些场景需要拿到完整上下文才能做业务判断但流式传输时后端只负责转发不能阻塞等待完整结果。一个平衡方案是普通对话走delta流式转发如果需要后处理比如敏感词检查、术语替换则在delta之间插入一个info事件告诉前端本条是后处理中段标记前端遇到info事件就暂时冻结光标。这样读者看到的仍然是连续打字但后端已经悄悄做了若干次校验。3. 流式响应在代理层的生命周期解决 idle timeout 与断连重连如果你只是本地演示跑通上面代码很容易。但一旦部署到线上最经典的问题就来了stream disconnected before completion: idle timeout waiting for sse。这个报错的核心原因很简单——代理或网关认为连接在空闲把它掐断了。3.1 谁在掐断连接从负载均衡到网关连接链路通常是浏览器 - 接入层Nginx/LB/网关 - FastAPI - LLM API。任何一个环节设置了空闲超时只要在超时时间内没有数据经过连接就会被关闭。LLM 生成长文时有些模型可能在两次 token 之间停顿好几秒尤其是深度推理模型这时接入层很容易判定空闲。常见的默认超时时间参考组件常见默认值影响Nginxproxy_read_timeout60s60s 内没有收到后端数据就会断开云负载均衡空闲超时30s~60s很多云的 CLB 默认 30~60sUvicorntimeout-keep-alive5s这里主要影响 HTTP keep-alive 空闲但也要调大客户端 fetch 超时设置无需要自己用 AbortController 控制3.2 三种行之有效的解决方案第一种调大接入层超时并且关掉缓冲Nginx 的配置最小集长这样location /chat/stream { proxy_pass http://backend; proxy_buffering off; proxy_cache off; proxy_read_timeout 3600s; proxy_send_timeout 3600s; chunked_transfer_encoding off; }proxy_buffering off对应前面服务端设置的X-Accel-Buffering: no两层都关掉才保险。proxy_read_timeout 3600s让 Nginx 愿意等一小时。注意调大超时不意味着连接一直占用只要流里有数据连接不会被掐极端情况下业务正常结束done事件发出后连接自然关闭。第二种服务端定期发送心跳注释SSE 规范里注释行: keep-alive不会触发事件但能让数据流保持活跃代理层就被骗过去认为连接不是空闲的。我在异步生成器里加一个心跳循环import asyncio async def event_generator(): heartbeat_interval 15 while True: try: async for chunk in model.astream([HumanMessage(contentq)]): # yield delta ... heartbeat_interval 15 except asyncio.CancelledError: break # 这个实现里心跳要放在独立循环比较优雅下面是一个简化写法 await asyncio.sleep(0)实际上完整的心跳实现更适合拆成独立任务核心思路是每 15 秒如果上游没有新 token就向客户端发一行: keep-alive。我建议直接把心跳循环做成一个后台任务和 token 转发协程并发跑两个协程共用同一个asyncio.Queue谁先产生内容就发给谁。这样既保证有心跳又能保证 token 不延迟。第三种客户端做断线检测与续传前端不能把连接存活完全押在服务端和代理上。我封装了一套基于fetch的解析逻辑核心包括用AbortController设置 60 秒空转超时如果 60 秒内没有任何数据块主动断开并提示用户。收到done事件之前遇到网络错误允许用户一键重连。重连时带上Last-Event-ID服务端拿到后可以从断点继续。注意这个需要后端配合维护会话上下文不是所有模型都支持从任意 token 位置续传更可靠的方式是前端先把已展示文本缓存重连后让后端基于完整历史重新生成。4. 前端打字机效果从 fetch 读取流到渲染前端拿到 SSE 流以后最难的不是打字机动画而是可靠地解析流。浏览器提供的response.body.getReader()读取的是字节流它会按网络包的大小返回数据而不是按 SSE 的事件边界返回。所以前端必须自己维护一个 buffer把分片合并成完整的 SSE 帧再按行解析。4.1 通用 SSE 解析器readSSEStream下面这个函数我改过很多版本最终形成了一个相对健壮的写法。它接受了用户传入的几个事件回调并且把协议解析和业务逻辑分开async function readSSEStream( response: Response, handlers: { onDelta: (delta: string) void; onDone: () void; onError: (err: Error) void; } ) { if (!response.body) { handlers.onError(new Error(response body is null)); return; } 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 }); const frames buffer.split(\n\n); buffer frames.pop() ?? ; for (const frame of frames) { const lines frame.split(\n); let eventName message; const dataLines: string[] []; for (const line of lines) { if (line.startsWith(event:)) { eventName line.slice(6).trim(); } else if (line.startsWith(data:)) { dataLines.push(line.slice(5).trim()); } } if (dataLines.length 0) continue; const rawData dataLines.join(\n); if (eventName delta) { try { const parsed JSON.parse(rawData); handlers.onDelta(parsed.delta ?? ); } catch { handlers.onDelta(rawData); } } else if (eventName done) { handlers.onDone(); } else if (eventName error) { handlers.onError(new Error(rawData)); } } } }buffer.split(\n\n)是关键。因为一个 event 一定以空行结尾网络分片可能把一个 event 切成几块也可能一次传来好几个 event。这种做法先把流按空行切成完整帧剩余的半截留在 buffer 里等下一轮。还要注意TextDecoder(utf-8, { stream: true })的第二个参数。如果不传stream: true当某个多字节中文字符恰好被拆在两个网络包里时会直接解码成替换字符。传了之后解码器内部会缓存半个字符。4.2 打字机效果找到增量文本后如何渲染拿到delta文本后最直接的打字机效果是innerHTML delta。但这在长对话里很危险因为模型偶尔会生成 HTML 标签、笑脸符号甚至 markdown 语法innerHTML可能把/div当真实标签解析轻则样式错乱重则 XSS 风险。正确做法是textContent累积或者把每一段增量渲染成一个独立的 span再通过 CSS 的动画控制出现节奏function appendDelta(container: HTMLElement, delta: string) { const span document.createElement(span); span.textContent delta; span.className typing-span; container.appendChild(span); }配合这样一段 CSS每个 span 会出现淡入效果视觉上就是打字机.typing-span { opacity: 0; animation: fadeInStep 50ms ease forwards; } keyframes fadeInStep { to { opacity: 1; } }如果还要做光标闪烁可以在容器末尾放一个span.cursor永远跟随最新内容blink动画即可。这里还有一个体验细节模型生成内容时delta 可能包含\n而textContent对换行不会自动渲染成视觉换行需要用white-space: pre-wrap。如果还想要 markdown 渲染那就得把增量缓冲区做成熟练工等done之后再统一用 markdown 渲染。这个取舍取决于你的产品定位如果强调实时打字感就先用纯文本 pre-wrap如果要追求排版漂亮就接受先打字后渲染。5. 结构化输出与 JSON 解析LangChain 流式方案全拆解聊完打字机接下来是重头戏怎么让模型输出稳定的 JSON并且在流式场景下优雅地解析它。5.1 LangChain 的with_structured_output到底改变了什么很多新手会直接在前端JSON.parse模型返回的完整文本然后祈祷模型别多输出一个逗号。这种方案在复杂 schema 下脆弱得离谱。LangChain 的with_structured_output背后本质是把结构化输出这件事从文本后解析变成了模型原生生成符合 schema 的 JSON。它的实现分两条路对支持工具调用tool calling / function calling的模型LangChain 会构造一个工具工具的参数就是你要的 Pydantic 模型然后强制模型以工具调用形式输出 JSON。模型的 token 流里会包含tool_call_chunks。对不支持工具调用的模型退化为 JSON mode 或指令模式系统提示词里塞入 JSON schema同时后置一个 Pydantic 校验器解析失败就重试。我的建议是如果你用的模型平台支持工具调用优先用工具调用路线稳定性和格式正确性高一大截。下面是用 Pydantic 定义 schema 并开启结构化输出的经典写法from pydantic import BaseModel, Field from langchain_openai import ChatOpenAI class OrderInfo(BaseModel): order_id: str Field(description订单号) items: list[str] Field(description商品列表) total_amount: float Field(description总金额) is_paid: bool Field(description是否已支付) llm ChatOpenAI(modelgpt-4o-mini, temperature0) structured_llm llm.with_structured_output(OrderInfo)调用structured_llm.invoke()返回的就是OrderInfo实例而不是字符串调用structured_llm.astream()时它返回的是生成 Partial 的工具调用 chunk。5.2 两个流派JsonOutputParser 与 Tool calling 的增量差异如果你用with_structured_output的astream()来流式输出 JSON需要区分模型返回的 chunk 里是content里的 JSON 增量还是tool_call_chunks。LangChain 官方推荐与流式兼容的方法是使用JsonOutputParserfrom langchain_core.output_parsers import JsonOutputParser parser JsonOutputParser() chain llm | parser async for partial_json in chain.astream(prompt): # partial_json 是已累积解析过的部分 JSON 对象 # 比如 {order_id: A123, items: 只解析到数组前几个元素} print(partial_json)JsonOutputParser内部的机制比较巧妙它并不是每来一个 token 就重新解析整个字符串而是维护了一个 JSON restoration 逻辑能把不完整的 JSON 文本补充成合法 JSON 再解析。也就是说当模型刚生成到{order_id: A时它会把字符串补成{order_id: A}来解析从而返回一个部分对象。这让前端可以一边展示生成内容一边实时拿到已经稳定下来的字段。但用with_structured_output时astream()返回的是 partial 的BaseModel实例或包含 raw 内容的复杂结构。所以我在后端一般这样做一个兼容层from langchain_core.output_parsers import JsonOutputParser from langchain_core.prompts import ChatPromptTemplate prompt ChatPromptTemplate.from_template( 提取用户的订单信息。\n{format_instructions}\n用户输入{input} ) parser JsonOutputParser(pydantic_objectOrderInfo) chain prompt | llm | parser async def event_generator(): async for partial in chain.astream({input: q}): yield fevent: json\ndata: {json.dumps({partial: partial}, ensure_asciiFalse)}\n\n yield fevent: done\ndata: {json.dumps({status: ok})}\n\n这里前端收到的json事件里的partial字段就是一个经过努力修复后的合法 JSON 对象。你可以在打字机显示的同时把已经确定的order_id展示到卡片上而不是等全部输出结束后才出结果。5.3 怎么判断 JSON 已经稳定字段级完成度用JsonOutputParser有一个必须理解的点partial里的字段是变化的。模型起初可能猜order_id: A123, 生成到后面又修正成B456, 此时 partial 会返回新值。如果你的业务字段一旦展示就不允许变前端必须自己做字段锁定逻辑。我常用的策略定义每个字段是否可变的规则。例如order_id这类标识字段第一次解析到非空值后就锁定后续忽略变化summary这类内容字段允许一直更新直到done。如果你偷懒不做锁定前端每次拿到 partial 就全量重新渲染会发现卡片上的订单号偶尔会闪变用户很容易认为是系统 bug。另外有个实用技巧想让 JSON 输出更稳定可以把 Pydantic 的字段描述写得极其具体比如Field(description订单号例如 A123456)。描述里的示例对模型很有引导力比泛泛写订单号准确率高很多。还可以把llm的temperature0结构化输出场景下保持确定性优先。6. 完整链路走通一次流式对话 实时 JSON 卡片的实战模板前面几节把协议、后端、前端、结构化输出拆开讲了。这里我把它们拼起来给一套能直接抄作业的完整模板你可以按这个骨架改造。6.1 后端完整文件示例import json import asyncio from fastapi import FastAPI from fastapi.responses import StreamingResponse from pydantic import BaseModel, Field from langchain_openai import ChatOpenAI from langchain_core.prompts import ChatPromptTemplate from langchain_core.output_parsers import JsonOutputParser app FastAPI() class OrderInfo(BaseModel): order_id: str Field(description订单号例如 A123) items: list[str] Field(description商品名称列表) total_amount: float Field(description总金额) is_paid: bool Field(description是否已支付) llm ChatOpenAI(modelgpt-4o-mini, temperature0) prompt ChatPromptTemplate.from_template( 从用户输入中提取订单信息。\n{format_instructions}\n用户输入{input} ) parser JsonOutputParser(pydantic_objectOrderInfo) chain prompt | llm | parser app.get(/order/parse) async def parse_order(q: str): async def event_generator(): # 心跳任务 stop_heartbeat asyncio.Event() async def heartbeat(): while not stop_heartbeat.is_set(): yield : keep-alive\n\n await asyncio.sleep(15) # 这里简化为先发 start yield event: start\ndata: {}\n\n try: async for partial in chain.astream({input: q, format_instructions: parser.get_format_instructions()}): payload json.dumps({partial: partial}, ensure_asciiFalse) yield fevent: json\ndata: {payload}\n\n finally: stop_heartbeat.set() yield event: done\ndata: {}\n\n response StreamingResponse(event_generator(), media_typetext/event-stream) response.headers[X-Accel-Buffering] no response.headers[Cache-Control] no-cache return response这个示例里heartbeat协程的写法是示意性的真实场景应该把它做成独立后台任务放进StreamingResponse。不过即使不放在同一个生成器里理解逻辑也没问题——核心点是在等待模型输出的间隙必须有东西在说话代理才不掐连接。6.2 前端完整调用示例前端把上文的readSSEStream跑起来配合业务事件fetch(/order/parse?q${encodeURIComponent(inputText)}, { headers: { Accept: text/event-stream }, }) .then((res) { if (!res.ok) throw new Error(HTTP ${res.status}); return readSSEStream(res, { onDelta: (delta) appendDelta(outputContainer, delta), onDone: () finishTyping(), onError: (err) showError(err.message), }); }) .catch((err) showError(err.message));此时如果后端发的是event: jsonreadSSEStream里的eventName json分支就需要处理 partial我这里为了简洁用了onDelta承接真实项目里建议给readSSEStream增加一个onCustomEvent(eventName, data)回调把所有自定义事件统一交给上层分发onCustomEvent: (eventName, rawData) { if (eventName json) { const partial JSON.parse(rawData).partial; updateOrderCard(partial); } }6.3 打印一个实际运行路径上的踩坑清单最后总结一下我跑通这套链路至少撞过以下问题提前写给你省时间模型文本里夹杂反引号或 markdown 分隔符时千万别用innerHTML直接拼。data:后如果有多行一定要用join(\n)不要只取第一行。流式 JSON 的partial值可能重复出现同一个 key前端处理要幂等。网关层开了 GZIP 压缩后有些 SSE 实现会因为压缩缓冲导致流不平滑建议对text/event-stream关闭动态压缩。Uvicorn 单 worker 下如果有长时间不活跃的 SSE 连接timeout-keep-alive保持默认值也可能掐断 keep-alive 连接记得调大。如果模型平台本身有最大输出 token 限制SSE 流再长也不会超过模型上限可以在done事件里带一个finish_reason指明是否被截断。踩过几次坑之后我现在做 AI 流式接入都会先把事件协议定死——前后端围绕start/delta/json/info/error/done这套事件名通信任何新功能都先加事件类型不轻易塞进data字符串里做私有协议。这样无论以后换 LangChain 版本还是换模型厂商前端解析层几乎不用动。这套设计下来最直接的收益是线上再没出现过流断了前端还在傻等的事故JSON 解析的成功率也稳定在接近百分之百剩下的少数失败基本都来自用户输入本身语义模糊这已经和传输架构无关了。
阅读完成 · 觉得有帮助？
