1. 流式输出的本质为什么我们需要 SSE1.1 从“等半天”到“边生成边看”的体验革命做过 AI 应用的人都有一个共同的痛用户点下发送按钮之后界面就像死了一样转圈转个十几秒然后“啪”一下蹦出一大段文字。这种体验在 2023 年之前大家还能忍毕竟那时候模型本身就慢。但现在模型推理速度上来了用户对交互体验的要求也水涨船高你再让用户盯着一个空白页面等十秒人家直接关掉走人。流式输出解决的就是这个问题。它的核心思路特别朴素模型生成一个 token我就往前端推一个 token用户看到文字像打字机一样一个字一个字蹦出来心理上会觉得“它在干活”等待焦虑感大幅降低。这不是什么玄学是实打实的用户体验优化。而 SSEServer-Sent Events就是实现这种推送最顺手的技术方案之一。它基于 HTTP 长连接服务端可以持续向客户端单向推送数据浏览器原生支持 EventSource API不需要额外引入 WebSocket 那套复杂的握手和心跳逻辑。对于“服务端推、客户端收”这种单向流场景SSE 就是最合适的锤子。1.2 SSE 和 WebSocket 到底怎么选很多人一上来就纠结我到底该用 SSE 还是 WebSocket我的经验是先问自己一个问题——你的通信是双向的吗如果你的场景是 AI 对话、日志实时推送、股票行情刷新这类“服务端持续推、客户端只管收”的模式SSE 完胜。它天然支持断线重连浏览器会自动重试协议简单到用 curl 就能调试服务端实现也就几十行代码的事。但如果你需要客户端频繁向服务端发消息比如多人协作编辑、实时游戏同步那 WebSocket 才是正解。SSE 虽然也能通过 POST 发请求但每次都得重新建连接效率不行。我自己的项目里AI 对话场景一律用 SSE从来没出过什么大问题。唯一需要注意的是连接数限制——浏览器对同域名的 SSE 连接数有上限HTTP/1.1 下通常是 6 个如果你的页面同时开多个流得考虑用 HTTP/2 或者做连接复用。1.3 SSE 协议格式比你想的简单得多SSE 的数据格式简单到令人发指。服务端返回的 Content-Type 是text/event-stream然后每条消息按固定格式拼接data: 这是第一条消息\n\n data: 这是第二条消息\n\n就这么简单。每条消息以data:开头以两个换行符\n\n结尾。浏览器收到之后会自动触发onmessage回调。如果你需要指定事件类型可以加event:字段需要设置重连时间可以加retry:字段。但这里有个坑很多后端框架默认会缓冲响应导致你明明写了流式输出前端却还是等全部生成完才收到。解决办法是设置响应头X-Accel-Buffering: no针对 Nginx和Cache-Control: no-cache同时确保框架层面没有开启 gzip 压缩——gzip 会缓冲数据直到攒够一定大小才发送流式效果直接废掉。2. LangChain 结构化输出让 AI 说人话也办人事2.1 为什么自由文本不够用用过 LangChain 的人都知道默认情况下 LLM 返回的是一坨自由文本。你问它“帮我提取这篇文章的作者、发布时间和核心观点”它可能给你返回一段散文式的回答读起来挺通顺但你想用代码解析对不起正则写到崩溃。这就是结构化输出的价值所在。我们需要的是 LLM 直接返回一个 JSON 对象字段名固定、类型明确程序拿到就能用。比如{ author: 张三, publish_date: 2024-01-15, key_points: [观点一, 观点二, 观点三] }LangChain 提供了多种方式来实现这个目标从最简单的PydanticOutputParser到更现代的with_structured_output方法各有适用场景。2.2 PydanticOutputParser经典但略显笨重早期 LangChain 项目里最常见的就是PydanticOutputParser。它的工作流程是你定义一个 Pydantic 模型parser 会自动生成格式说明你把这段说明塞进 prompt 里告诉模型“请按这个格式返回”然后模型返回文本parser 再解析成 Pydantic 对象。from langchain.output_parsers import PydanticOutputParser from pydantic import BaseModel, Field class ArticleInfo(BaseModel): author: str Field(description文章作者) publish_date: str Field(description发布日期格式YYYY-MM-DD) key_points: list[str] Field(description核心观点列表) parser PydanticOutputParser(pydantic_objectArticleInfo) format_instructions parser.get_format_instructions()这种方式的好处是兼容性好几乎所有模型都能用。缺点也很明显你得把格式说明塞进 prompt占 token 不说模型还不一定听话。我遇到过模型返回的 JSON 里多了一个逗号、少了一个引号parser 直接抛异常的情况。所以实际项目中我一般会加一层重试机制解析失败就重新请求一次。2.3 with_structured_output更优雅的现代方案LangChain 后来推出了with_structured_output方法直接利用模型原生的 function calling 或 JSON mode 能力。你只需要把 Pydantic 模型传进去LangChain 会自动处理格式约束from langchain_openai import ChatOpenAI llm ChatOpenAI(modelgpt-4o) structured_llm llm.with_structured_output(ArticleInfo) result structured_llm.invoke(提取这篇文章的信息...)这种方式的好处是模型层面就保证了输出格式解析成功率极高。但前提是你用的模型支持 function calling。如果你用的是本地部署的开源模型可能还得回退到 PydanticOutputParser 那套方案。2.4 结构化输出与流式的冲突与调和这里有一个很多人会踩的坑结构化输出和流式输出天然有冲突。流式输出是一个 token 一个 token 往外蹦但 JSON 必须完整才能解析。你不可能在收到{author: 张的时候就解析出作者是谁。解决方案有两种思路。第一种是“先流式展示后结构化解析”——前端先把原始文本流式展示给用户看等流结束了再调用 parser 解析成结构化数据。这种方式适合对实时性要求不高的场景。第二种是“流式 JSON 解析”也就是边接收边解析。这需要用到一些支持增量解析的库比如ijson或者自己写一个状态机。实现起来复杂一些但用户体验最好。我在一个合同信息提取项目里用过这种方式用户能看到字段一个一个被填满反馈非常好。3. 打字机效果的前端实现不只是 CSS 动画3.1 真正的流式渲染 vs 假打字机市面上很多所谓的“打字机效果”其实是假的等全部内容返回之后用 setInterval 每隔几十毫秒往页面上加一个字。这种做法在内容短的时候看起来还行但内容一长就露馅了——用户会发现文字蹦出来的速度和网络请求完成的时间对不上。真正的流式渲染是后端推一个 chunk前端就渲染一个 chunk。Vue 里可以用fetchReadableStream来实现const response await fetch(/api/chat, { method: POST, body: JSON.stringify({ query }) }) const reader response.body.getReader() const decoder new TextDecoder() while (true) { const { done, value } await reader.read() if (done) break const chunk decoder.decode(value) // 解析 SSE 格式提取 data 字段 const lines chunk.split(\n) for (const line of lines) { if (line.startsWith(data: )) { const text line.slice(6) appendToDisplay(text) } } }这种方式的好处是真正的实时后端推多快前端就显示多快。但要注意处理粘包问题——多个 SSE 消息可能被合并到一个 chunk 里你需要按\n\n分割。3.2 处理 Markdown 流式渲染的坑AI 返回的内容通常是 Markdown 格式包含代码块、列表、表格等。流式渲染 Markdown 有个经典问题代码块还没闭合的时候Markdown 解析器会把后面的内容全部当成代码。比如模型正在输出python def hello():这时候 Markdown 解析器看到三个反引号就认为代码块开始了但结束的反引号还没来于是后面的所有内容都被渲染成代码。等结束反引号到了页面又会突然重排体验很差。我的解决办法是在流式渲染阶段先做一次“预判”如果检测到未闭合的代码块就暂时不渲染 Markdown而是用纯文本展示等代码块闭合后再切换成 Markdown 渲染。虽然实现麻烦一点但效果确实好很多。3.3 自动滚动与用户打断的平衡流式输出的时候页面内容不断增长你需要自动滚动到底部让用户看到最新内容。但如果用户手动往上滚去看历史消息你还强制滚动到底部用户会想砸键盘。正确的做法是监听滚动事件判断用户是否在底部附近。如果在底部就自动滚动如果用户主动往上滚了就暂停自动滚动并显示一个“回到底部”的按钮。这个细节虽小但直接影响用户体验。4. 后端流式接口封装从 FastAPI 到 LangChain 的完整链路4.1 FastAPI 的 StreamingResponse 实战FastAPI 提供了StreamingResponse来支持流式输出。基本用法是传入一个生成器函数from fastapi import FastAPI from fastapi.responses import StreamingResponse app FastAPI() async def event_generator(query: str): async for chunk in llm.astream(query): yield fdata: {chunk.content}\n\n app.post(/api/chat) async def chat(query: str): return StreamingResponse( event_generator(query), media_typetext/event-stream, headers{ Cache-Control: no-cache, X-Accel-Buffering: no } )这里有几个关键点。第一生成器必须是异步的否则会阻塞事件循环。第二media_type必须设为text/event-stream否则浏览器不会按 SSE 处理。第三X-Accel-Buffering: no是给 Nginx 看的告诉它不要缓冲这个响应。4.2 LangChain 的 astream 与 astream_eventsLangChain 提供了两个流式接口astream和astream_events。astream返回的是 LLM 输出的 token 流适合简单的对话场景。astream_events则更强大它能返回整个链路上所有事件包括 LLM 开始、LLM 结束、工具调用、工具返回等。如果你只是做简单的对话用astream就够了async for chunk in llm.astream(你好): print(chunk.content, end, flushTrue)但如果你用了 Agent 或者 Chainastream_events才是正确的选择。它的事件类型包括on_chat_model_stream、on_tool_start、on_tool_end等你可以根据事件类型决定往前端推什么内容。比如工具调用的时候你可以推一个“正在查询数据库...”的提示让用户知道系统在干活。4.3 封装一个可复用的 SSE 流式接口在实际项目中我一般会把 SSE 的封装逻辑抽成一个独立的模块统一处理格式拼接、错误捕获和心跳保活import json import asyncio from typing import AsyncGenerator class SSEStreamer: def __init__(self, heartbeat_interval: int 15): self.heartbeat_interval heartbeat_interval async def stream(self, generator: AsyncGenerator) - AsyncGenerator[str, None]: try: async for chunk in generator: yield fdata: {json.dumps({content: chunk})}\n\n except Exception as e: yield fdata: {json.dumps({error: str(e)})}\n\n finally: yield data: [DONE]\n\n心跳保活是很多人忽略的点。如果 LLM 生成速度慢中间可能有十几秒没有数据推送某些代理服务器会认为连接空闲而主动断开。解决办法是每隔一段时间推一个空注释行: heartbeat\n\n保持连接活跃。5. 常见问题与排查技巧实录5.1 流式输出突然中断怎么办这是被问得最多的问题。流式输出跑到一半突然断了前端收到一个不完整的响应。常见原因有三个第一代理服务器超时。Nginx 默认的proxy_read_timeout是 60 秒如果 LLM 生成时间超过这个值连接会被切断。解决办法是调大这个值或者用心跳保活。第二后端异常未捕获。生成器函数里抛了异常但没有 try-except 包裹导致连接直接断开。解决办法是在生成器里加全局异常捕获把错误信息也通过 SSE 推给前端。第三前端读取逻辑有 bug。比如reader.read()返回的 chunk 没有正确处理粘包导致部分数据丢失。解决办法是维护一个缓冲区按\n\n分割完整的 SSE 消息。5.2 JSON 解析失败的排查思路结构化输出解析失败是另一个高频问题。我的排查顺序是这样的先看模型原始输出。把 LLM 返回的原始文本打印出来看看是不是格式不对。常见问题包括模型在 JSON 外面包了一层 Markdown 代码块、JSON 里有尾随逗号、字段名用了中文引号等。再看 prompt 是否清晰。如果你用的是 PydanticOutputParser检查format_instructions是否完整塞进了 prompt。有时候 prompt 太长模型会忽略格式要求。最后看模型能力。有些小模型对 JSON 格式的遵循能力确实差这时候要么换模型要么加 few-shot 示例要么用 output parser 的重试机制。5.3 常见问题速查表问题现象可能原因解决方案流式输出变成一次性返回代理或框架开启了缓冲设置X-Accel-Buffering: no关闭 gzip连接中途断开代理超时或后端异常调大超时时间加心跳保活捕获异常JSON 解析失败模型输出格式不对检查 prompt加 few-shot加重试机制前端显示乱码编码问题或粘包处理不当确保 UTF-8 编码按\n\n分割消息打字机效果卡顿前端渲染频率过高用 requestAnimationFrame 节流渲染5.4 几个我踩过的坑第一个坑在 FastAPI 的生成器里用了同步的time.sleep()导致整个事件循环被阻塞所有请求都卡住。后来改成await asyncio.sleep()才解决。第二个坑前端用EventSource的时候发现无法自定义请求头导致无法传认证 token。后来改用fetchReadableStream才解决。如果你的认证逻辑依赖 header千万别用EventSource。第三个坑LangChain 的astream_events在某些版本里会重复触发事件导致前端收到重复内容。解决办法是升级到最新版本或者在事件处理里加去重逻辑。6. 从对话到 Agent流式输出的进阶玩法6.1 Agent 场景下的流式事件处理当你从简单的 LLM 对话升级到 Agent 的时候流式输出的复杂度会上一个台阶。因为 Agent 不只是生成文本它还会调用工具、查询数据库、执行代码。用户需要知道 Agent 当前在干什么而不是盯着一个空白页面等。astream_events在这里就派上用场了。你可以监听不同的事件类型给用户不同的反馈async for event in agent.astream_events(input, versionv2): kind event[event] if kind on_chat_model_stream: # LLM 正在生成文本 yield event[data][chunk].content elif kind on_tool_start: # 工具开始执行 yield f\n[正在调用工具: {event[name]}]\n elif kind on_tool_end: # 工具执行完成 yield f\n[工具执行完成]\n这样用户就能看到 Agent 的完整思考过程体验比单纯等一个最终答案好得多。6.2 多轮对话中的上下文管理流式输出和多轮对话结合的时候有一个容易被忽略的问题上下文长度。每轮对话都会往历史记录里追加内容几轮之后 token 数就爆了。解决办法是定期做上下文压缩或者只保留最近 N 轮对话。我一般会在后端维护一个会话状态每次请求的时候把历史消息和当前消息一起传给 LLM。但要注意流式输出的时候不能把历史消息也流式推给前端否则用户会看到之前的内容重复出现。正确的做法是只推当前轮次的增量内容。6.3 结构化输出在 Agent 中的应用Agent 调用工具的时候工具的入参通常需要结构化数据。比如一个查询天气的工具需要{city: string, date: string}这样的参数。这时候就可以用with_structured_output来让 LLM 直接生成工具入参避免手动解析文本。LangChain 的 Agent 框架其实已经内置了这个能力你只需要用tool装饰器定义工具框架会自动处理参数解析。但如果你是自己手写 Agent 循环那就需要自己实现结构化输出到工具调用的转换。7. 性能优化与生产环境注意事项7.1 减少首 token 延迟流式输出的体验好坏很大程度上取决于首 token 延迟。用户点下发送之后如果 2 秒内能看到第一个字蹦出来心理上就能接受。如果超过 5 秒还是空白用户就会开始怀疑是不是卡了。减少首 token 延迟的方法有几个。第一用更快的模型比如 GPT-4o-mini 就比 GPT-4 快很多。第二精简 promptprompt 越短模型开始生成的速度越快。第三预热连接提前建立好到模型服务的连接避免每次请求都重新握手。7.2 并发流式请求的资源管理每个 SSE 连接都会占用一个服务端连接和一个协程。如果并发量上来了比如同时有几百个用户在对话服务端资源会吃紧。解决办法是用连接池限制最大并发数超出的请求排队等待。另外要注意Python 的 GIL 在 IO 密集型场景下影响不大但如果你的生成器里有 CPU 密集型操作比如复杂的文本处理会阻塞其他协程。这种操作应该放到线程池里执行。7.3 日志与监控生产环境里流式接口的日志和监控特别重要。你需要记录每个请求的开始时间、首 token 时间、结束时间、token 总数、是否异常等信息。这些数据能帮你发现性能瓶颈和异常情况。我一般会在 SSEStreamer 里加一个计时器记录首 token 延迟和总耗时然后上报到监控系统。如果发现某个时间段首 token 延迟飙升就能及时排查是不是模型服务出了问题。8. 一个完整的实战案例合同信息提取8.1 需求分析与方案设计假设我们要做一个合同信息提取工具。用户上传一份合同 PDF系统自动提取甲方、乙方、合同金额、签署日期等关键信息并以结构化形式展示。这个场景的特点是用户需要看到提取过程哪些字段已经提取到了同时最终结果需要是结构化的 JSON。所以我们需要流式输出 结构化输出的组合方案。8.2 后端实现后端用 FastAPI LangChain 实现。首先定义 Pydantic 模型from pydantic import BaseModel, Field class ContractInfo(BaseModel): party_a: str Field(description甲方名称) party_b: str Field(description乙方名称) amount: float Field(description合同金额) sign_date: str Field(description签署日期)然后用with_structured_output让 LLM 直接返回结构化数据。但为了流式展示我们先用普通 LLM 流式输出提取过程等流结束后再调用结构化提取app.post(/api/extract) async def extract_contract(file: UploadFile): text extract_text(file) async def generate(): # 第一阶段流式展示提取过程 async for chunk in llm.astream(f请提取以下合同的关键信息{text}): yield fdata: {json.dumps({type: process, content: chunk.content})}\n\n # 第二阶段结构化提取 structured_llm llm.with_structured_output(ContractInfo) result structured_llm.invoke(f提取合同信息{text}) yield fdata: {json.dumps({type: result, content: result.dict()})}\n\n yield data: [DONE]\n\n return StreamingResponse(generate(), media_typetext/event-stream)8.3 前端展示前端用 Vue 实现监听 SSE 消息根据type字段区分是过程内容还是最终结果。过程内容用打字机效果展示最终结果用表格展示。const eventSource new EventSource(/api/extract) eventSource.onmessage (event) { if (event.data [DONE]) { eventSource.close() return } const data JSON.parse(event.data) if (data.type process) { processText.value data.content } else if (data.type result) { contractInfo.value data.content } }这个方案上线之后用户反馈很好。他们能看到系统在“读”合同而不是干等一个结果。虽然总耗时没变但感知上的等待时间短了很多。9. 一些个人经验与建议流式输出这个东西看起来简单但细节特别多。我从第一个版本到现在前前后后改了七八次才把各种边界情况处理干净。最大的体会是不要等到所有东西都完美了再上线。先做一个能跑的版本让用户用起来然后根据反馈逐步优化。我第一版的时候连心跳保活都没做用户经常遇到连接断开但至少能用。后来慢慢加了心跳、加了异常捕获、加了结构化输出体验才越来越好。另一个建议是多写测试。流式接口的测试比普通接口麻烦因为你要模拟各种异常情况——连接中断、超时、格式错误等。但这些测试能帮你提前发现很多问题避免上线之后被用户投诉。最后如果你也在做类似的东西欢迎交流。这个领域变化很快LangChain 的 API 每隔几个月就变一次保持学习的心态很重要。
1. 流式输出的本质为什么我们需要 SSE1.1 从“等半天”到“边生成边看”的体验革命做过 AI 应用的人都有一个共同的痛用户点下发送按钮之后界面就像死了一样转圈转个十几秒然后“啪”一下蹦出一大段文字。这种体验在 2023 年之前大家还能忍毕竟那时候模型本身就慢。但现在模型推理速度上来了用户对交互体验的要求也水涨船高你再让用户盯着一个空白页面等十秒人家直接关掉走人。流式输出解决的就是这个问题。它的核心思路特别朴素模型生成一个 token我就往前端推一个 token用户看到文字像打字机一样一个字一个字蹦出来心理上会觉得“它在干活”等待焦虑感大幅降低。这不是什么玄学是实打实的用户体验优化。而 SSEServer-Sent Events就是实现这种推送最顺手的技术方案之一。它基于 HTTP 长连接服务端可以持续向客户端单向推送数据浏览器原生支持 EventSource API不需要额外引入 WebSocket 那套复杂的握手和心跳逻辑。对于“服务端推、客户端收”这种单向流场景SSE 就是最合适的锤子。1.2 SSE 和 WebSocket 到底怎么选很多人一上来就纠结我到底该用 SSE 还是 WebSocket我的经验是先问自己一个问题——你的通信是双向的吗如果你的场景是 AI 对话、日志实时推送、股票行情刷新这类“服务端持续推、客户端只管收”的模式SSE 完胜。它天然支持断线重连浏览器会自动重试协议简单到用 curl 就能调试服务端实现也就几十行代码的事。但如果你需要客户端频繁向服务端发消息比如多人协作编辑、实时游戏同步那 WebSocket 才是正解。SSE 虽然也能通过 POST 发请求但每次都得重新建连接效率不行。我自己的项目里AI 对话场景一律用 SSE从来没出过什么大问题。唯一需要注意的是连接数限制——浏览器对同域名的 SSE 连接数有上限HTTP/1.1 下通常是 6 个如果你的页面同时开多个流得考虑用 HTTP/2 或者做连接复用。1.3 SSE 协议格式比你想的简单得多SSE 的数据格式简单到令人发指。服务端返回的 Content-Type 是text/event-stream然后每条消息按固定格式拼接data: 这是第一条消息\n\n data: 这是第二条消息\n\n就这么简单。每条消息以data:开头以两个换行符\n\n结尾。浏览器收到之后会自动触发onmessage回调。如果你需要指定事件类型可以加event:字段需要设置重连时间可以加retry:字段。但这里有个坑很多后端框架默认会缓冲响应导致你明明写了流式输出前端却还是等全部生成完才收到。解决办法是设置响应头X-Accel-Buffering: no针对 Nginx和Cache-Control: no-cache同时确保框架层面没有开启 gzip 压缩——gzip 会缓冲数据直到攒够一定大小才发送流式效果直接废掉。2. LangChain 结构化输出让 AI 说人话也办人事2.1 为什么自由文本不够用用过 LangChain 的人都知道默认情况下 LLM 返回的是一坨自由文本。你问它“帮我提取这篇文章的作者、发布时间和核心观点”它可能给你返回一段散文式的回答读起来挺通顺但你想用代码解析对不起正则写到崩溃。这就是结构化输出的价值所在。我们需要的是 LLM 直接返回一个 JSON 对象字段名固定、类型明确程序拿到就能用。比如{ author: 张三, publish_date: 2024-01-15, key_points: [观点一, 观点二, 观点三] }LangChain 提供了多种方式来实现这个目标从最简单的PydanticOutputParser到更现代的with_structured_output方法各有适用场景。2.2 PydanticOutputParser经典但略显笨重早期 LangChain 项目里最常见的就是PydanticOutputParser。它的工作流程是你定义一个 Pydantic 模型parser 会自动生成格式说明你把这段说明塞进 prompt 里告诉模型“请按这个格式返回”然后模型返回文本parser 再解析成 Pydantic 对象。from langchain.output_parsers import PydanticOutputParser from pydantic import BaseModel, Field class ArticleInfo(BaseModel): author: str Field(description文章作者) publish_date: str Field(description发布日期格式YYYY-MM-DD) key_points: list[str] Field(description核心观点列表) parser PydanticOutputParser(pydantic_objectArticleInfo) format_instructions parser.get_format_instructions()这种方式的好处是兼容性好几乎所有模型都能用。缺点也很明显你得把格式说明塞进 prompt占 token 不说模型还不一定听话。我遇到过模型返回的 JSON 里多了一个逗号、少了一个引号parser 直接抛异常的情况。所以实际项目中我一般会加一层重试机制解析失败就重新请求一次。2.3 with_structured_output更优雅的现代方案LangChain 后来推出了with_structured_output方法直接利用模型原生的 function calling 或 JSON mode 能力。你只需要把 Pydantic 模型传进去LangChain 会自动处理格式约束from langchain_openai import ChatOpenAI llm ChatOpenAI(modelgpt-4o) structured_llm llm.with_structured_output(ArticleInfo) result structured_llm.invoke(提取这篇文章的信息...)这种方式的好处是模型层面就保证了输出格式解析成功率极高。但前提是你用的模型支持 function calling。如果你用的是本地部署的开源模型可能还得回退到 PydanticOutputParser 那套方案。2.4 结构化输出与流式的冲突与调和这里有一个很多人会踩的坑结构化输出和流式输出天然有冲突。流式输出是一个 token 一个 token 往外蹦但 JSON 必须完整才能解析。你不可能在收到{author: 张的时候就解析出作者是谁。解决方案有两种思路。第一种是“先流式展示后结构化解析”——前端先把原始文本流式展示给用户看等流结束了再调用 parser 解析成结构化数据。这种方式适合对实时性要求不高的场景。第二种是“流式 JSON 解析”也就是边接收边解析。这需要用到一些支持增量解析的库比如ijson或者自己写一个状态机。实现起来复杂一些但用户体验最好。我在一个合同信息提取项目里用过这种方式用户能看到字段一个一个被填满反馈非常好。3. 打字机效果的前端实现不只是 CSS 动画3.1 真正的流式渲染 vs 假打字机市面上很多所谓的“打字机效果”其实是假的等全部内容返回之后用 setInterval 每隔几十毫秒往页面上加一个字。这种做法在内容短的时候看起来还行但内容一长就露馅了——用户会发现文字蹦出来的速度和网络请求完成的时间对不上。真正的流式渲染是后端推一个 chunk前端就渲染一个 chunk。Vue 里可以用fetchReadableStream来实现const response await fetch(/api/chat, { method: POST, body: JSON.stringify({ query }) }) const reader response.body.getReader() const decoder new TextDecoder() while (true) { const { done, value } await reader.read() if (done) break const chunk decoder.decode(value) // 解析 SSE 格式提取 data 字段 const lines chunk.split(\n) for (const line of lines) { if (line.startsWith(data: )) { const text line.slice(6) appendToDisplay(text) } } }这种方式的好处是真正的实时后端推多快前端就显示多快。但要注意处理粘包问题——多个 SSE 消息可能被合并到一个 chunk 里你需要按\n\n分割。3.2 处理 Markdown 流式渲染的坑AI 返回的内容通常是 Markdown 格式包含代码块、列表、表格等。流式渲染 Markdown 有个经典问题代码块还没闭合的时候Markdown 解析器会把后面的内容全部当成代码。比如模型正在输出python def hello():这时候 Markdown 解析器看到三个反引号就认为代码块开始了但结束的反引号还没来于是后面的所有内容都被渲染成代码。等结束反引号到了页面又会突然重排体验很差。我的解决办法是在流式渲染阶段先做一次“预判”如果检测到未闭合的代码块就暂时不渲染 Markdown而是用纯文本展示等代码块闭合后再切换成 Markdown 渲染。虽然实现麻烦一点但效果确实好很多。3.3 自动滚动与用户打断的平衡流式输出的时候页面内容不断增长你需要自动滚动到底部让用户看到最新内容。但如果用户手动往上滚去看历史消息你还强制滚动到底部用户会想砸键盘。正确的做法是监听滚动事件判断用户是否在底部附近。如果在底部就自动滚动如果用户主动往上滚了就暂停自动滚动并显示一个“回到底部”的按钮。这个细节虽小但直接影响用户体验。4. 后端流式接口封装从 FastAPI 到 LangChain 的完整链路4.1 FastAPI 的 StreamingResponse 实战FastAPI 提供了StreamingResponse来支持流式输出。基本用法是传入一个生成器函数from fastapi import FastAPI from fastapi.responses import StreamingResponse app FastAPI() async def event_generator(query: str): async for chunk in llm.astream(query): yield fdata: {chunk.content}\n\n app.post(/api/chat) async def chat(query: str): return StreamingResponse( event_generator(query), media_typetext/event-stream, headers{ Cache-Control: no-cache, X-Accel-Buffering: no } )这里有几个关键点。第一生成器必须是异步的否则会阻塞事件循环。第二media_type必须设为text/event-stream否则浏览器不会按 SSE 处理。第三X-Accel-Buffering: no是给 Nginx 看的告诉它不要缓冲这个响应。4.2 LangChain 的 astream 与 astream_eventsLangChain 提供了两个流式接口astream和astream_events。astream返回的是 LLM 输出的 token 流适合简单的对话场景。astream_events则更强大它能返回整个链路上所有事件包括 LLM 开始、LLM 结束、工具调用、工具返回等。如果你只是做简单的对话用astream就够了async for chunk in llm.astream(你好): print(chunk.content, end, flushTrue)但如果你用了 Agent 或者 Chainastream_events才是正确的选择。它的事件类型包括on_chat_model_stream、on_tool_start、on_tool_end等你可以根据事件类型决定往前端推什么内容。比如工具调用的时候你可以推一个“正在查询数据库...”的提示让用户知道系统在干活。4.3 封装一个可复用的 SSE 流式接口在实际项目中我一般会把 SSE 的封装逻辑抽成一个独立的模块统一处理格式拼接、错误捕获和心跳保活import json import asyncio from typing import AsyncGenerator class SSEStreamer: def __init__(self, heartbeat_interval: int 15): self.heartbeat_interval heartbeat_interval async def stream(self, generator: AsyncGenerator) - AsyncGenerator[str, None]: try: async for chunk in generator: yield fdata: {json.dumps({content: chunk})}\n\n except Exception as e: yield fdata: {json.dumps({error: str(e)})}\n\n finally: yield data: [DONE]\n\n心跳保活是很多人忽略的点。如果 LLM 生成速度慢中间可能有十几秒没有数据推送某些代理服务器会认为连接空闲而主动断开。解决办法是每隔一段时间推一个空注释行: heartbeat\n\n保持连接活跃。5. 常见问题与排查技巧实录5.1 流式输出突然中断怎么办这是被问得最多的问题。流式输出跑到一半突然断了前端收到一个不完整的响应。常见原因有三个第一代理服务器超时。Nginx 默认的proxy_read_timeout是 60 秒如果 LLM 生成时间超过这个值连接会被切断。解决办法是调大这个值或者用心跳保活。第二后端异常未捕获。生成器函数里抛了异常但没有 try-except 包裹导致连接直接断开。解决办法是在生成器里加全局异常捕获把错误信息也通过 SSE 推给前端。第三前端读取逻辑有 bug。比如reader.read()返回的 chunk 没有正确处理粘包导致部分数据丢失。解决办法是维护一个缓冲区按\n\n分割完整的 SSE 消息。5.2 JSON 解析失败的排查思路结构化输出解析失败是另一个高频问题。我的排查顺序是这样的先看模型原始输出。把 LLM 返回的原始文本打印出来看看是不是格式不对。常见问题包括模型在 JSON 外面包了一层 Markdown 代码块、JSON 里有尾随逗号、字段名用了中文引号等。再看 prompt 是否清晰。如果你用的是 PydanticOutputParser检查format_instructions是否完整塞进了 prompt。有时候 prompt 太长模型会忽略格式要求。最后看模型能力。有些小模型对 JSON 格式的遵循能力确实差这时候要么换模型要么加 few-shot 示例要么用 output parser 的重试机制。5.3 常见问题速查表问题现象可能原因解决方案流式输出变成一次性返回代理或框架开启了缓冲设置X-Accel-Buffering: no关闭 gzip连接中途断开代理超时或后端异常调大超时时间加心跳保活捕获异常JSON 解析失败模型输出格式不对检查 prompt加 few-shot加重试机制前端显示乱码编码问题或粘包处理不当确保 UTF-8 编码按\n\n分割消息打字机效果卡顿前端渲染频率过高用 requestAnimationFrame 节流渲染5.4 几个我踩过的坑第一个坑在 FastAPI 的生成器里用了同步的time.sleep()导致整个事件循环被阻塞所有请求都卡住。后来改成await asyncio.sleep()才解决。第二个坑前端用EventSource的时候发现无法自定义请求头导致无法传认证 token。后来改用fetchReadableStream才解决。如果你的认证逻辑依赖 header千万别用EventSource。第三个坑LangChain 的astream_events在某些版本里会重复触发事件导致前端收到重复内容。解决办法是升级到最新版本或者在事件处理里加去重逻辑。6. 从对话到 Agent流式输出的进阶玩法6.1 Agent 场景下的流式事件处理当你从简单的 LLM 对话升级到 Agent 的时候流式输出的复杂度会上一个台阶。因为 Agent 不只是生成文本它还会调用工具、查询数据库、执行代码。用户需要知道 Agent 当前在干什么而不是盯着一个空白页面等。astream_events在这里就派上用场了。你可以监听不同的事件类型给用户不同的反馈async for event in agent.astream_events(input, versionv2): kind event[event] if kind on_chat_model_stream: # LLM 正在生成文本 yield event[data][chunk].content elif kind on_tool_start: # 工具开始执行 yield f\n[正在调用工具: {event[name]}]\n elif kind on_tool_end: # 工具执行完成 yield f\n[工具执行完成]\n这样用户就能看到 Agent 的完整思考过程体验比单纯等一个最终答案好得多。6.2 多轮对话中的上下文管理流式输出和多轮对话结合的时候有一个容易被忽略的问题上下文长度。每轮对话都会往历史记录里追加内容几轮之后 token 数就爆了。解决办法是定期做上下文压缩或者只保留最近 N 轮对话。我一般会在后端维护一个会话状态每次请求的时候把历史消息和当前消息一起传给 LLM。但要注意流式输出的时候不能把历史消息也流式推给前端否则用户会看到之前的内容重复出现。正确的做法是只推当前轮次的增量内容。6.3 结构化输出在 Agent 中的应用Agent 调用工具的时候工具的入参通常需要结构化数据。比如一个查询天气的工具需要{city: string, date: string}这样的参数。这时候就可以用with_structured_output来让 LLM 直接生成工具入参避免手动解析文本。LangChain 的 Agent 框架其实已经内置了这个能力你只需要用tool装饰器定义工具框架会自动处理参数解析。但如果你是自己手写 Agent 循环那就需要自己实现结构化输出到工具调用的转换。7. 性能优化与生产环境注意事项7.1 减少首 token 延迟流式输出的体验好坏很大程度上取决于首 token 延迟。用户点下发送之后如果 2 秒内能看到第一个字蹦出来心理上就能接受。如果超过 5 秒还是空白用户就会开始怀疑是不是卡了。减少首 token 延迟的方法有几个。第一用更快的模型比如 GPT-4o-mini 就比 GPT-4 快很多。第二精简 promptprompt 越短模型开始生成的速度越快。第三预热连接提前建立好到模型服务的连接避免每次请求都重新握手。7.2 并发流式请求的资源管理每个 SSE 连接都会占用一个服务端连接和一个协程。如果并发量上来了比如同时有几百个用户在对话服务端资源会吃紧。解决办法是用连接池限制最大并发数超出的请求排队等待。另外要注意Python 的 GIL 在 IO 密集型场景下影响不大但如果你的生成器里有 CPU 密集型操作比如复杂的文本处理会阻塞其他协程。这种操作应该放到线程池里执行。7.3 日志与监控生产环境里流式接口的日志和监控特别重要。你需要记录每个请求的开始时间、首 token 时间、结束时间、token 总数、是否异常等信息。这些数据能帮你发现性能瓶颈和异常情况。我一般会在 SSEStreamer 里加一个计时器记录首 token 延迟和总耗时然后上报到监控系统。如果发现某个时间段首 token 延迟飙升就能及时排查是不是模型服务出了问题。8. 一个完整的实战案例合同信息提取8.1 需求分析与方案设计假设我们要做一个合同信息提取工具。用户上传一份合同 PDF系统自动提取甲方、乙方、合同金额、签署日期等关键信息并以结构化形式展示。这个场景的特点是用户需要看到提取过程哪些字段已经提取到了同时最终结果需要是结构化的 JSON。所以我们需要流式输出 结构化输出的组合方案。8.2 后端实现后端用 FastAPI LangChain 实现。首先定义 Pydantic 模型from pydantic import BaseModel, Field class ContractInfo(BaseModel): party_a: str Field(description甲方名称) party_b: str Field(description乙方名称) amount: float Field(description合同金额) sign_date: str Field(description签署日期)然后用with_structured_output让 LLM 直接返回结构化数据。但为了流式展示我们先用普通 LLM 流式输出提取过程等流结束后再调用结构化提取app.post(/api/extract) async def extract_contract(file: UploadFile): text extract_text(file) async def generate(): # 第一阶段流式展示提取过程 async for chunk in llm.astream(f请提取以下合同的关键信息{text}): yield fdata: {json.dumps({type: process, content: chunk.content})}\n\n # 第二阶段结构化提取 structured_llm llm.with_structured_output(ContractInfo) result structured_llm.invoke(f提取合同信息{text}) yield fdata: {json.dumps({type: result, content: result.dict()})}\n\n yield data: [DONE]\n\n return StreamingResponse(generate(), media_typetext/event-stream)8.3 前端展示前端用 Vue 实现监听 SSE 消息根据type字段区分是过程内容还是最终结果。过程内容用打字机效果展示最终结果用表格展示。const eventSource new EventSource(/api/extract) eventSource.onmessage (event) { if (event.data [DONE]) { eventSource.close() return } const data JSON.parse(event.data) if (data.type process) { processText.value data.content } else if (data.type result) { contractInfo.value data.content } }这个方案上线之后用户反馈很好。他们能看到系统在“读”合同而不是干等一个结果。虽然总耗时没变但感知上的等待时间短了很多。9. 一些个人经验与建议流式输出这个东西看起来简单但细节特别多。我从第一个版本到现在前前后后改了七八次才把各种边界情况处理干净。最大的体会是不要等到所有东西都完美了再上线。先做一个能跑的版本让用户用起来然后根据反馈逐步优化。我第一版的时候连心跳保活都没做用户经常遇到连接断开但至少能用。后来慢慢加了心跳、加了异常捕获、加了结构化输出体验才越来越好。另一个建议是多写测试。流式接口的测试比普通接口麻烦因为你要模拟各种异常情况——连接中断、超时、格式错误等。但这些测试能帮你提前发现很多问题避免上线之后被用户投诉。最后如果你也在做类似的东西欢迎交流。这个领域变化很快LangChain 的 API 每隔几个月就变一次保持学习的心态很重要。
