做 AI 应用做了快两年我最大的感受是能不能把一个流式对话做顺决定了用户打开你产品的第一印象。你辛辛苦苦调好了 Prompt结果前端一直在转圈用户等了三秒没有反应直接关页面了——这类问题十有八九不是模型不行而是你压根没把数据流的链路打通。最近在做项目的过程中我专门把 SSE 流式传输、LangChain 结构化输出、前端打字机效果这一整条链路重新梳理了一遍踩了不少坑也沉淀了一套可以直接抄走的方案。这篇文章就把这套东西完整拆开从协议原理到前后端代码再到那些文档里绝对不会告诉你的排查经验一次讲透。这篇实战内容适合谁后端同学想给 AI 接口加流式输出但还没想清楚消息格式前端同学被「怎么把流式数据渲染成打字机效果」卡住或者你已经在用 LangChain 但发现chain.stream()出来的东西和想象的不一样——那这篇文章就是给你准备的。我会尽量说人话复杂的地方用类比讲明白保证你看完能直接动手改代码。1. AI 对话为什么绕不开 SSE流式协议的本质与选型1.1 从轮询到长连接SSE 到底解决了什么问题传统 Web 应用里前端想知道后端有没有新数据最笨的办法是轮询——每隔一两秒发一次请求问好了吗好了吗。这个模式在老系统里很常见但放到大模型对话场景里就是灾难。LLM 生成一段 300 字的回答哪怕速度已经很快也要好几秒才能吐完。如果前端傻等全部生成完再一次性展示用户在这几秒里面对的就是一片空白既不知道请求到底有没有成功也不知道模型是不是卡住了。SSEServer-Sent Events服务器发送事件就是专门解决这个问题的。它让服务器把响应内容拆成一个个小片段通过一条 HTTP 长连接持续推送。前端建好连接之后不需要反复发请求只要坐在那里接收数据就行。用大白话说轮询是你每隔一会儿跑去问店家饭好了没SSE 是店家做好一道菜就端一道菜上来你坐那儿等着吃。SSE 的技术基础是 HTTP 协议本身不需要额外安装 WebSocket 之类的依赖而且它的消息格式极其简单每条消息用data:开头结尾是一个空行。比如data: 你好 data: 世界这样一个文本流前端收到后按空行切分就能拿到两条消息。SSE 还支持event:字段自定义事件类型、id:字段做断点续传机制虽然简单但足够支撑大部分实时推送场景。1.2 SSE 协议格式与 LangChain 流式输出的天然契合LangChain 的流式接口astream()返回的是一个异步迭代器模型每生成一个 Token或者一小段 Token这个迭代器就往外吐一次数据。这种一边生成一边吐的模式恰好和 SSE 的一端推一端收是天生一对。打个比方LLM 是个很能说的人SSE 是一条电话线。LangChain 负责让这个人不停嘴地往外说话SSE 负责把每一句话实时传到电话那头前端的打字机效果就是听众看着记录员一个字一个字地记下来。实际编码里你只需要在 FastAPI 的StreamingResponse里放一个异步生成器把 LangChain 吐出来的每一个 chunk 按 SSE 格式包装网络层就会把这些消息逐条推给前端。这里有一个关键点SSE 的心跳机制。协议规定如果服务器持续 15 秒以上没有发送任何数据连接就可能被中间的网络设备关闭。所以规范的 SSE 服务器即使没有业务数据也应该定期发送一个注释行:开头的一行作为心跳相当于对着电话线喂喂喂地确认连接还活着。1.3 SSE 与 WebSocket 的选型对比很多同学第一反应是流式传输那必然 WebSocket。实际做下来你会发现在 AI 对话这个场景里SSE 的优势比 WebSocket 更明显。我整理了一张对比表方便你根据场景选型维度SSEWebSocket传输方向服务器单向推送给客户端双向实时通信底层协议HTTP天然穿过网关和代理TCP需要协议升级消息格式纯文本data:前缀可读性极强二进制或文本需要自定义协议自动重连协议内置连接断开会自动重试需要自己实现重连逻辑携带参数依赖 URL/Query头部受限可以自定义请求头浏览器支持现代浏览器全支持全支持但需要额外的事件处理单看最后两行你可能觉得 WebSocket 更强但 AI 对话这个场景里数据传输方向就是单向的——服务器把模型生成的 Token 推给前端前端基本不需要反向发送数据请求参数已经在 POST 里带过去了。用 WebSocket 等于杀鸡用牛刀还要额外操心二进制分帧、粘包、重连这些破事。SSE 自带断线重传机制浏览器 EventSource 对象甚至能配置重试间隔这对我来说已经足够香了。唯一的硬伤是 EventSource 只能发 GET 请求、不能自定义 Header——所以我后面实战部分直接绕开了它用 fetch 流式读取来拿 SSE这样既保留 POST 传参灵活性又能按需处理各个事件。2. LangChain 流式接口逐层拆解stream、astream 与消息结构2.1 Runnable 协议的四种调用方式LangChain 从 0.1.x 开始主推 Runnable 协议所有 Chain 组件都实现了统一的调用接口。和咱这篇文章相关的就是下面四种invoke()同步调用输入一个字典输出最终结果。适合调试和脚本场景。ainvoke()异步调用还用 await 接。适合在 FastAPI 请求处理函数里用。stream()同步流式返回一个迭代器每次吐一块中间结果。astream()异步流式返回异步迭代器配合async for使用。后台接口推荐用这个。关键是理解invoke和stream的区别不是快和慢而是是否暴露中间过程。invoke把整条链路当成一个黑盒你只能拿到最终输出stream把链路里的每一步都摊开给你看模型每生成一个 Token迭代器就吐一次。from langchain_openai import ChatOpenAI from langchain_core.prompts import ChatPromptTemplate from langchain_core.output_parsers import StrOutputParser llm ChatOpenAI( modelgpt-4o-mini, temperature0.7, streamingTrue, # 强烈建议在生产环境按实际模型服务商配置 base_url、api_key ) prompt ChatPromptTemplate.from_messages([ (system, 你是一个简洁的中文写作助手回答控制在300字以内。), (human, {question}), ]) chain prompt | llm | StrOutputParser() # 同步逐 Token 输出 for chunk in chain.stream({question: 用一句话解释什么是SSE}): print(chunk, end, flushTrue)跑一下这段代码你会看到控制台像打字机一样逐字输出。那个streamingTrue很关键——它告诉 OpenAI SDK 开启流式模式服务端用 chunked 编码逐个发 TokenLangChain 才能逐块往外吐。如果漏掉这个参数chain.stream()依然会一次性把结果给你所谓流式就名存实亡了。2.2 流式链里真正会吐出来的数据类型刚开始做流式接入的同学往往会犯一个错误以为astream()每次吐出来的都是纯字符串。实际上不一定。这取决于你的 Chain 最后挂的是什么组件。如果 Chain 结构是prompt | llm | StrOutputParser()那么最后一步StrOutputParser会把 AIMessageChunk 的内容转成字符串所以astream()吐出来的是字符串直接拼起来就是最终文本。但如果你的 Chain 结构是prompt | llm没有 StrOutputParser那么astream()吐出来的是AIMessageChunk对象——它里面有content属性还可能有tool_calls如果模型调用了工具。这时候你要么自己加一个(msg) - msg.content的魔法方法把那层壳剥掉要么在业务代码里用hasattr(chunk, content)判断一下再决定怎么取文本。顺着这个思路往下走你还会发现结构化的输出解析器比如PydanticOutputParser和流式组合时有个特性解析器本身不是流式的。它必须等模型把整段 JSON 吐完才能一次性解析出结构化对象。这意味着整个 Chain 的最后一段数据是在憋大招它会先攒着然后突然蹦出一个大对象。我用一个简单 Chain 验证过prompt | llm | PydanticOutputParser走astream()前面模型吐了十几个 token 数据解析器那边一个都不会冒出来最后一次性输出最终结果。做前端的同学如果发现流式到一半就停了然后又一下子出现完整内容问题十有八九出在这里。2.3 非 Token 数据的处理元数据与工具调用片段另外一个容易被忽略的环节是现代 LLM 的流式响应里不一定只有文本 Token。如果你开启了函数调用Function Calling / Tool Calling模型可能先吐出一段tool_calls的增量参数再吐出文本如果你在 LangChain 里挂了回调处理器callbacks流式过程中还会穿插各种事件的回调。我在实际项目里处理过一种脏数据模型同时返回content和tool_calls的 chunk前端只关心content但如果不加过滤这些tool_calls的字符串会被直接拼进正文里用户看到的就是满屏乱码结构。解决方案是做一个统一的流式消息格式化层把不同类型的输出拆成不同的 SSE 事件token事件纯文本增量直接拼到正文tool事件工具调用信息前端可以展示正在检索资料之类的状态meta事件最终的结构化数据比如 JSON 提取结果、意图分类、引用列表。这么设计的好处是前端只负责展示后端告诉它这块是正文这块是状态这块是 JSON两边职责清楚调试也方便。后面第 4 节我会给出完整代码。3. FastAPI LangChain 流水线实战后端 SSE 封装与心跳机制3.1 用 StreamingResponse 正确构建 SSE 响应FastAPI 可以说是和 LangChain 流式配合得最顺的 Python 框架。它的StreamingResponse接收一个可迭代对象然后通过 HTTP 分块传输把数据一块一块发出去。我们要做的就是把 LangChain 的异步生成器包装成 SSE 格式。核心代码如下我尽量把注释写清楚import asyncio import json import time from fastapi import FastAPI from fastapi.responses import StreamingResponse from pydantic import BaseModel from langchain_openai import ChatOpenAI from langchain_core.prompts import ChatPromptTemplate from langchain_core.output_parsers import StrOutputParser app FastAPI() llm ChatOpenAI( modelgpt-4o-mini, temperature0.7, streamingTrue, ) prompt ChatPromptTemplate.from_messages([ (system, 你是专业的中文助手。回答要结构清晰、语言自然。), (human, {question}), ]) # 这个 chain 只负责流式文本 chat_chain prompt | llm | StrOutputParser() def sse_format(event: str, payload: dict) - str: 把事件和数据包装成 SSE 协议的一帧 return fevent: {event}\ndata: {json.dumps(payload, ensure_asciiFalse)}\n\n async def sse_generator(question: str): 生成器把 LangChain 的流式输出转成 SSE 帧 # 1. 先发一个多行注释作为心跳顺便告诉前端“我开始了” yield : connected\n\n yield sse_format(meta, {type: start, timestamp: time.time()}) try: # 2. 核心循环逐 token 包装 async for chunk in chat_chain.astream({question: question}): # 兼容 AIMessageChunk 的情况 if isinstance(chunk, str): content chunk elif hasattr(chunk, content): content chunk.content or else: content str(chunk) # 有真实内容才发 token 事件空白跳过 if content.strip(): yield sse_format(token, {content: content}) # 3. 结束标记 yield sse_format(end, {type: done}) except Exception as exc: # 生成器内捕获异常否则客户端只会看到连接中断什么都拿不到 yield sse_format(error, {message: str(exc)}) class ChatRequest(BaseModel): question: str app.post(/chat) async def chat_endpoint(req: ChatRequest): return StreamingResponse( sse_generator(req.question), media_typetext/event-stream, headers{ Cache-Control: no-cache, Connection: keep-alive, X-Accel-Buffering: no, # 告诉 Nginx 别缓冲 }, )有几个细节我要特别强调这些是我实际运行时踩出来的坑第一media_type必须是text/event-stream。这是浏览器识别 SSE 响应的硬条件。如果你漏掉了前端即使收到了数据EventSource 也不会正确解析。用 fetch 流式读取时这个字段依然重要因为有些代理服务器会根据 Content-Type 决定要不要缓冲。第二X-Accel-Buffering: no这个响应头不能省。当你把服务部署到 Nginx 后面时Nginx 默认会缓存 FastAPI 发出来的所有数据攒够一定量再一次性发给客户端。那打字机效果就废了——用户看到的还是转圈半天猛一下全出来。加了这个头就是告诉 Nginx这条响应你不要碰来一个帧发一个帧。第三生成器里必须 try-except。模型生成过程中抛异常是常有的事上下文超限、内容过滤、网络抖动。不 catch 的话FastAPI 会把连接直接断开前端拿到的是半截消息什么错误信息都没有。把异常包装成 SSE 的error事件发给前端至少用户能知道出错了不至于以为是网络挂了。3.2 同步生成器和异步生成器前端卡顿的隐患如果你的业务代码里拿的是同步的chain.stream()在 FastAPI 里有两种用法把同步生成器传给StreamingResponseFastAPI 会在线程池里跑它不会阻塞主事件循环但要注意同步生成器里如果调用了阻塞的 I/O 或 CPU 密集操作还是会占住线程池资源。我踩过的一个坑是这样的一开始我图省事在async def端点里直接用chain.stream()同步方法循环结果事件循环被阻塞前端同时发起的第二个请求要等第一个跑完才开始响应。这个问题特别隐蔽因为单个请求测试时根本感觉不出来一旦并发量上来接口响应曲线直接爆炸。正确姿势是用astream()配合async for或者在生成器函数前面加一个线程池装饰器/asyncio.to_thread包装。而且我建议把生成器本身也设计成异步的让 LangChain 的异步流式能力真正发挥出来。对于大多数生产环境来说这就是你要的代码。3.3 心跳机制与空闲超时问题先给你复现一个真实场景。我用 LangChain 接了个长思考链路的模型内部套了好几个子 Agent有一次用户问了一个复杂计算题模型在内部工具推理阶段沉默了 40 秒没吐任何 Token。连接中断前端报了一个让我们排查了很久的错误——消息流在完成前就断开了像极了超时等待 SSE 数据。这个问题的根因是空闲超时idle timeout。网络设备、Nginx、Gunicorn 都以这条连接最近有没有数据流动作为判断依据。模型 40 秒没吐数据Nginx 觉得这连接废了直接掐断。解决思路有两个层次治本在生成器里加心跳。每隔一段时间我一般设 15 秒主动发送一个 SSE 注释帧: heartbeat\n\n让连接一直有数据流动网络设备就不会误判。治标调大 Nginx 和 Gunicorn 的超时时间比如proxy_read_timeout 300s、Gunicorn--timeout 300。但这只是给服务器更多的耐心如果模型真的长时间无响应该断还是会断。心跳帧用:开头是 SSE 规范规定的注释格式接收方看到会无视它不会当成业务数据。所以它可以安全地混在真实数据流中间。async def sse_generator_with_heartbeat(question: str): last_heartbeat time.time() async for chunk in chat_chain.astream({question: question}): content chunk if isinstance(chunk, str) else getattr(chunk, content, ) if content.strip(): yield sse_format(token, {content: content}) last_heartbeat time.time() elif time.time() - last_heartbeat 15: yield : heartbeat\n\n last_heartbeat time.time()这样即使模型长时间思考连接也不会因为空闲而中断。4. 结构化输出与 JSON 解析让模型不只“说话”还能“填表单”4.1 LangChain 结构化输出的三种主流方式流式文本解决的是用户看得到响应的问题但 AI 应用落地时还面临另一个问题模型输出的是自然语言系统怎么才能拿到结构化的数据去做后续处理比如从简历里提取姓名、工作年限、技能列表或者让 Agent 返回一个可执行的 JSON 指令。LangChain 提供三种主流方式我一一说清楚它们各自的适用场景方式一Pydantic 输出解析器PydanticOutputParser这是最经典的方式。你先定义一个 Pydantic 模型就是数据结构LangChain 会把它的 schema 描述自动塞进 Prompt 的format_instructions部分要求模型按这个格式输出 JSON。模型输出后解析器把 JSON 转成 Pydantic 对象。整个过程不依赖特定模型的函数调用能力适用面广。方式二with_structured_output()方法这是 ChatOpenAI 等模型封装类上的推荐方法。底层原理是让模型走 Function Calling / 原生 JSON 模式把 Pydantic schema 转成工具的 JSON Schema模型直接以工具参数的形式返回结构化数据。这种方式在 OpenAI 系模型上准确率极高因为模型天生被训练过调用函数时严格按 schema 输出。方式三自定义response_format参数如果你用的是 OpenAI 兼容接口可以直接在ChatOpenAI里传model_kwargs{response_format: {type: json_object}}强制模型输出 JSON。这个方式的约束力不如前两种适合场景极简单的需求。我用一张表对比它们的差异方式约束力依赖模型能力适用场景PydanticOutputParser中等靠 Prompt 引导所有模型通用场景兼容各类模型with_structured_output强走工具调用协议OpenAI 及支持工具的模型需要高准确率解析response_format弱只强制 JSON 格式OpenAI 兼容接口简单、无嵌套结构的输出4.2 用 Pydantic 定义输出 Schema 并接入链路我实际项目里最常用的组合是方式二 Pydantic。下面是一个从面试评价文本中提取结构化信息的例子from langchain_core.pydantic_v1 import BaseModel, Field class InterviewEvaluation(BaseModel): 面试评价结构化提取 candidate_name: str Field(description候选人姓名) overall_score: int Field(description综合评分0到100分) strengths: list[str] Field(description候选人的优点列出2到4条) risks: list[str] Field(description候选人的风险点列出1到3条) hiring_verdict: str Field(description录用建议只能是hire/no_hire/on_hold之一) # 直接使用 with_structured_output把 Pydantic 模型转成 OpenAI 工具协议 structured_llm llm.with_structured_output(InterviewEvaluation) result structured_llm.invoke( 张三的编程能力很强算法题全过但沟通表达存在明显欠缺 ) print(result.candidate_name) # 张三 print(result.overall_score) # 某个整数 print(result.hiring_verdict) # 某个枚举值这里要解释两个细节。第一Pydantic 模型里的Field(description...)不是写给人看的注释LangChain 会把 description 拼进给模型的工具 schema模型靠它理解字段含义所以描述写得越具体输出正确率越高。第二枚举值hire/no_hire/on_hold直接用字段值和字符串描述比单纯让模型自由发挥可靠得多。4.3 流式文本和结构化输出如何共存项目做到一半你会发现一个很现实的需求既要让用户看到逐字生成的回复又要让后端拿到一个结构化的 JSON 做业务逻辑。这两个需求能不能同时满足能但需要设计。我的做法是在一条 SSE 流里同时发两种事件。token事件走流式文本让前端做打字机效果meta事件在文本流结束后发送结构化结果。示例代码如下router.post(/chat_interview) async def chat_interview(req: InterviewRequest): async def gen(): # 先流式输出评价文本 async for token in interview_chain.astream({question: req.question}): yield sse_format(token, {content: token}) # 再调用结构化模型提取评价要点 eval_result await structured_llm.ainvoke(req.question) yield sse_format(meta, { type: structured, data: eval_result.dict(), }) yield sse_format(end, {type: done}) return StreamingResponse(gen(), media_typetext/event-stream)前端拿到meta事件后可以把它渲染成侧边栏的智能摘要卡片或者直接塞进表单。这样用户看到的是自然流畅的对话系统拿到的是可入库的结构化数据一举两得。不过要提醒你这种流式文本 尾部结构化的模式有一个小问题结构化结果要等模型完整生成才能出所以如果是长文本前端可能要等好几秒才看到侧边栏更新。如果对实时性要求更高你可以考虑让结构化模型和流式模型并行跑两个 LLM 调用同时发出去结构化结果先到就先发文本流自然也不会受影响。代价是成本翻倍。大多数场景下先文本后结构化已经够用没必要提前优化。4.4 JSON 解析的兜底容错方案理论上with_structured_output已经保证了模型按 schema 返回 JSON但现实世界永远有意外。比如你接入了一个第三方模型它照样走 OpenAI 兼容接口但偶尔会把 JSON 包在 markdown 代码块里或者输出到一半被截断。这种时候如果你直接json.loads()就是一个JSONDecodeError等着你。我的兜底解析流程一共四层按顺序尝试大量减少了线上报错import json import re def robust_json_parse(text: str): 四层兜底解析模型输出的 JSON # 第一层去掉 markdown 代码块包裹 if text.startswith(): text re.sub(r^(?:json)?\s*|\s*$, , text, flagsre.MULTILINE).strip() # 第二层标准解析 try: return json.loads(text) except json.JSONDecodeError: pass # 第三层截取最外层大括号忽略前后杂音 match re.search(r\{.*\}, text, re.DOTALL) if match: try: return json.loads(match.group(0)) except json.JSONDecodeError: pass # 第四层用 json_repair 做宽松修复处理缺失引号、末尾逗号等 try: from json_repair import repair_json repaired repair_json(text) return json.loads(repaired) except Exception: return None这里有个容易踩的坑第三层的正则\{.*\}用的是贪婪模式如果 JSON 里有嵌套的对象它会一直匹配到最后一个}正好是完整 JSON 的外层所以能正确提取。但如果模型输出里同时有两个 JSON 对象这个正则也会把它们拼在一起导致解析失败。所以这层只是兜底不能作为主路径。还有一个思路很多人会忽略利用错误反馈让模型自我修正。当解析失败时把异常信息和提取到的残缺 JSON 一起塞回 Prompt让模型在不重新生成全部内容的前提下只补全出错的部分。我在带错误重试机制的场景里测试过修复成功率能到 80% 以上。5. 前端打字机效果实现Vue fetch 流式读取全流程5.1 为什么最终选了 fetch 而不是 EventSource浏览器自带 SSE 客户端就是EventSource它开箱即用、自动重连很多教程都推荐它。但我在项目里实际对接 LangChain 的后端接口时踩了三个让它不那么好用的坑EventSource 只能发 GET 请求可我需要在请求体里带对话上下文、用户配置、历史消息等一堆 JSON硬塞到 URL 里既丑陋又容易超长。EventSource 不能自定义请求头。我想在 Authorization 头里带用户的登录凭证EventSource 就是加不上去只能退而求其次用 Cookie 或 URL 参数既不安全又别扭。EventSource 对事件处理的粒度不够。它是面向接收推送设计的对于token、meta、end这种需要区分处理的自定义事件你得写一堆 event listener 去注册不如手写 fetch 解析来得直观。所以我最终采用的方案是用 fetch 的 POST 请求去打 SSE 接口然后通过response.body.getReader()读取流式数据自己解析 SSE 帧。看起来多写了几行代码但换来的是完全可控的请求参数和响应处理。5.2 ReadableStream 解析 SSE 帧核心代码逐段说明前端解析的关键在于SSE 数据不是一次到位的而是一块一块地来。网络字节流可能把一条消息拆成两半也可能把多条消息粘在一起。所以解析器必须维护一个buffer每收到一块数据就追加进 buffer然后按\n\n空行切分完整帧剩余的半截留到下一轮继续拼。一段可用的 Vue 3 组合式函数代码如下export function useSSE() { const fullText ref() const isStreaming ref(false) const structuredData ref(null) async function sendMessage(question) { fullText.value structuredData.value null isStreaming.value true const response await fetch(/api/chat, { method: POST, headers: { Content-Type: application/json }, body: JSON.stringify({ question }) }) if (!response.ok) { throw new Error(请求失败: ${response.status}) } const reader response.body.getReader() const decoder new TextDecoder(utf-8) let buffer while (true) { const { done, value } await reader.read() if (done) break // 把 Uint8Array 解码成字符串注意 stream: true 处理多字节字符 buffer decoder.decode(value, { stream: true }) // 按空行切分 SSE 帧最后一个可能是半截帧先放回 buffer const frames buffer.split(\n\n) buffer frames.pop() for (const frame of frames) { const parsed parseSSEFrame(frame) if (parsed.event token) { fullText.value parsed.data.content } else if (parsed.event meta) { structuredData.value parsed.data } else if (parsed.event error) { console.error(服务端错误:, parsed.data.message) } } } isStreaming.value false } return { fullText, structuredData, isStreaming, sendMessage } } function parseSSEFrame(frame) { const lines frame.split(\n) let event message let data for (const line of lines) { if (line.startsWith(event:)) { event line.slice(6).trim() } else if (line.startsWith(data:)) { // SSE 允许多行 data拼一个换行再 trim 保持 JSON 可解析 data line.slice(5) \n } } // 去掉末尾多余的换行位再解析 JSON const trimmed data.replace(/\n$/, ) return { event, data: trimmed ? JSON.parse(trimmed) : {} } }这段代码里我特别想提醒一个细节TextDecoder的第二个参数{ stream: true }千万别漏。UTF-8 的中文字符在二进制流里可能横跨两个 chunk比如第一个 chunk 来了 2 个字节第二个 chunk 才来剩余的 1 个字节。如果你在每一轮都单独decode(value)中文就会在中间断开变成乱码。{ stream: true }会告诉解码器数据没完我帮你缓存到完整字符再输出这是中文流式场景下的必备参数。5.3 打字机效果与 Markdown 实时渲染的协调拿到流式文本后最朴素的打字机效果就是把fullText直接绑定到某个 DOM 节点。但如果你在文本里混了 Markdown 语法直接v-html绑定原始文本会暴露一堆##、**符号体验极差。正确做法是每一次 token 增量后重新跑一次 Markdown 渲染用渲染结果替换 DOM 内容。这时候会有新的问题重新渲染会让光标跳动或者滚动位置不稳定。原因很简单Markdown 解析器输出的是完整的 HTML 字符串替换 innerHTML 会重置 DOM 结构导致用户阅读位置跳到页面顶部。解决思路是用nextTick等 DOM 更新完成后手动把滚动容器 scrollTop 设置到 scrollHeight如果渲染量大考虑做节流渲染比如每秒最多重新渲染两次让打字效果不至于因为频繁解析 markdown 而卡顿如果文本较长且包含表格、代码块这类复杂 Markdown建议前端用marked.js加highlight.js的组合解析性能比直接套 Vue 的v-html强很多。在这个环节你要注意异步流式状态下不要直接用computed对fullText做marked.parse()然后绑到 v-html 上。因为每次fullText更新computed 都会同步重新计算一旦文本量大了UI 线程会被 Markdown 解析阻塞打字机效果直接卡成 PPT。我的处理方案是把marked.parse放到watch的回调里或者干脆在 Message 对象上用requestIdleCallback做延迟解析把渲染压力分散到浏览器空闲时段。6. 线上常见问题排查实录SSE 断连、超时与解析失败6.1 stream disconnected before completion 的根因排查这是我在评论区里被问得最多的一个错误。完整报错是类似stream disconnected before completion: idle timeout waiting for sse。我排查这类问题有一套固定流程按顺序操作能快速定位先看是不是空闲超时。打开浏览器开发者工具看 Network 面板里这条 SSE 请求的响应时间线。如果请求在某个平静期突然结束曲线变平很久后断开基本可以锁定是中间设备在超时。直接在后端生成器里加心跳帧参考 3.3 节。再看代理层。把服务直连和走 Nginx 各测一次。如果直连正常、走 Nginx 断连那九成是 Nginx 对长连接的 read timeout 设置太短。用我下面的配置location /api/ { proxy_pass http://backend; proxy_http_version 1.1; proxy_set_header Connection ; proxy_buffering off; proxy_cache off; proxy_read_timeout 300s; }注意proxy_http_version 1.1很关键HTTP/1.0 不支持分块传输编码SSE 会直接失效。最后看应用层超时配置。如果你用 Gunicorn 跑 FastAPI默认 worker 超时是 30 秒模型生成超过 30 秒 Gunicorn 会强制杀 worker。启动命令里加--timeout 300。6.2 常见问题速查表下面这张表是我在实际项目中整理出来的高频问题按现象—原因—解决方案列出来排查时直接对照现象根因解决方案前端转圈很久最后一次性显示全部内容Nginx 缓冲导致数据攒批发送加X-Accel-Buffering: no响应头proxy_buffering off对话进行中突然断连前端无错误提示空闲超时idle timeout生成器内加: heartbeat心跳帧流式接口同时并发请求时变慢同步chain.stream()阻塞事件循环改用chain.astream()配合异步生成器前端收到中文乱码TextDecoder没有设置{ stream: true }按 5.2 节代码加上JSON.parse报错模型输出被 markdown 包裹或截断用 4.4 节的四层兜底解析网页开着多个会话时某些流式连接不定时断开浏览器对同一域名连接数限制收紧页面数量或用 HTTP/2LangChain 流式中间有长停顿模型内部在等待工具调用心跳帧 前端展示正在处理状态提示6.3 一个让我印象深刻的坑EventSource 的隐式缓冲陷阱最后分享一个非常隐蔽的坑。某个页面我一开始用的 EventSource 实现打字机效果单条消息没有任何问题。但在一个后台管理页里同时开了 5 个 EventSource监听不同 schema 的推送结果其中两个连接经常悄悄断掉怎么查都查不出原因。后来才反应过来HTTP/1.1 协议中浏览器对同一个域名的并发连接数限制是 6。页面里其他资源请求占了一部分再开多个 EventSource连超了之后浏览器就会挂起连接。加上我用的是 Docker 部署的 FastAPI服务端自身没有维护长连接池压力一大连接就有可能被系统层面回收。解决方案有两个方向要么改用 HTTP/2多路复用让连接数限制不再成为瓶颈要么像我后来做的把多个 SSE 订阅合并成一个接口用多 event 类型区分业务。对大部分中小项目来说后者实现成本更低效果也很直接。写在最后也是我踩过坑之后最想告诉你的把这套方案完整落地之后我回头看的最大的体会是流式传输这件事难的不是协议本身而是链路中每一层的配合。LangChain 输出 Token、FastAPI 包装 SSE、Nginx 不捣乱、前端增量渲染——任何一环想当然最后的体验就会打折扣。我自己最开始就是被流式接口能跑通这个表象骗了前端拿到 token 就直接渲染结果 JSON 全被拼在正文里、Nginx 缓冲把打字机效果吞成了整块输出、结构化解析失败直接让 Agent 中间状态崩掉。所以建议各位动手之前先画一张自己项目的 SSE 链路图标清楚每一环负责什么、可能在哪里出问题再动手写代码会少走很多弯路。另外一个后来觉得非常值的工作是把 SSE 的消息格式从服务端想发啥就发啥收敛成了事件类型 JSON 数据的规范。只要消息流动的地方都遵循这个规范前端就能精准地处理 token、meta、end、error 四类事件后端加新功能比如加个中间状态推送也不需要前端跟着改逻辑。如果你打算长期迭代一个 AI 产品我强烈建议你一开始就把这套消息约定定下来。最后再分享一个日常调试的实用小技巧先用curl测后端再连前端。curl -N http://localhost:8000/chat -d {question:你好}-N参数会禁用缓冲让你直接在终端看到每一个 SSE 帧的到达时间。如果终端里能逐帧刷出来后端基本没问题再去排查前端如果终端也是一次性输出问题就在后端或网络层。这个习惯帮我省了至少一半的联调时间。
SSE与LangChain流式输出实战：AI对话打字机效果全链路指南
做 AI 应用做了快两年我最大的感受是能不能把一个流式对话做顺决定了用户打开你产品的第一印象。你辛辛苦苦调好了 Prompt结果前端一直在转圈用户等了三秒没有反应直接关页面了——这类问题十有八九不是模型不行而是你压根没把数据流的链路打通。最近在做项目的过程中我专门把 SSE 流式传输、LangChain 结构化输出、前端打字机效果这一整条链路重新梳理了一遍踩了不少坑也沉淀了一套可以直接抄走的方案。这篇文章就把这套东西完整拆开从协议原理到前后端代码再到那些文档里绝对不会告诉你的排查经验一次讲透。这篇实战内容适合谁后端同学想给 AI 接口加流式输出但还没想清楚消息格式前端同学被「怎么把流式数据渲染成打字机效果」卡住或者你已经在用 LangChain 但发现chain.stream()出来的东西和想象的不一样——那这篇文章就是给你准备的。我会尽量说人话复杂的地方用类比讲明白保证你看完能直接动手改代码。1. AI 对话为什么绕不开 SSE流式协议的本质与选型1.1 从轮询到长连接SSE 到底解决了什么问题传统 Web 应用里前端想知道后端有没有新数据最笨的办法是轮询——每隔一两秒发一次请求问好了吗好了吗。这个模式在老系统里很常见但放到大模型对话场景里就是灾难。LLM 生成一段 300 字的回答哪怕速度已经很快也要好几秒才能吐完。如果前端傻等全部生成完再一次性展示用户在这几秒里面对的就是一片空白既不知道请求到底有没有成功也不知道模型是不是卡住了。SSEServer-Sent Events服务器发送事件就是专门解决这个问题的。它让服务器把响应内容拆成一个个小片段通过一条 HTTP 长连接持续推送。前端建好连接之后不需要反复发请求只要坐在那里接收数据就行。用大白话说轮询是你每隔一会儿跑去问店家饭好了没SSE 是店家做好一道菜就端一道菜上来你坐那儿等着吃。SSE 的技术基础是 HTTP 协议本身不需要额外安装 WebSocket 之类的依赖而且它的消息格式极其简单每条消息用data:开头结尾是一个空行。比如data: 你好 data: 世界这样一个文本流前端收到后按空行切分就能拿到两条消息。SSE 还支持event:字段自定义事件类型、id:字段做断点续传机制虽然简单但足够支撑大部分实时推送场景。1.2 SSE 协议格式与 LangChain 流式输出的天然契合LangChain 的流式接口astream()返回的是一个异步迭代器模型每生成一个 Token或者一小段 Token这个迭代器就往外吐一次数据。这种一边生成一边吐的模式恰好和 SSE 的一端推一端收是天生一对。打个比方LLM 是个很能说的人SSE 是一条电话线。LangChain 负责让这个人不停嘴地往外说话SSE 负责把每一句话实时传到电话那头前端的打字机效果就是听众看着记录员一个字一个字地记下来。实际编码里你只需要在 FastAPI 的StreamingResponse里放一个异步生成器把 LangChain 吐出来的每一个 chunk 按 SSE 格式包装网络层就会把这些消息逐条推给前端。这里有一个关键点SSE 的心跳机制。协议规定如果服务器持续 15 秒以上没有发送任何数据连接就可能被中间的网络设备关闭。所以规范的 SSE 服务器即使没有业务数据也应该定期发送一个注释行:开头的一行作为心跳相当于对着电话线喂喂喂地确认连接还活着。1.3 SSE 与 WebSocket 的选型对比很多同学第一反应是流式传输那必然 WebSocket。实际做下来你会发现在 AI 对话这个场景里SSE 的优势比 WebSocket 更明显。我整理了一张对比表方便你根据场景选型维度SSEWebSocket传输方向服务器单向推送给客户端双向实时通信底层协议HTTP天然穿过网关和代理TCP需要协议升级消息格式纯文本data:前缀可读性极强二进制或文本需要自定义协议自动重连协议内置连接断开会自动重试需要自己实现重连逻辑携带参数依赖 URL/Query头部受限可以自定义请求头浏览器支持现代浏览器全支持全支持但需要额外的事件处理单看最后两行你可能觉得 WebSocket 更强但 AI 对话这个场景里数据传输方向就是单向的——服务器把模型生成的 Token 推给前端前端基本不需要反向发送数据请求参数已经在 POST 里带过去了。用 WebSocket 等于杀鸡用牛刀还要额外操心二进制分帧、粘包、重连这些破事。SSE 自带断线重传机制浏览器 EventSource 对象甚至能配置重试间隔这对我来说已经足够香了。唯一的硬伤是 EventSource 只能发 GET 请求、不能自定义 Header——所以我后面实战部分直接绕开了它用 fetch 流式读取来拿 SSE这样既保留 POST 传参灵活性又能按需处理各个事件。2. LangChain 流式接口逐层拆解stream、astream 与消息结构2.1 Runnable 协议的四种调用方式LangChain 从 0.1.x 开始主推 Runnable 协议所有 Chain 组件都实现了统一的调用接口。和咱这篇文章相关的就是下面四种invoke()同步调用输入一个字典输出最终结果。适合调试和脚本场景。ainvoke()异步调用还用 await 接。适合在 FastAPI 请求处理函数里用。stream()同步流式返回一个迭代器每次吐一块中间结果。astream()异步流式返回异步迭代器配合async for使用。后台接口推荐用这个。关键是理解invoke和stream的区别不是快和慢而是是否暴露中间过程。invoke把整条链路当成一个黑盒你只能拿到最终输出stream把链路里的每一步都摊开给你看模型每生成一个 Token迭代器就吐一次。from langchain_openai import ChatOpenAI from langchain_core.prompts import ChatPromptTemplate from langchain_core.output_parsers import StrOutputParser llm ChatOpenAI( modelgpt-4o-mini, temperature0.7, streamingTrue, # 强烈建议在生产环境按实际模型服务商配置 base_url、api_key ) prompt ChatPromptTemplate.from_messages([ (system, 你是一个简洁的中文写作助手回答控制在300字以内。), (human, {question}), ]) chain prompt | llm | StrOutputParser() # 同步逐 Token 输出 for chunk in chain.stream({question: 用一句话解释什么是SSE}): print(chunk, end, flushTrue)跑一下这段代码你会看到控制台像打字机一样逐字输出。那个streamingTrue很关键——它告诉 OpenAI SDK 开启流式模式服务端用 chunked 编码逐个发 TokenLangChain 才能逐块往外吐。如果漏掉这个参数chain.stream()依然会一次性把结果给你所谓流式就名存实亡了。2.2 流式链里真正会吐出来的数据类型刚开始做流式接入的同学往往会犯一个错误以为astream()每次吐出来的都是纯字符串。实际上不一定。这取决于你的 Chain 最后挂的是什么组件。如果 Chain 结构是prompt | llm | StrOutputParser()那么最后一步StrOutputParser会把 AIMessageChunk 的内容转成字符串所以astream()吐出来的是字符串直接拼起来就是最终文本。但如果你的 Chain 结构是prompt | llm没有 StrOutputParser那么astream()吐出来的是AIMessageChunk对象——它里面有content属性还可能有tool_calls如果模型调用了工具。这时候你要么自己加一个(msg) - msg.content的魔法方法把那层壳剥掉要么在业务代码里用hasattr(chunk, content)判断一下再决定怎么取文本。顺着这个思路往下走你还会发现结构化的输出解析器比如PydanticOutputParser和流式组合时有个特性解析器本身不是流式的。它必须等模型把整段 JSON 吐完才能一次性解析出结构化对象。这意味着整个 Chain 的最后一段数据是在憋大招它会先攒着然后突然蹦出一个大对象。我用一个简单 Chain 验证过prompt | llm | PydanticOutputParser走astream()前面模型吐了十几个 token 数据解析器那边一个都不会冒出来最后一次性输出最终结果。做前端的同学如果发现流式到一半就停了然后又一下子出现完整内容问题十有八九出在这里。2.3 非 Token 数据的处理元数据与工具调用片段另外一个容易被忽略的环节是现代 LLM 的流式响应里不一定只有文本 Token。如果你开启了函数调用Function Calling / Tool Calling模型可能先吐出一段tool_calls的增量参数再吐出文本如果你在 LangChain 里挂了回调处理器callbacks流式过程中还会穿插各种事件的回调。我在实际项目里处理过一种脏数据模型同时返回content和tool_calls的 chunk前端只关心content但如果不加过滤这些tool_calls的字符串会被直接拼进正文里用户看到的就是满屏乱码结构。解决方案是做一个统一的流式消息格式化层把不同类型的输出拆成不同的 SSE 事件token事件纯文本增量直接拼到正文tool事件工具调用信息前端可以展示正在检索资料之类的状态meta事件最终的结构化数据比如 JSON 提取结果、意图分类、引用列表。这么设计的好处是前端只负责展示后端告诉它这块是正文这块是状态这块是 JSON两边职责清楚调试也方便。后面第 4 节我会给出完整代码。3. FastAPI LangChain 流水线实战后端 SSE 封装与心跳机制3.1 用 StreamingResponse 正确构建 SSE 响应FastAPI 可以说是和 LangChain 流式配合得最顺的 Python 框架。它的StreamingResponse接收一个可迭代对象然后通过 HTTP 分块传输把数据一块一块发出去。我们要做的就是把 LangChain 的异步生成器包装成 SSE 格式。核心代码如下我尽量把注释写清楚import asyncio import json import time from fastapi import FastAPI from fastapi.responses import StreamingResponse from pydantic import BaseModel from langchain_openai import ChatOpenAI from langchain_core.prompts import ChatPromptTemplate from langchain_core.output_parsers import StrOutputParser app FastAPI() llm ChatOpenAI( modelgpt-4o-mini, temperature0.7, streamingTrue, ) prompt ChatPromptTemplate.from_messages([ (system, 你是专业的中文助手。回答要结构清晰、语言自然。), (human, {question}), ]) # 这个 chain 只负责流式文本 chat_chain prompt | llm | StrOutputParser() def sse_format(event: str, payload: dict) - str: 把事件和数据包装成 SSE 协议的一帧 return fevent: {event}\ndata: {json.dumps(payload, ensure_asciiFalse)}\n\n async def sse_generator(question: str): 生成器把 LangChain 的流式输出转成 SSE 帧 # 1. 先发一个多行注释作为心跳顺便告诉前端“我开始了” yield : connected\n\n yield sse_format(meta, {type: start, timestamp: time.time()}) try: # 2. 核心循环逐 token 包装 async for chunk in chat_chain.astream({question: question}): # 兼容 AIMessageChunk 的情况 if isinstance(chunk, str): content chunk elif hasattr(chunk, content): content chunk.content or else: content str(chunk) # 有真实内容才发 token 事件空白跳过 if content.strip(): yield sse_format(token, {content: content}) # 3. 结束标记 yield sse_format(end, {type: done}) except Exception as exc: # 生成器内捕获异常否则客户端只会看到连接中断什么都拿不到 yield sse_format(error, {message: str(exc)}) class ChatRequest(BaseModel): question: str app.post(/chat) async def chat_endpoint(req: ChatRequest): return StreamingResponse( sse_generator(req.question), media_typetext/event-stream, headers{ Cache-Control: no-cache, Connection: keep-alive, X-Accel-Buffering: no, # 告诉 Nginx 别缓冲 }, )有几个细节我要特别强调这些是我实际运行时踩出来的坑第一media_type必须是text/event-stream。这是浏览器识别 SSE 响应的硬条件。如果你漏掉了前端即使收到了数据EventSource 也不会正确解析。用 fetch 流式读取时这个字段依然重要因为有些代理服务器会根据 Content-Type 决定要不要缓冲。第二X-Accel-Buffering: no这个响应头不能省。当你把服务部署到 Nginx 后面时Nginx 默认会缓存 FastAPI 发出来的所有数据攒够一定量再一次性发给客户端。那打字机效果就废了——用户看到的还是转圈半天猛一下全出来。加了这个头就是告诉 Nginx这条响应你不要碰来一个帧发一个帧。第三生成器里必须 try-except。模型生成过程中抛异常是常有的事上下文超限、内容过滤、网络抖动。不 catch 的话FastAPI 会把连接直接断开前端拿到的是半截消息什么错误信息都没有。把异常包装成 SSE 的error事件发给前端至少用户能知道出错了不至于以为是网络挂了。3.2 同步生成器和异步生成器前端卡顿的隐患如果你的业务代码里拿的是同步的chain.stream()在 FastAPI 里有两种用法把同步生成器传给StreamingResponseFastAPI 会在线程池里跑它不会阻塞主事件循环但要注意同步生成器里如果调用了阻塞的 I/O 或 CPU 密集操作还是会占住线程池资源。我踩过的一个坑是这样的一开始我图省事在async def端点里直接用chain.stream()同步方法循环结果事件循环被阻塞前端同时发起的第二个请求要等第一个跑完才开始响应。这个问题特别隐蔽因为单个请求测试时根本感觉不出来一旦并发量上来接口响应曲线直接爆炸。正确姿势是用astream()配合async for或者在生成器函数前面加一个线程池装饰器/asyncio.to_thread包装。而且我建议把生成器本身也设计成异步的让 LangChain 的异步流式能力真正发挥出来。对于大多数生产环境来说这就是你要的代码。3.3 心跳机制与空闲超时问题先给你复现一个真实场景。我用 LangChain 接了个长思考链路的模型内部套了好几个子 Agent有一次用户问了一个复杂计算题模型在内部工具推理阶段沉默了 40 秒没吐任何 Token。连接中断前端报了一个让我们排查了很久的错误——消息流在完成前就断开了像极了超时等待 SSE 数据。这个问题的根因是空闲超时idle timeout。网络设备、Nginx、Gunicorn 都以这条连接最近有没有数据流动作为判断依据。模型 40 秒没吐数据Nginx 觉得这连接废了直接掐断。解决思路有两个层次治本在生成器里加心跳。每隔一段时间我一般设 15 秒主动发送一个 SSE 注释帧: heartbeat\n\n让连接一直有数据流动网络设备就不会误判。治标调大 Nginx 和 Gunicorn 的超时时间比如proxy_read_timeout 300s、Gunicorn--timeout 300。但这只是给服务器更多的耐心如果模型真的长时间无响应该断还是会断。心跳帧用:开头是 SSE 规范规定的注释格式接收方看到会无视它不会当成业务数据。所以它可以安全地混在真实数据流中间。async def sse_generator_with_heartbeat(question: str): last_heartbeat time.time() async for chunk in chat_chain.astream({question: question}): content chunk if isinstance(chunk, str) else getattr(chunk, content, ) if content.strip(): yield sse_format(token, {content: content}) last_heartbeat time.time() elif time.time() - last_heartbeat 15: yield : heartbeat\n\n last_heartbeat time.time()这样即使模型长时间思考连接也不会因为空闲而中断。4. 结构化输出与 JSON 解析让模型不只“说话”还能“填表单”4.1 LangChain 结构化输出的三种主流方式流式文本解决的是用户看得到响应的问题但 AI 应用落地时还面临另一个问题模型输出的是自然语言系统怎么才能拿到结构化的数据去做后续处理比如从简历里提取姓名、工作年限、技能列表或者让 Agent 返回一个可执行的 JSON 指令。LangChain 提供三种主流方式我一一说清楚它们各自的适用场景方式一Pydantic 输出解析器PydanticOutputParser这是最经典的方式。你先定义一个 Pydantic 模型就是数据结构LangChain 会把它的 schema 描述自动塞进 Prompt 的format_instructions部分要求模型按这个格式输出 JSON。模型输出后解析器把 JSON 转成 Pydantic 对象。整个过程不依赖特定模型的函数调用能力适用面广。方式二with_structured_output()方法这是 ChatOpenAI 等模型封装类上的推荐方法。底层原理是让模型走 Function Calling / 原生 JSON 模式把 Pydantic schema 转成工具的 JSON Schema模型直接以工具参数的形式返回结构化数据。这种方式在 OpenAI 系模型上准确率极高因为模型天生被训练过调用函数时严格按 schema 输出。方式三自定义response_format参数如果你用的是 OpenAI 兼容接口可以直接在ChatOpenAI里传model_kwargs{response_format: {type: json_object}}强制模型输出 JSON。这个方式的约束力不如前两种适合场景极简单的需求。我用一张表对比它们的差异方式约束力依赖模型能力适用场景PydanticOutputParser中等靠 Prompt 引导所有模型通用场景兼容各类模型with_structured_output强走工具调用协议OpenAI 及支持工具的模型需要高准确率解析response_format弱只强制 JSON 格式OpenAI 兼容接口简单、无嵌套结构的输出4.2 用 Pydantic 定义输出 Schema 并接入链路我实际项目里最常用的组合是方式二 Pydantic。下面是一个从面试评价文本中提取结构化信息的例子from langchain_core.pydantic_v1 import BaseModel, Field class InterviewEvaluation(BaseModel): 面试评价结构化提取 candidate_name: str Field(description候选人姓名) overall_score: int Field(description综合评分0到100分) strengths: list[str] Field(description候选人的优点列出2到4条) risks: list[str] Field(description候选人的风险点列出1到3条) hiring_verdict: str Field(description录用建议只能是hire/no_hire/on_hold之一) # 直接使用 with_structured_output把 Pydantic 模型转成 OpenAI 工具协议 structured_llm llm.with_structured_output(InterviewEvaluation) result structured_llm.invoke( 张三的编程能力很强算法题全过但沟通表达存在明显欠缺 ) print(result.candidate_name) # 张三 print(result.overall_score) # 某个整数 print(result.hiring_verdict) # 某个枚举值这里要解释两个细节。第一Pydantic 模型里的Field(description...)不是写给人看的注释LangChain 会把 description 拼进给模型的工具 schema模型靠它理解字段含义所以描述写得越具体输出正确率越高。第二枚举值hire/no_hire/on_hold直接用字段值和字符串描述比单纯让模型自由发挥可靠得多。4.3 流式文本和结构化输出如何共存项目做到一半你会发现一个很现实的需求既要让用户看到逐字生成的回复又要让后端拿到一个结构化的 JSON 做业务逻辑。这两个需求能不能同时满足能但需要设计。我的做法是在一条 SSE 流里同时发两种事件。token事件走流式文本让前端做打字机效果meta事件在文本流结束后发送结构化结果。示例代码如下router.post(/chat_interview) async def chat_interview(req: InterviewRequest): async def gen(): # 先流式输出评价文本 async for token in interview_chain.astream({question: req.question}): yield sse_format(token, {content: token}) # 再调用结构化模型提取评价要点 eval_result await structured_llm.ainvoke(req.question) yield sse_format(meta, { type: structured, data: eval_result.dict(), }) yield sse_format(end, {type: done}) return StreamingResponse(gen(), media_typetext/event-stream)前端拿到meta事件后可以把它渲染成侧边栏的智能摘要卡片或者直接塞进表单。这样用户看到的是自然流畅的对话系统拿到的是可入库的结构化数据一举两得。不过要提醒你这种流式文本 尾部结构化的模式有一个小问题结构化结果要等模型完整生成才能出所以如果是长文本前端可能要等好几秒才看到侧边栏更新。如果对实时性要求更高你可以考虑让结构化模型和流式模型并行跑两个 LLM 调用同时发出去结构化结果先到就先发文本流自然也不会受影响。代价是成本翻倍。大多数场景下先文本后结构化已经够用没必要提前优化。4.4 JSON 解析的兜底容错方案理论上with_structured_output已经保证了模型按 schema 返回 JSON但现实世界永远有意外。比如你接入了一个第三方模型它照样走 OpenAI 兼容接口但偶尔会把 JSON 包在 markdown 代码块里或者输出到一半被截断。这种时候如果你直接json.loads()就是一个JSONDecodeError等着你。我的兜底解析流程一共四层按顺序尝试大量减少了线上报错import json import re def robust_json_parse(text: str): 四层兜底解析模型输出的 JSON # 第一层去掉 markdown 代码块包裹 if text.startswith(): text re.sub(r^(?:json)?\s*|\s*$, , text, flagsre.MULTILINE).strip() # 第二层标准解析 try: return json.loads(text) except json.JSONDecodeError: pass # 第三层截取最外层大括号忽略前后杂音 match re.search(r\{.*\}, text, re.DOTALL) if match: try: return json.loads(match.group(0)) except json.JSONDecodeError: pass # 第四层用 json_repair 做宽松修复处理缺失引号、末尾逗号等 try: from json_repair import repair_json repaired repair_json(text) return json.loads(repaired) except Exception: return None这里有个容易踩的坑第三层的正则\{.*\}用的是贪婪模式如果 JSON 里有嵌套的对象它会一直匹配到最后一个}正好是完整 JSON 的外层所以能正确提取。但如果模型输出里同时有两个 JSON 对象这个正则也会把它们拼在一起导致解析失败。所以这层只是兜底不能作为主路径。还有一个思路很多人会忽略利用错误反馈让模型自我修正。当解析失败时把异常信息和提取到的残缺 JSON 一起塞回 Prompt让模型在不重新生成全部内容的前提下只补全出错的部分。我在带错误重试机制的场景里测试过修复成功率能到 80% 以上。5. 前端打字机效果实现Vue fetch 流式读取全流程5.1 为什么最终选了 fetch 而不是 EventSource浏览器自带 SSE 客户端就是EventSource它开箱即用、自动重连很多教程都推荐它。但我在项目里实际对接 LangChain 的后端接口时踩了三个让它不那么好用的坑EventSource 只能发 GET 请求可我需要在请求体里带对话上下文、用户配置、历史消息等一堆 JSON硬塞到 URL 里既丑陋又容易超长。EventSource 不能自定义请求头。我想在 Authorization 头里带用户的登录凭证EventSource 就是加不上去只能退而求其次用 Cookie 或 URL 参数既不安全又别扭。EventSource 对事件处理的粒度不够。它是面向接收推送设计的对于token、meta、end这种需要区分处理的自定义事件你得写一堆 event listener 去注册不如手写 fetch 解析来得直观。所以我最终采用的方案是用 fetch 的 POST 请求去打 SSE 接口然后通过response.body.getReader()读取流式数据自己解析 SSE 帧。看起来多写了几行代码但换来的是完全可控的请求参数和响应处理。5.2 ReadableStream 解析 SSE 帧核心代码逐段说明前端解析的关键在于SSE 数据不是一次到位的而是一块一块地来。网络字节流可能把一条消息拆成两半也可能把多条消息粘在一起。所以解析器必须维护一个buffer每收到一块数据就追加进 buffer然后按\n\n空行切分完整帧剩余的半截留到下一轮继续拼。一段可用的 Vue 3 组合式函数代码如下export function useSSE() { const fullText ref() const isStreaming ref(false) const structuredData ref(null) async function sendMessage(question) { fullText.value structuredData.value null isStreaming.value true const response await fetch(/api/chat, { method: POST, headers: { Content-Type: application/json }, body: JSON.stringify({ question }) }) if (!response.ok) { throw new Error(请求失败: ${response.status}) } const reader response.body.getReader() const decoder new TextDecoder(utf-8) let buffer while (true) { const { done, value } await reader.read() if (done) break // 把 Uint8Array 解码成字符串注意 stream: true 处理多字节字符 buffer decoder.decode(value, { stream: true }) // 按空行切分 SSE 帧最后一个可能是半截帧先放回 buffer const frames buffer.split(\n\n) buffer frames.pop() for (const frame of frames) { const parsed parseSSEFrame(frame) if (parsed.event token) { fullText.value parsed.data.content } else if (parsed.event meta) { structuredData.value parsed.data } else if (parsed.event error) { console.error(服务端错误:, parsed.data.message) } } } isStreaming.value false } return { fullText, structuredData, isStreaming, sendMessage } } function parseSSEFrame(frame) { const lines frame.split(\n) let event message let data for (const line of lines) { if (line.startsWith(event:)) { event line.slice(6).trim() } else if (line.startsWith(data:)) { // SSE 允许多行 data拼一个换行再 trim 保持 JSON 可解析 data line.slice(5) \n } } // 去掉末尾多余的换行位再解析 JSON const trimmed data.replace(/\n$/, ) return { event, data: trimmed ? JSON.parse(trimmed) : {} } }这段代码里我特别想提醒一个细节TextDecoder的第二个参数{ stream: true }千万别漏。UTF-8 的中文字符在二进制流里可能横跨两个 chunk比如第一个 chunk 来了 2 个字节第二个 chunk 才来剩余的 1 个字节。如果你在每一轮都单独decode(value)中文就会在中间断开变成乱码。{ stream: true }会告诉解码器数据没完我帮你缓存到完整字符再输出这是中文流式场景下的必备参数。5.3 打字机效果与 Markdown 实时渲染的协调拿到流式文本后最朴素的打字机效果就是把fullText直接绑定到某个 DOM 节点。但如果你在文本里混了 Markdown 语法直接v-html绑定原始文本会暴露一堆##、**符号体验极差。正确做法是每一次 token 增量后重新跑一次 Markdown 渲染用渲染结果替换 DOM 内容。这时候会有新的问题重新渲染会让光标跳动或者滚动位置不稳定。原因很简单Markdown 解析器输出的是完整的 HTML 字符串替换 innerHTML 会重置 DOM 结构导致用户阅读位置跳到页面顶部。解决思路是用nextTick等 DOM 更新完成后手动把滚动容器 scrollTop 设置到 scrollHeight如果渲染量大考虑做节流渲染比如每秒最多重新渲染两次让打字效果不至于因为频繁解析 markdown 而卡顿如果文本较长且包含表格、代码块这类复杂 Markdown建议前端用marked.js加highlight.js的组合解析性能比直接套 Vue 的v-html强很多。在这个环节你要注意异步流式状态下不要直接用computed对fullText做marked.parse()然后绑到 v-html 上。因为每次fullText更新computed 都会同步重新计算一旦文本量大了UI 线程会被 Markdown 解析阻塞打字机效果直接卡成 PPT。我的处理方案是把marked.parse放到watch的回调里或者干脆在 Message 对象上用requestIdleCallback做延迟解析把渲染压力分散到浏览器空闲时段。6. 线上常见问题排查实录SSE 断连、超时与解析失败6.1 stream disconnected before completion 的根因排查这是我在评论区里被问得最多的一个错误。完整报错是类似stream disconnected before completion: idle timeout waiting for sse。我排查这类问题有一套固定流程按顺序操作能快速定位先看是不是空闲超时。打开浏览器开发者工具看 Network 面板里这条 SSE 请求的响应时间线。如果请求在某个平静期突然结束曲线变平很久后断开基本可以锁定是中间设备在超时。直接在后端生成器里加心跳帧参考 3.3 节。再看代理层。把服务直连和走 Nginx 各测一次。如果直连正常、走 Nginx 断连那九成是 Nginx 对长连接的 read timeout 设置太短。用我下面的配置location /api/ { proxy_pass http://backend; proxy_http_version 1.1; proxy_set_header Connection ; proxy_buffering off; proxy_cache off; proxy_read_timeout 300s; }注意proxy_http_version 1.1很关键HTTP/1.0 不支持分块传输编码SSE 会直接失效。最后看应用层超时配置。如果你用 Gunicorn 跑 FastAPI默认 worker 超时是 30 秒模型生成超过 30 秒 Gunicorn 会强制杀 worker。启动命令里加--timeout 300。6.2 常见问题速查表下面这张表是我在实际项目中整理出来的高频问题按现象—原因—解决方案列出来排查时直接对照现象根因解决方案前端转圈很久最后一次性显示全部内容Nginx 缓冲导致数据攒批发送加X-Accel-Buffering: no响应头proxy_buffering off对话进行中突然断连前端无错误提示空闲超时idle timeout生成器内加: heartbeat心跳帧流式接口同时并发请求时变慢同步chain.stream()阻塞事件循环改用chain.astream()配合异步生成器前端收到中文乱码TextDecoder没有设置{ stream: true }按 5.2 节代码加上JSON.parse报错模型输出被 markdown 包裹或截断用 4.4 节的四层兜底解析网页开着多个会话时某些流式连接不定时断开浏览器对同一域名连接数限制收紧页面数量或用 HTTP/2LangChain 流式中间有长停顿模型内部在等待工具调用心跳帧 前端展示正在处理状态提示6.3 一个让我印象深刻的坑EventSource 的隐式缓冲陷阱最后分享一个非常隐蔽的坑。某个页面我一开始用的 EventSource 实现打字机效果单条消息没有任何问题。但在一个后台管理页里同时开了 5 个 EventSource监听不同 schema 的推送结果其中两个连接经常悄悄断掉怎么查都查不出原因。后来才反应过来HTTP/1.1 协议中浏览器对同一个域名的并发连接数限制是 6。页面里其他资源请求占了一部分再开多个 EventSource连超了之后浏览器就会挂起连接。加上我用的是 Docker 部署的 FastAPI服务端自身没有维护长连接池压力一大连接就有可能被系统层面回收。解决方案有两个方向要么改用 HTTP/2多路复用让连接数限制不再成为瓶颈要么像我后来做的把多个 SSE 订阅合并成一个接口用多 event 类型区分业务。对大部分中小项目来说后者实现成本更低效果也很直接。写在最后也是我踩过坑之后最想告诉你的把这套方案完整落地之后我回头看的最大的体会是流式传输这件事难的不是协议本身而是链路中每一层的配合。LangChain 输出 Token、FastAPI 包装 SSE、Nginx 不捣乱、前端增量渲染——任何一环想当然最后的体验就会打折扣。我自己最开始就是被流式接口能跑通这个表象骗了前端拿到 token 就直接渲染结果 JSON 全被拼在正文里、Nginx 缓冲把打字机效果吞成了整块输出、结构化解析失败直接让 Agent 中间状态崩掉。所以建议各位动手之前先画一张自己项目的 SSE 链路图标清楚每一环负责什么、可能在哪里出问题再动手写代码会少走很多弯路。另外一个后来觉得非常值的工作是把 SSE 的消息格式从服务端想发啥就发啥收敛成了事件类型 JSON 数据的规范。只要消息流动的地方都遵循这个规范前端就能精准地处理 token、meta、end、error 四类事件后端加新功能比如加个中间状态推送也不需要前端跟着改逻辑。如果你打算长期迭代一个 AI 产品我强烈建议你一开始就把这套消息约定定下来。最后再分享一个日常调试的实用小技巧先用curl测后端再连前端。curl -N http://localhost:8000/chat -d {question:你好}-N参数会禁用缓冲让你直接在终端看到每一个 SSE 帧的到达时间。如果终端里能逐帧刷出来后端基本没问题再去排查前端如果终端也是一次性输出问题就在后端或网络层。这个习惯帮我省了至少一半的联调时间。