1. 从打字机效果说起为什么流式输出不是锦上添花很多人第一次接触大模型应用开发时都会有一个疑问明明等几秒钟一次性返回完整结果也能用为什么非要折腾流式输出我刚开始做 AI 应用的时候也是这么想的直到用户反馈页面卡住了不知道是不是死机了我才意识到问题的严重性。大模型的响应时间通常在 2 到 15 秒之间复杂推理任务甚至可能超过 30 秒。如果采用传统的请求-等待-完整返回模式用户在这段时间内看到的是一个空白的加载动画体验极差。而流式输出Streaming的核心价值在于把等待时间从不可感知的黑盒变成可见的渐进过程。用户看到文字一个一个蹦出来心理上会觉得系统在思考、在工作即使总耗时不变感知体验也完全不同。这就是所谓的打字机效果——本质上是一种时间换感知的产品设计策略。但实现它并不简单涉及三个层面的技术挑战传输层如何让服务端持续推送数据而不关闭连接答案是 SSEServer-Sent Events。框架层LangChain 如何处理流式输出尤其是结构化输出场景下的流式解析前端层如何接收流式数据并实时渲染同时处理 JSON 解析和异常情况这三个层面环环相扣任何一环出问题用户看到的可能就是转圈圈或者文字突然全部蹦出来。接下来我会按照数据流动的顺序从 SSE 协议原理开始一路讲到 LangChain 结构化输出的流式处理和前端解析方案把每个环节的坑和解决方案都摊开来讲。提示本文假设你已经对 Python、FastAPI 和 Vue 有基础了解不需要精通但至少写过简单的接口和页面。如果你完全没接触过 LangChain也没关系我会在涉及的地方补充必要的背景知识。2. SSE 协议的本质一条永不关闭的 HTTP 连接2.1 SSE 和 WebSocket 到底选哪个说到服务端推送很多人第一反应是 WebSocket。但 SSE 和 WebSocket 的定位完全不同选错了会带来不必要的复杂度。对比维度SSEWebSocket通信方向单向服务端到客户端双向协议基础纯 HTTP独立协议ws://自动重连浏览器原生支持需要手动实现数据格式文本UTF-8文本 二进制实现复杂度低中高适用场景大模型流式输出、通知推送实时聊天、协同编辑、游戏大模型对话场景下客户端只需要发送一次请求然后接收服务端持续推送的 token 流根本不需要双向通信。用 WebSocket 就像用大炮打蚊子——不是不行但没必要。SSE 基于标准 HTTP天然穿透大部分代理和防火墙浏览器原生支持自动重连实现成本低得多。2.2 SSE 的数据帧格式别小看那几个换行符SSE 的协议格式看起来极其简单但魔鬼藏在细节里。一个标准的 SSE 事件流长这样data: {token: 你} data: {token: 好} data: {token: }注意几个关键点每条消息以data:开头后面跟内容以两个换行符\n\n结束。如果数据本身包含换行需要拆成多个data:行。可以自定义事件类型event: custom_event\ndata: xxx\n\n。可以设置重连间隔retry: 3000\n\n。我踩过的一个坑是服务端返回的每条消息如果没有正确以\n\n结尾前端EventSource的onmessage回调根本不会触发。数据会一直缓冲在浏览器里直到超时或者连接关闭。当时我调试了半个小时以为是跨域问题结果是少了一个换行符。2.3 FastAPI 中实现 SSE 的正确姿势在 FastAPI 中实现 SSE最直接的方式是使用StreamingResponsefrom fastapi import FastAPI from fastapi.responses import StreamingResponse import asyncio import json app FastAPI() async def event_generator(): tokens [你, 好, , 世, 界] for token in tokens: data json.dumps({token: token}, ensure_asciiFalse) yield fdata: {data}\n\n await asyncio.sleep(0.1) 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, } )这里有几个容易忽略的细节media_type必须是text/event-stream。虽然不设置浏览器也能收到数据但EventSource会拒绝连接。这个坑我在第一次实现时踩过浏览器控制台报了一个很模糊的错误排查了半天才发现是 content-type 不对。X-Accel-Buffering: no这个 header 非常关键。如果你用了 Nginx 做反向代理Nginx 默认会缓冲响应数据导致流式效果完全失效——用户看到的还是一次性返回。加上这个 header 可以告诉 Nginx 不要缓冲。当然Nginx 配置里也需要配合proxy_buffering off;。Cache-Control: no-cache防止中间层缓存 SSE 响应。虽然 SSE 响应通常不会被缓存但显式声明更安全。2.4 那个让人头疼的 idle timeout 问题如果你在生产环境部署过 SSE 服务大概率遇到过这个报错stream disconnected before completion: idle timeout waiting for sse这个问题的本质是SSE 连接在空闲一段时间后被中间层负载均衡、代理、网关强制断开了。不同的中间件默认超时时间不同有的是 30 秒有的是 60 秒有的是 120 秒。解决方案有两个方向方案一心跳保活。在数据流中定期插入注释行以:开头的行会被 SSE 客户端忽略async def event_generator_with_heartbeat(): while True: try: token await asyncio.wait_for(queue.get(), timeout15) yield fdata: {json.dumps(token)}\n\n except asyncio.TimeoutError: yield : heartbeat\n\n每 15 秒发送一次心跳确保连接不会因为空闲而被断开。这个间隔要根据你的中间件超时时间设置一般取超时时间的三分之一比较安全。方案二调整中间件超时配置。如果你能控制 Nginx 或负载均衡的配置直接把proxy_read_timeout调大。但生产环境中中间件往往不止一层每层都要改很容易遗漏。所以心跳保活是更可靠的方案不依赖外部配置。注意心跳间隔不能太短否则会产生大量无用请求也不能太长否则起不到保活作用。15 到 30 秒是比较合理的范围。3. LangChain 流式输出从 LLM 到结构化数据的完整链路3.1 LangChain 的流式回调机制LangChain 的流式输出依赖于回调系统Callback System。当你调用llm.stream()或者chain.stream()时LangChain 会在每个 token 生成时触发on_llm_new_token回调。最基础的用法from langchain_openai import ChatOpenAI llm ChatOpenAI(modelgpt-4o-mini, streamingTrue) for chunk in llm.stream(用一句话介绍 Python): print(chunk.content, end, flushTrue)这段代码会逐 token 打印结果。但实际项目中我们通常不会直接这样用而是把流式输出集成到 FastAPI 的 SSE 接口中async def langchain_stream(query: str): llm ChatOpenAI(modelgpt-4o-mini, streamingTrue) async for chunk in llm.astream(query): if chunk.content: yield fdata: {json.dumps({token: chunk.content}, ensure_asciiFalse)}\n\n yield data: [DONE]\n\n看起来很简单对吧但当你需要结构化输出时事情就变得复杂了。3.2 结构化输出的流式困境LangChain 的结构化输出Structured Output通常通过with_structured_output()实现from pydantic import BaseModel, Field class PersonInfo(BaseModel): name: str Field(description姓名) age: int Field(description年龄) skills: list[str] Field(description技能列表) structured_llm llm.with_structured_output(PersonInfo) result structured_llm.invoke(提取信息张三28岁会Python和Java) # result 是一个 PersonInfo 对象问题来了with_structured_output()返回的是一个完整的 Pydantic 对象不是流式的。你没法在 token 级别获取部分解析的 JSON。如果直接把这个结果通过 SSE 推送用户还是要等完整结果生成完毕才能看到内容打字机效果完全丧失。这个问题的根源在于结构化输出需要在 LLM 生成完整 JSON 后才能进行解析和验证。JSON 是一种全有或全无的格式——{name: 张这样的片段是无法被解析的。3.3 三种可行的流式结构化输出方案经过多次实践我总结了三种方案各有适用场景方案一流式输出原始 JSON前端做增量解析不让 LangChain 做结构化解析直接让 LLM 输出 JSON 字符串流式推送给前端前端负责拼接和解析。async def stream_json(query: str): prompt f请以 JSON 格式输出以下信息不要包含任何其他内容 查询{query} 格式{{name: 姓名, age: 年龄, skills: [技能1, 技能2]}} async for chunk in llm.astream(prompt): if chunk.content: yield fdata: {json.dumps({token: chunk.content}, ensure_asciiFalse)}\n\n yield data: [DONE]\n\n前端收到完整 JSON 后再解析。这种方案实现简单但前端需要处理JSON 还没生成完的中间状态。方案二使用JsonOutputParser的流式解析LangChain 提供了JsonOutputParser它支持部分 JSON 解析from langchain_core.output_parsers import JsonOutputParser parser JsonOutputParser(pydantic_objectPersonInfo) chain prompt | llm | parser async for chunk in chain.astream({query: 张三28岁会Python和Java}): # chunk 是部分解析的字典 yield fdata: {json.dumps(chunk, ensure_asciiFalse)}\n\nJsonOutputParser内部会在每个 token 到达时尝试解析能解析出多少就返回多少。比如{name: 张三会被解析为{name: 张三}后续字段逐步补全。这个方案的好处是前端拿到的始终是当前已解析的部分结构化数据不需要自己处理 JSON 拼接。方案三双通道输出——结构化数据 自然语言如果场景允许可以让 LLM 同时输出自然语言描述和结构化数据自然语言部分用于打字机效果结构化数据用于后续处理class OutputWithExplanation(BaseModel): explanation: str Field(description自然语言解释) data: PersonInfo Field(description结构化数据)但这种方案需要 LLM 支持并行输出且 prompt 设计要非常精确否则容易出错。3.4 方案选型我的实际经验方案实现复杂度前端复杂度用户体验适用场景原始 JSON 流式低中中简单场景前端可控JsonOutputParser中低好推荐方案平衡性好双通道输出高中最好对体验要求极高的场景我个人最推荐方案二。JsonOutputParser的流式解析能力被很多人低估了它能在 LLM 生成 JSON 的过程中逐步输出已解析的字段前端只需要监听字段变化并更新 UI 即可。而且 LangChain 的解析器会自动处理 markdown 代码块包裹比如 LLM 输出json ...的情况省去了很多手动清洗的工作。不过要注意JsonOutputParser在 JSON 格式错误时会抛出异常。如果你的 LLM 输出不稳定建议加上handle_parsing_errorsTruechain prompt | llm | parser.with_retry(stop_after_attempt2)或者用OutputFixingParser做自动修复。但修复本身会消耗额外 token所以更好的做法是在 prompt 中明确要求输出格式从源头减少格式错误。4. 前端接收与渲染EventSource 的坑与 JSON 增量解析4.1 EventSource 的基本用法和限制前端接收 SSE 最直接的方式是使用EventSourceconst eventSource new EventSource(/api/stream) eventSource.onmessage (event) { if (event.data [DONE]) { eventSource.close() return } const data JSON.parse(event.data) appendText(data.token) } eventSource.onerror (error) { console.error(SSE error:, error) eventSource.close() }但EventSource有几个硬伤不支持自定义请求头。你没法在请求中加Authorization头只能通过 URL 参数传 token安全性差。只支持 GET 请求。如果你的接口需要 POST 传复杂参数EventSource无能为力。自动重连可能带来问题。连接断开后浏览器会自动重连如果你的接口有副作用比如每次都触发 LLM 调用会导致重复请求。4.2 用 fetch ReadableStream 替代 EventSource因为EventSource的限制实际项目中我更推荐用fetchReadableStream手动处理 SSEasync function streamRequest(url, body) { const response await fetch(url, { method: POST, headers: { Content-Type: application/json, Authorization: Bearer ${token}, }, body: JSON.stringify(body), }) 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 try { const parsed JSON.parse(data) handleToken(parsed) } catch (e) { console.warn(Parse error:, e) } } } } }这段代码有几个关键点buffer的处理是核心。SSE 数据可能被 TCP 分片一次reader.read()拿到的可能是不完整的数据。必须用缓冲区累积按\n\n分割最后一段不完整的留到下次处理。我见过很多实现直接对每次read()的结果做split导致 JSON 解析频繁失败。decoder.decode(value, { stream: true })的stream: true参数很重要。UTF-8 编码的中文字符占 3 个字节可能被分片到两次read()中。不加这个参数会导致中文乱码。[DONE]标记用于优雅结束。虽然done为true时循环也会结束但显式发送结束标记可以让前端更早地关闭连接避免等待 TCP 超时。4.3 JSON 增量解析前端如何优雅处理半成品当使用JsonOutputParser方案时前端收到的是逐步补全的 JSON 对象。比如第一次收到{name: 张} 第二次收到{name: 张三, age: 28} 第三次收到{name: 张三, age: 28, skills: [Python]}前端需要做的是用最新数据覆盖旧数据而不是追加。这和服务端推送原始 token 时的处理逻辑完全不同。let currentData {} function handleStructuredChunk(chunk) { currentData { ...currentData, ...chunk } renderStructuredData(currentData) }但这里有个问题如果字段是数组直接覆盖会丢失之前的元素。比如skills从[Python]变成[Python, Java]覆盖是没问题的但如果解析器每次只返回新增的元素就需要追加而不是覆盖。实际使用中JsonOutputParser每次返回的是完整的已解析对象所以直接覆盖是安全的。但如果你自己实现了增量解析逻辑一定要确认解析器的行为。4.4 Vue 中的响应式渲染优化在 Vue 中流式渲染的性能问题容易被忽视。如果每个 token 都触发一次响应式更新高频更新会导致页面卡顿。我的做法是用requestAnimationFrame做节流import { ref, onUnmounted } from vue const displayText ref() let pendingText let rafId null function appendToken(token) { pendingText token if (!rafId) { rafId requestAnimationFrame(() { displayText.value pendingText pendingText rafId null }) } } onUnmounted(() { if (rafId) cancelAnimationFrame(rafId) })这样每帧最多更新一次 DOM即使 LLM 每秒生成 50 个 token页面也能保持流畅。实测下来不做节流的话在低端设备上滚动会明显卡顿。另外Vue 的v-text比{{ }}插值在大量文本更新时性能更好因为v-text直接设置textContent而插值会触发更多的虚拟 DOM diff。5. 完整链路实战从 FastAPI 到 Vue 的端到端实现5.1 服务端FastAPI LangChain 流式接口把前面的知识点串起来一个完整的服务端实现如下from fastapi import FastAPI from fastapi.responses import StreamingResponse from langchain_openai import ChatOpenAI from langchain_core.prompts import ChatPromptTemplate from langchain_core.output_parsers import JsonOutputParser from pydantic import BaseModel, Field import json import asyncio app FastAPI() class PersonInfo(BaseModel): name: str Field(description姓名) age: int Field(description年龄) skills: list[str] Field(description技能列表) parser JsonOutputParser(pydantic_objectPersonInfo) prompt ChatPromptTemplate.from_messages([ (system, 你是一个信息提取助手。请严格按照以下格式输出\n{format_instructions}), (human, {query}) ]).partial(format_instructionsparser.get_format_instructions()) llm ChatOpenAI(modelgpt-4o-mini, streamingTrue, temperature0) chain prompt | llm | parser app.post(/api/extract) async def extract(query: str): async def generate(): try: async for chunk in chain.astream({query: query}): yield fdata: {json.dumps(chunk, ensure_asciiFalse)}\n\n yield data: [DONE]\n\n except Exception as e: error_data json.dumps({error: str(e)}, ensure_asciiFalse) yield fdata: {error_data}\n\n return StreamingResponse( generate(), media_typetext/event-stream, headers{ Cache-Control: no-cache, Connection: keep-alive, X-Accel-Buffering: no, } )这个实现有几个值得注意的地方temperature0对结构化输出很重要。温度越高LLM 输出格式越不稳定JSON 解析失败的概率越大。结构化提取场景下创造性不是我们需要的。异常处理要返回 SSE 格式的错误。如果直接抛出 HTTP 异常前端可能收不到有意义的错误信息。把错误包装成 SSE 数据帧前端可以统一处理。parser.get_format_instructions()自动生成格式说明。这比手写 JSON schema 描述更可靠因为 LangChain 会根据 Pydantic 模型自动生成精确的格式要求。5.2 前端Vue 3 完整实现// useStreamExtract.js import { ref } from vue export function useStreamExtract() { const data ref({}) const loading ref(false) const error ref(null) async function extract(query) { loading.value true error.value null data.value {} try { const response await fetch(/api/extract, { method: POST, headers: { Content-Type: application/json }, body: JSON.stringify({ query }), }) if (!response.ok) { throw new Error(HTTP ${response.status}) } 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 parts buffer.split(\n\n) buffer parts.pop() for (const part of parts) { if (!part.startsWith(data: )) continue const payload part.slice(6) if (payload [DONE]) return try { const parsed JSON.parse(payload) if (parsed.error) { error.value parsed.error return } data.value { ...data.value, ...parsed } } catch (e) { console.warn(Chunk parse failed:, payload) } } } } catch (e) { error.value e.message } finally { loading.value false } } return { data, loading, error, extract } }在组件中使用template div classextract-panel input v-modelquery keyup.enterhandleExtract placeholder输入查询... / button clickhandleExtract :disabledloading {{ loading ? 提取中... : 开始提取 }} /button div v-iferror classerror{{ error }}/div div v-ifdata.name classresult p姓名{{ data.name }}/p p年龄{{ data.age }}/p p技能{{ data.skills?.join(、) }}/p /div /div /template script setup import { ref } from vue import { useStreamExtract } from ./useStreamExtract const query ref() const { data, loading, error, extract } useStreamExtract() function handleExtract() { if (!query.value.trim()) return extract(query.value) } /script5.3 实测中的意外情况与处理情况一LLM 输出 markdown 代码块包裹的 JSON有时候 LLM 会输出json {name: 张三, age: 28}JsonOutputParser 能自动处理这种情况但如果你用的是自定义解析逻辑需要先剥离代码块标记。我的做法是用正则预处理 python import re def strip_code_block(text: str) - str: match re.search(r(?:json)?\s*([\s\S]*?), text) return match.group(1).strip() if match else text情况二流式过程中连接断开用户网络不稳定时SSE 连接可能中途断开。前端需要检测这种情况并提示用户重试。我的做法是在reader.read()返回done: true但还没收到[DONE]标记时认为连接异常中断let receivedDone false // ... 在收到 [DONE] 时设置 receivedDone true // 循环结束后 if (!receivedDone) { error.value 连接中断请重试 }情况三并发请求导致的状态混乱如果用户快速连续点击提取按钮多个流式请求会同时进行后返回的数据可能覆盖先返回的。解决方案是用一个请求 ID 标记每次请求只处理最新请求的数据let currentRequestId 0 async function extract(query) { const requestId currentRequestId // ... 在每次处理数据前检查 if (requestId ! currentRequestId) return }这个细节在开发阶段很容易被忽略但上线后用户操作快一点就会暴露问题。6. 性能调优与生产环境注意事项6.1 减少首 token 延迟用户感知的快主要取决于首 token 到达时间TTFTTime To First Token而不是总耗时。优化 TTFT 的几个方向使用更小的模型。gpt-4o-mini 的 TTFT 通常比 gpt-4o 快 2 到 3 倍如果任务不复杂没必要用大模型。精简 prompt。prompt 越长模型处理时间越长。把不必要的示例和说明删掉。预热连接。如果使用云服务 API保持 HTTP 连接池可以减少 TLS 握手时间。6.2 SSE 连接数限制浏览器对同域名的 HTTP/1.1 连接数限制通常是 6 个。如果页面同时打开多个 SSE 连接会阻塞其他请求。解决方案使用 HTTP/2多路复用不受此限制。合并 SSE 连接用一个连接推送多种类型的数据。及时关闭不再使用的 SSE 连接。6.3 日志与监控流式接口的日志记录需要特别注意。不要在每个 token 到达时都打日志否则日志量会爆炸。我的做法是记录请求开始和结束时间计算 TTFT 和总耗时。记录 token 总数和平均生成速度。只在异常时记录详细内容。import time start time.time() first_token_time None token_count 0 async for chunk in chain.astream({query: query}): if first_token_time is None: first_token_time time.time() token_count 1 yield fdata: {json.dumps(chunk, ensure_asciiFalse)}\n\n # 请求结束后记录 logger.info(fTTFT: {first_token_time - start:.2f}s, fTotal: {time.time() - start:.2f}s, fTokens: {token_count})这些指标对于排查性能问题和容量规划非常有价值。我在生产环境中就是通过监控 TTFT 发现某个模型供应商在高峰期响应变慢及时切换了备用供应商。6.4 我踩过的最大的一个坑最后分享一个让我印象最深刻的坑。有一次上线后用户反馈打字机效果时有时无。排查后发现问题出在负载均衡的健康检查上。我们的 SSE 接口和普通接口共用同一个域名负载均衡器对 SSE 长连接也做健康检查导致连接被意外中断。解决方案是把 SSE 接口单独部署使用不同的域名或端口并调整负载均衡器的超时配置。这个问题的隐蔽性在于本地开发环境完全正常只有生产环境的多层网络架构才会触发。从那以后我在设计任何流式接口时都会先画一张网络拓扑图确认每一层的超时和缓冲配置。这个习惯帮我避免了很多类似的问题。