1. 为什么流式输出不是“锦上添花”而是体验分水岭很多人第一次做大模型应用都会经历一个相同的心理落差本地跑通一个问答 Demo输入问题等两三秒答案“啪”地一下整段出现。功能没问题但用起来就是别扭——像对着一个沉默的人说话你不知道他是在思考、卡住了还是根本没收到你的问题。而一旦你体验过那种文字一个个往外蹦、像有人在实时敲键盘的效果就再也回不去了。这就是打字机效果的魔力它背后依赖的核心技术就是SSE也就是 Server-Sent Events。SSE 本质上是一种基于 HTTP 的单向流式推送协议。注意关键词单向、基于 HTTP、流式。它不像 WebSocket 那样需要双向握手、维护长连接状态而是让服务端在一个普通的 HTTP 响应里持续不断地往客户端“吐”数据。浏览器端用EventSource就能接服务端只要保持连接不关闭、按固定格式写数据即可。这种“轻量”的特性让它天然适合大模型这种“请求一次、持续返回”的场景。但问题也随之而来。SSE 传过来的不是一段完整 JSON而是一堆被切碎的文本片段。你可能收到data: {content: 你}下一帧是data: {content: 好}再下一帧突然变成data: [DONE]。如果直接把这种半成品丢给前端渲染轻则乱码重则解析崩溃。更麻烦的是当你用LangChain做结构化输出时模型返回的往往是一个正在生成中的 JSON 字符串——它可能缺右括号、缺引号甚至字段名只写了一半。这时候你既要保证打字机效果流畅又要在流式过程中尽可能早地解析出结构化数据难度直接翻倍。这篇内容就是冲着这个矛盾来的。我会从 SSE 的底层数据格式讲起拆解流式分片的边界问题再进入 LangChain 结构化输出的流式解析实战最后给出一套能同时兼顾“打字机体验”和“JSON 容错解析”的完整方案。适合已经用 FastAPI 或类似框架搭过后端、用 Vue 或 React 写过前端、并且正在被流式解析折磨的开发者。如果你还在用“等全部返回再渲染”的老路子看完这篇你应该会想立刻重构。2. SSE 数据帧的真实长相别被EventSource惯坏了2.1 一个 SSE 响应到底长什么样很多人对 SSE 的认知停留在“服务端yield字符串前端onmessage接收”。但真正调试时你会发现服务端写出去的内容和浏览器收到的内容之间隔着一层协议规范。一个标准的 SSE 事件流每个事件由若干行组成行与行之间用\n分隔事件之间用空行分隔。常见字段有四个data实际传输的数据内容可以有多行多行会被拼接。event自定义事件类型前端可以用addEventListener监听。id事件 ID用于断线重连时告诉服务端“我从哪继续”。retry重连等待时间单位毫秒。一个典型的大模型流式响应服务端实际写出的字节流大概是这样data: {choices:[{delta:{content:你}}]} data: {choices:[{delta:{content:好}}]} data: [DONE]注意每个data:后面跟一个空格然后才是内容最后是两个换行。浏览器端的EventSource会自动帮你解析这些格式把data字段的内容通过onmessage回调抛出来。但这里有个巨大的坑EventSource只支持 GET 请求且不能自定义请求头。这意味着你没法在请求里带Authorization也没法传复杂的 JSON body。所以实际项目中绝大多数人不会用原生EventSource而是用fetch配合ReadableStream手动解析。2.2 手动解析 SSE 时最容易踩的三个坑第一个坑是分片边界。TCP 是流式协议不保证你一次read()就能拿到一个完整的事件。你可能收到data: {content: 你下一段才是好}。如果你按“每次读取就解析 JSON”的思路写必然报错。正确做法是维护一个缓冲区按\n\n切分事件只有拿到完整事件块才去解析。第二个坑是多行 data 的拼接规则。SSE 规范里如果同一个事件里有多个data:行它们会用\n连接。但很多服务端实现包括部分大模型 API其实只发一行data所以很多人忽略了这条规则。一旦你对接的服务端发了多行而你没做拼接解析就会丢内容。第三个坑是**[DONE]标记的处理**。OpenAI 风格的流式接口会在最后发一个data: [DONE]这不是合法 JSON。如果你统一走JSON.parse到这里就会抛异常。必须显式判断这个标记遇到就结束流读取。下面是一段经过实战检验的fetch流式解析核心逻辑用 JavaScript 写前端通用async function parseSSEStream(response, onDelta, onDone) { 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 events buffer.split(\n\n); // 最后一段可能不完整留在缓冲区 buffer events.pop() || ; for (const event of events) { const lines event.split(\n); let dataContent ; for (const line of lines) { if (line.startsWith(data:)) { // 去掉 data: 和紧随其后的一个空格 const chunk line.slice(5).replace(/^ /, ); dataContent chunk; } } if (!dataContent) continue; if (dataContent [DONE]) { onDone onDone(); return; } try { const parsed JSON.parse(dataContent); onDelta onDelta(parsed); } catch (e) { // 这里不要直接抛记录日志继续避免单个坏帧中断整个流 console.warn(SSE 帧解析失败:, dataContent, e); } } } }这段代码里有两个细节值得单独说。一是decoder.decode(value, { stream: true })里的stream: true它保证多字节字符比如中文在跨分片时不会被截断成乱码。二是buffer events.pop() || 把最后一个可能不完整的事件留在缓冲区等下一段数据来了再拼。这两点如果漏掉中文场景下大概率会出现“半个字”的乱码。2.3 服务端怎么写才规范后端如果用 FastAPI返回 SSE 流的标准写法是返回一个StreamingResponse媒体类型设为text/event-stream。关键是要设置几个响应头否则可能被中间层缓冲导致前端迟迟收不到数据from fastapi import FastAPI from fastapi.responses import StreamingResponse app FastAPI() async def event_generator(): for chunk in [你, 好, 呀]: yield fdata: {{\content\: \{chunk}\}}\n\n yield data: [DONE]\n\n app.get(/stream) async def stream(): return StreamingResponse( event_generator(), media_typetext/event-stream, headers{ Cache-Control: no-cache, Connection: keep-alive, X-Accel-Buffering: no, }, )X-Accel-Buffering: no这个头是给 Nginx 看的告诉它不要缓冲这个响应。我见过太多案例本地跑得好好的一上生产环境打字机效果就变成“憋一大段突然全出来”十有八九就是反向代理层开了缓冲。Cache-Control: no-cache和Connection: keep-alive也是标配前者防止中间缓存后者保持连接。提示如果你的流式接口经过网关或负载均衡务必确认它们没有对text/event-stream做缓冲或超时截断。很多“流到一半断了”的问题根源不在代码而在中间层配置。3. LangChain 结构化输出的流式困境JSON 是“长出来”的3.1 结构化输出为什么和流式天生冲突LangChain 的结构化输出通常是通过with_structured_output或者PydanticOutputParser来实现的。它的目标是让模型返回一个符合指定 Schema 的 JSON 对象比如{ title: 文章标题, tags: [AI, 流式], summary: 一段摘要 }问题在于当这个 JSON 还在流式生成时它是一棵“长到一半的树”。你可能先收到{title: 文章然后是标题, tags: [再是AI, 流式], summary: 一段……在任意一个中间时刻这个字符串都不是合法 JSON。你没法直接JSON.parse也没法用 Pydantic 校验。那为什么还要在流式场景下做结构化输出因为用户体验。如果等整个 JSON 生成完再解析再渲染那打字机效果就没了用户还是要干等。理想状态是模型一边生成前端一边把已经确定的字段值显示出来比如标题先出来、标签逐个出现、摘要慢慢填充。这要求我们在流式过程中做增量解析。3.2 LangChain 的astream_events能给你什么LangChain 提供了astream_events这个 API它能在流式过程中抛出各种事件包括on_chat_model_stream也就是模型每吐一个 token 就触发一次。你可以从中拿到chunk.content这就是当前新增的文本片段。对于普通文本输出直接拼接即可但对于结构化输出你需要把这些片段累积起来然后尝试解析。这里有个关键认知LangChain 的结构化输出在流式模式下底层依然是文本流。模型并不是真的在“生成对象”它只是在生成一段符合 JSON 格式的文本。所以你要做的就是把这串文本当作“正在生长的 JSON”来处理。一个常见的错误做法是每收到一个 chunk 就尝试json.loads整个累积字符串失败就跳过。这样做有两个问题。第一性能差每次都要重新解析整个字符串。第二当 JSON 很长时中间态几乎永远不合法你直到最后才能解析成功等于没有流式效果。3.3 增量解析的核心思路从“整体校验”转向“字段提取”真正可行的方案是放弃在流式过程中做完整 JSON 校验转而做“字段级提取”。也就是说我不关心整个 JSON 是否合法我只关心“当前已经出现的字段里哪些字段的值已经完整了”。比如我检测到title: ...这个模式已经闭合就把 title 提取出来检测到 tags 数组里某个字符串闭合了就把它追加到标签列表。这种思路下解析器不需要等整个 JSON 完成而是像一个状态机一样逐字符扫描识别键值对的边界。这也是很多流式 JSON 解析库比如partial-json、json-stream的核心原理。下面我会给出一个自己手写的轻量级方案不依赖第三方库方便你理解原理并灵活改造。4. 手写一个能扛住半截 JSON 的流式解析器4.1 状态机设计把 JSON 当成一条流水线我们要解析的目标是一个对象里面可能有字符串、数组、数字、布尔值。流式场景下最常需要“提前展示”的是字符串字段和字符串数组。所以我设计的解析器优先支持这两类其他类型等完整后再处理。核心状态可以简化为几个OUTSIDE还没进入任何字段值。IN_STRING正在读取一个字符串值。IN_ARRAY正在读取一个数组。DONE整个 JSON 已闭合。但更实用的做法是基于正则的增量匹配。因为流式文本是逐步累积的我可以在每次累积后用正则去匹配“已经闭合的字符串字段”。比如匹配title\s*:\s*([^]*)只要这个模式匹配到了就说明 title 字段的值已经完整出现。对于数组匹配tags\s*:\s*\[([^\]]*)\]只能等数组闭合但如果我想让标签逐个出现就需要更细的匹配([^]*)在数组范围内逐个提取。下面是一个 Python 版的增量解析器思路是维护累积文本每次新 chunk 到来后尝试提取所有“已闭合”的字符串值import re import json class StreamingJSONExtractor: def __init__(self): self.buffer self.extracted {} self.array_items {} def feed(self, chunk: str): self.buffer chunk self._extract_string_fields() self._extract_array_items() return self.extracted def _extract_string_fields(self): # 匹配 key: value 且 value 已闭合 pattern r(\w)\s*:\s*([^\\]*(?:\\.[^\\]*)*) for match in re.finditer(pattern, self.buffer): key, value match.group(1), match.group(2) # 跳过数组内部的字符串简单判断该 key 后面是否紧跟 [ if f{key} in self.extracted: continue self.extracted[key] value def _extract_array_items(self): # 匹配 key: [ ... ] 中已经出现的字符串项 array_pattern r(\w)\s*:\s*\[([^\]]*)\] for match in re.finditer(array_pattern, self.buffer): key match.group(1) inner match.group(2) items re.findall(r([^\\]*(?:\\.[^\\]*)*), inner) if items: self.array_items[key] items self.extracted[key] items这段代码的逻辑是每次feed新内容后用正则扫描整个缓冲区找出所有已经闭合的字符串字段和数组项。已经提取过的字段不重复覆盖保证“先到先得”。对于数组只要数组还没闭合正则\[([^\]]*)\]匹配不到但我们可以退一步用一个更宽松的方式先定位到tags: [之后的内容再从中提取所有已闭合的字符串。4.2 处理转义字符和嵌套结构上面的正则用了[^\\]*(?:\\.[^\\]*)*这个模式它能正确处理字符串里的转义引号比如他说\你好\。这是流式解析里很容易忽略的点——如果模型返回的内容里包含引号简单用[^]*会在转义引号处提前截断。对于嵌套对象比如{author: {name: 张三}}上面的扁平正则就无能为力了。这时候需要引入一个轻量的括号计数状态机。我的建议是流式阶段只提取顶层简单字段嵌套结构等完整后再用标准json.loads解析。因为嵌套结构的中间态太复杂强行增量解析收益低、风险高。用户体验上先把标题、摘要这些顶层字段展示出来已经足够形成“打字机感”。4.3 和 LangChain 的对接方式在 LangChain 里你可以这样把流式事件和解析器串起来from langchain_openai import ChatOpenAI from langchain_core.prompts import ChatPromptTemplate llm ChatOpenAI(modelgpt-4o-mini, streamingTrue) prompt ChatPromptTemplate.from_template( 请以 JSON 格式输出包含 title、tags、summary 三个字段。主题{topic} ) chain prompt | llm extractor StreamingJSONExtractor() async for event in chain.astream_events({topic: SSE 流式原理}, versionv2): if event[event] on_chat_model_stream: chunk event[data][chunk].content if chunk: result extractor.feed(chunk) # 这里可以把 result 推给前端实现字段级打字机 print(result)注意astream_events的versionv2参数不同版本的 LangChain 事件结构有差异v2 是目前比较稳定的。另外chunk.content在某些模型下可能是列表比如包含多模态内容需要做类型判断。注意不要在每个 chunk 到来时都重新创建解析器解析器必须是有状态的跨 chunk 累积。这是流式解析和普通解析最大的区别。5. 前端打字机效果别用setInterval硬凑5.1 流式渲染的两种模式前端拿到流式数据后渲染方式大致分两种。一种是逐字渲染每收到一个字符就追加到 DOM配合 CSS 光标闪烁形成打字机效果。另一种是逐块渲染收到一个语义块比如一个字段值就更新对应区域。前者适合纯文本对话后者适合结构化卡片。很多人做打字机效果时喜欢用setInterval定时从完整文本里“吐”字符。这种做法在流式场景下是多余的因为数据本身就是流式来的你只需要在onDelta回调里追加内容即可。用定时器反而会引入延迟和不同步问题。一个简洁的 Vue 3 实现template div classoutput h3{{ structured.title }}/h3 div classtags span v-fortag in structured.tags :keytag{{ tag }}/span /div p{{ structured.summary }}span classcursor|/span/p /div /template script setup import { reactive } from vue; const structured reactive({ title: , tags: [], summary: }); async function startStream() { const response await fetch(/api/stream, { method: POST, headers: { Content-Type: application/json }, body: JSON.stringify({ topic: SSE 流式原理 }), }); await parseSSEStream( response, (delta) { // delta 是后端推来的结构化片段 if (delta.title) structured.title delta.title; if (delta.tags) structured.tags delta.tags; if (delta.summary) structured.summary delta.summary; }, () console.log(流结束) ); } /script这里的parseSSEStream就是第 2 节里那段函数。后端每次把解析器提取到的最新字段推给前端前端直接覆盖对应字段。这样标题会先出现标签逐个增加摘要逐渐变长视觉上就是自然的打字机效果而且不需要任何定时器。5.2 光标闪烁和滚动跟随的细节打字机效果里光标闪烁是个小细节但很影响观感。用 CSS 动画实现即可.cursor { display: inline-block; width: 2px; animation: blink 1s step-end infinite; } keyframes blink { 50% { opacity: 0; } }滚动跟随则是另一个坑。如果输出区域是固定高度的内容增长时需要自动滚到底部。但要注意如果用户手动往上滚了就不应该强制拉回底部。判断逻辑是当滚动条距离底部小于某个阈值比如 50px时才自动滚动。function autoScroll(el) { const threshold 50; const isNearBottom el.scrollHeight - el.scrollTop - el.clientHeight threshold; if (isNearBottom) { el.scrollTop el.scrollHeight; } }这个细节在长文本流式输出时特别重要否则用户想回看前面内容却被不断拉到底部体验极差。6. 那些文档不会告诉你的实战坑6.1 流式接口的“假死”与超时流式接口最让人抓狂的问题之一是“流到一半不动了”。表现是前端光标一直闪但不再有新内容过一会儿连接断开。这种情况通常有几个原因。一是模型侧生成卡住比如遇到长上下文或复杂推理token 生成速度骤降。二是中间层网关、负载均衡有 idle timeout比如 60 秒没有数据就断开。三是后端代码里某个await阻塞了事件循环。排查时先在服务端每个 chunk 发出时打时间戳日志确认是“没生成”还是“没发出”。如果是中间层超时可以在流式过程中定期发送注释行以:开头的行SSE 规范里会被忽略作为心跳async def event_generator(): while generating: chunk await get_next_chunk() if chunk: yield fdata: {json.dumps(chunk)}\n\n else: yield : heartbeat\n\n # 心跳保持连接心跳间隔建议 15 到 30 秒太频繁浪费带宽太稀疏起不到保活作用。6.2 结构化输出字段顺序不稳定LangChain 结构化输出时模型返回的字段顺序是不确定的。有时候先出summary有时候先出title。如果你的前端代码假设“title 一定先到”就会出问题。正确做法是前端对每个字段独立响应谁先到就先渲染谁不要有顺序依赖。另外模型有时会“重复输出”某个字段比如 title 出现了两次。增量解析器要做好去重已经提取过的字段不要被后面的覆盖除非你明确需要“最后一次为准”。我的经验是对于标题、摘要这类字段第一次出现的通常最准确后续可能是模型在自我修正但流式场景下频繁跳变反而影响体验所以取第一次即可。6.3 JSON 里的中文和特殊字符中文在流式传输中最大的风险是多字节字符被分片截断。一个中文字符在 UTF-8 里占 3 个字节如果 TCP 分片刚好切在中间你拿到的就是半个字符。这就是为什么第 2 节里强调TextDecoder要加{ stream: true }。Python 侧如果用yield字符串FastAPI 会帮你编码一般不会出问题但如果你手动encode再yield就要注意别在字符中间切。特殊字符方面模型返回的 JSON 里如果包含换行符会被转义成\n。你的增量解析器要能识别转义序列否则会把\n当成两个字符处理。前面正则里的(?:\\.[^\\]*)*就是干这个的。6.4 什么时候该放弃流式结构化不是所有场景都值得做流式结构化解析。如果 JSON 结构非常复杂嵌套三四层或者字段之间有依赖关系比如后面的字段需要引用前面的值那强行增量解析的复杂度会急剧上升收益却有限。这时候更务实的做法是流式传输原始文本等完整后再解析结构化数据。用户看到的是文字在流动虽然结构化卡片要等最后才出现但至少打字机体验保住了。判断标准很简单如果顶层字段不超过 5 个且都是字符串或字符串数组那就做增量解析如果结构复杂就退化为“流式文本 最终结构化”。不要为了技术而技术。7. 一套可复用的端到端方案骨架把前面的内容串起来一个完整的方案包含四层。协议层用 SSE服务端设置正确的响应头和心跳客户端用fetchReadableStream手动解析处理好分片边界和[DONE]标记。模型层用 LangChain 的astream_events获取 token 流注意版本和 chunk 类型判断。解析层用有状态的增量解析器优先提取顶层字符串字段和数组项做好去重和转义处理。渲染层用响应式框架按字段更新配合光标动画和智能滚动。这套骨架我在几个项目里复用下来最深的体会是流式体验的瓶颈往往不在模型速度而在中间环节的缓冲和解析策略。把 SSE 的帧边界处理干净、把增量解析的状态维护好剩下的就是水到渠成。反过来如果这两层没做扎实模型再快用户看到的也可能是“卡半天然后一大段”。最后分享一个调试技巧在开发阶段把服务端每个 chunk 的原始内容和时间戳打到日志里同时在前端把每次onDelta收到的数据也打出来。两边对照能快速定位是“没发出来”还是“没解析对”。这个笨办法帮我省下了大量猜测时间。