1. 从“打字机”到“结构化”为什么这套组合拳值得死磕做过 AI 应用的人大概都有过这种体验模型明明在后台已经吐字了前端却像卡死一样白屏等半天用户以为程序崩了直接关页面或者反过来字是一个个蹦出来了但你想拿结果去做下一步业务处理时发现拿到手的是一坨没法直接用的字符串还得自己写正则去抠 JSON抠到怀疑人生。这两个问题一个属于流式传输一个属于结构化输出看起来是两件事实际上在真实项目里它们是同一条链路上的上下游。我这次要聊的就是把这条链路从头到尾打通底层用SSEServer-Sent Events做流式推送中间用LangChain做编排和结构化输出约束前端做出顺滑的打字机效果后端保证最终能拿到一份干净可解析的JSON。这套方案解决的核心痛点很明确——既要“快”首字尽快出现用户有反馈又要“准”最终结果可被程序消费而不是只给人看。适合谁来参考如果你正在用 FastAPI、LangChain 搭 AI Agent 类应用前端是 Vue 或者 React并且已经踩过“流式接口封装混乱”“JSON 解析时好时坏”“流到一半断了不知道为啥”这些坑那这篇基本就是给你写的。哪怕你只是刚入门 LangChain想搞清楚stream和astream到底差在哪、结构化输出怎么和流式共存也能顺着往下看我会把每个关键选择背后的“为什么”讲清楚。先说结论性的判断流式和结构化输出不是二选一而是要在同一条流里分阶段处理。流式负责“体验”结构化负责“可用”两者通过合理的缓冲与解析策略共存。下面我按实际搭建顺序一层层拆。2. SSE 流式原理别把它当成“简单的长连接”2.1 SSE 到底是什么和 WebSocket 差在哪SSE 全称 Server-Sent Events本质上是基于 HTTP 的单向、长连接、文本流协议。服务端保持连接不关闭持续往客户端写数据客户端通过EventSource浏览器原生或者 fetch 的流式读取来接收。它的数据格式非常朴素就是纯文本每条消息以data:开头消息之间用两个换行\n\n分隔还可以带event:、id:、retry:这些字段。很多人第一反应是“那它和 WebSocket 有啥区别”。我一般这么解释WebSocket 是双向对讲机SSE 是广播喇叭。AI 对话场景里绝大多数时候是“服务端一直说客户端偶尔发一次请求”这种单向推送用 SSE 就够了而且它有个巨大优势——走的是标准 HTTP天然兼容现有的鉴权、网关、负载均衡体系不用像 WebSocket 那样单独处理升级握手和连接保活。代价是它只能服务端推客户端客户端要发消息得另开一个普通 POST 请求。注意SSE 是文本协议二进制数据比如图片、音频不能直接塞进去需要 base64 编码后再传这会带来约 33% 的体积膨胀做多模态流式时要提前评估。2.2 一条 SSE 消息的完整生命周期我拿一个最典型的 AI 对话流来拆。客户端发起 POST 请求注意原生EventSource只支持 GET所以实际项目里我们通常用 fetch ReadableStream 来手动解析这样才能带 body 和自定义 header。服务端收到后设置响应头Content-Type: text/event-stream Cache-Control: no-cache Connection: keep-alive X-Accel-Buffering: no这里X-Accel-Buffering: no是个关键Nginx 默认会缓冲响应不加这个头你的流会被 Nginx 攒成一大块再发出去打字机效果直接消失。这个坑我踩过不止一次本地测试好好的一上生产就变成“一次性全出来”排查半天才发现是反向代理在缓冲。然后服务端开始逐条写data: {type:token,content:你} data: {type:token,content:好} data: {type:done,content:}每条消息之间必须有两个换行这是协议规定的分隔符少一个都会导致客户端解析错位。最后服务端主动关闭连接或者客户端调用abort()中断。2.3 为什么流式能做出打字机效果打字机效果的本质是把一次完整响应拆成 N 次小响应让浏览器有机会在每次收到数据后立即渲染。这里有个容易被忽略的点即使服务端真的在逐字发如果前端是等整个response.text()读完再渲染那还是白屏。所以前端必须用response.body.getReader()拿到ReadableStream边读边解析边更新 DOM。我用一个生活类比SSE 就像水龙头一滴一滴放水EventSource或 fetch reader 是你手里的杯子你得每接一滴就喝一口渲染一次而不是等杯子满了再喝。很多人只做了“服务端逐字发”忘了“前端逐字读”结果就是没效果。3. LangChain 结构化输出让模型吐出能直接用的 JSON3.1 为什么“让模型输出 JSON”这么难直接跟模型说“请输出 JSON”它大概率会给你好的以下是结果 json {name: 张三, age: 25}希望对你有所帮助这种带前后缀、带 markdown 代码块、甚至字段名拼错的输出程序根本没法直接 JSON.parse。早期大家用正则去抠抠到后面发现模型稍微换个措辞就崩了。LangChain 的结构化输出就是为了解决这个“最后一公里”问题。 它的核心思路有两层**一是用 schema 约束模型输出格式**通过 function calling / tool calling 或者 JSON mode**二是用 Pydantic 做解析和校验**。模型返回的原始字符串会被自动解析成 Pydantic 对象字段类型不对、缺字段都会在解析阶段报错而不是等到业务逻辑里才炸。 ### 3.2 with_structured_output 的三种实现路径 LangChain 里最常用的入口是 with_structured_output它底层会根据模型能力自动选择策略主要有三种 | 策略 | 触发条件 | 优点 | 缺点 | |------|----------|------|------| | Function Calling | 模型支持工具调用 | 约束最强格式最稳 | 部分模型不支持 | | JSON Mode | 模型支持 JSON 输出模式 | 兼容性好 | 仍需 schema 校验 | | Prompt 提示 | 兜底方案 | 通用 | 稳定性最差 | 我实测下来只要模型支持 function calling就优先走这条路格式稳定性明显高一个档次。用 Pydantic 定义 schema 的时候字段描述Field(description...)一定要写清楚这不是给人看的是给模型看的描述越精确模型填错字段的概率越低。 python from pydantic import BaseModel, Field from langchain_openai import ChatOpenAI class PersonInfo(BaseModel): name: str Field(description人物姓名中文全名) age: int Field(description年龄整数未知填 -1) skills: list[str] Field(description技能列表每项为简短名词) llm ChatOpenAI(modelgpt-4o-mini) structured_llm llm.with_structured_output(PersonInfo) result structured_llm.invoke(张三今年25岁会Python和Go) print(result.name, result.age, result.skills)这段代码跑下来result直接就是PersonInfo实例不用你手动解析。这就是结构化输出的价值——把“解析字符串”这件脏活从业务代码里彻底剥离。3.3 结构化输出和流式的天然矛盾问题来了结构化输出要求模型吐完整个 JSON 才能解析而流式要求逐字往外发。这俩看起来是冲突的。如果你直接对with_structured_output的结果调stream很多模型会直接报错或者退化成非流式。这个矛盾的解法是分层处理底层模型仍然流式输出 token但我们在应用层做缓冲把 token 拼成完整字符串后再做结构化解析。或者更高级一点用 LangChain 的astream_events监听事件流在流式过程中同时做增量解析。下面实操部分我会详细讲这两种方案怎么选。4. 实操全流程从 FastAPI 后端到 Vue 前端的完整链路4.1 后端FastAPI 封装 SSE 流式接口先上后端骨架。我用 FastAPI 的StreamingResponse来做核心是把 LangChain 的异步流转换成 SSE 格式的字节流。from fastapi import FastAPI from fastapi.responses import StreamingResponse from langchain_openai import ChatOpenAI import json app FastAPI() llm ChatOpenAI(modelgpt-4o-mini, streamingTrue) async def sse_generator(prompt: str): try: async for chunk in llm.astream(prompt): if chunk.content: payload {type: token, content: chunk.content} yield fdata: {json.dumps(payload, ensure_asciiFalse)}\n\n yield fdata: {json.dumps({type: done})}\n\n except Exception as e: err {type: error, message: str(e)} yield fdata: {json.dumps(err, ensure_asciiFalse)}\n\n app.post(/chat/stream) async def chat_stream(prompt: str): return StreamingResponse( sse_generator(prompt), media_typetext/event-stream, headers{ Cache-Control: no-cache, X-Accel-Buffering: no, }, )几个关键点解释一下。ensure_asciiFalse是为了让中文正常显示不然会变成\u4f60\u597d这种转义虽然前端解析后也对但调试时看着难受。astream是异步流配合 FastAPI 的异步响应不会阻塞事件循环。异常必须捕获并作为一条error类型消息发出去否则连接会直接断前端只能看到一个莫名其妙的网络错误。实操心得StreamingResponse里如果生成器抛异常且没捕获客户端收到的是连接中断而不是错误信息。所以生成器内部一定要 try/except 包住把错误也当成一条正常消息发出去。4.2 前端fetch 流式读取与打字机渲染原生EventSource不支持 POST所以实际项目我基本都用 fetch reader。Vue 里大概长这样async function streamChat(prompt, onToken, onDone) { const resp await fetch(/chat/stream?prompt encodeURIComponent(prompt), { method: POST, headers: { Accept: text/event-stream } }) const reader resp.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 parts buffer.split(\n\n) buffer parts.pop() // 最后一段可能不完整留到下次 for (const part of parts) { const line part.trim() if (!line.startsWith(data:)) continue const data JSON.parse(line.slice(5).trim()) if (data.type token) onToken(data.content) else if (data.type done) onDone() } } }这里有个极其容易踩的坑buffer.split(\n\n)之后最后一段很可能是半条消息因为网络分片不一定按消息边界切必须pop()出来留到下一轮拼接。我见过太多人直接遍历所有 parts结果偶尔报 JSON 解析错误就是因为把半条消息也拿去 parse 了。decoder.decode(value, { stream: true })的stream: true也很关键它保证多字节的 UTF-8 字符比如中文不会被从中间截断导致乱码。打字机效果本身就是每次onToken时把内容 append 到响应式变量上Vue 会自动触发重渲染。如果想要更细腻的“逐字”感可以在 onToken 里再做一层字符级队列用requestAnimationFrame控制节奏避免模型一次吐一大段时视觉上太突兀。4.3 结构化输出与流式的融合方案现在把结构化输出接进来。有两种主流做法我分别说适用场景。方案 A先流式展示结束后再结构化解析。适合“展示为主、后续处理为辅”的场景比如聊天摘要。做法是流式阶段正常吐 token 给前端同时后端把完整文本攒起来流结束后再调一次结构化解析或者用同一个模型再跑一遍。缺点是多一次调用成本和延迟都增加。方案 B流式过程中增量解析。适合“边流边用”的场景比如实时抽取实体。LangChain 的astream_events可以监听on_chat_model_stream事件拿到每个 token 后自己维护一个缓冲区尝试增量解析 JSON。这个方案更优雅但实现复杂因为 JSON 在没闭合前是不合法的你得用增量 JSON 解析器比如ijson或者自己写状态机。from langchain_core.prompts import ChatPromptTemplate prompt ChatPromptTemplate.from_messages([ (system, 你是一个信息抽取助手只输出 JSON。), (human, {input}) ]) chain prompt | llm.with_structured_output(PersonInfo) async def structured_stream(user_input: str): buffer async for event in chain.astream_events({input: user_input}, versionv2): if event[event] on_chat_model_stream: token event[data][chunk].content if token: buffer token yield {type: token, content: token} # 流结束后 buffer 是完整 JSON 字符串 parsed PersonInfo.model_validate_json(buffer) yield {type: result, data: parsed.model_dump()}注意astream_events的version参数必须显式传不同版本事件结构有差异不传可能拿不到预期事件。这个我在升级 LangChain 版本时被坑过一次事件名从on_llm_stream变成了on_chat_model_stream找了好久。5. 常见问题与排查技巧实录5.1 流到一半断了idle timeout 到底是谁的锅热搜里那个stream disconnected before completion: idle timeout waiting for sse是高频问题。这个报错通常来自三个地方反向代理的空闲超时、网关的读超时、客户端自己的超时设置。排查顺序我一般这么走先看 Nginx 配置proxy_read_timeout默认 60 秒如果模型思考时间长比如推理模型60 秒内没吐任何字节连接就被掐了。改成 300 秒甚至更长。然后看云厂商的负载均衡很多默认空闲超时也是 60 秒。最后看客户端 fetch 有没有设AbortController的超时。还有一个隐蔽原因模型在“思考”阶段不吐 token。有些推理模型会先内部推理几十秒再输出这期间 SSE 连接上没有任何数据流动就会被判定为空闲。解法是服务端定期发心跳比如每 15 秒发一条: ping\n\n注释行保持连接活跃。现象可能原因排查方法60 秒左右必断Nginx/网关空闲超时查 proxy_read_timeout首字迟迟不来后断模型思考期无数据加心跳注释行偶发 JSON 解析失败消息分片未拼接检查 buffer 处理逻辑中文乱码未用 stream 模式解码TextDecoder 加 stream:true生产环境无打字机效果代理缓冲加 X-Accel-Buffering:no5.2 结构化输出解析失败的几种典型情况即使走了 function calling偶尔还是会解析失败。我总结了几类模型返回了 schema 里没定义的额外字段Pydantic 默认会忽略但如果开了extraforbid就会报错字段类型不匹配比如该是 int 的返回了字符串 25必填字段缺失。应对策略是给 Pydantic 模型加合理的默认值和容错比如age: int Field(default-1)并且在 prompt 里明确“未知信息填默认值不要编造”。另一个高频问题是流式和非流式结果不一致。同一个 prompt非流式调用结构化输出正常流式就崩。原因是流式模式下模型可能不会触发 function calling而是直接吐文本。这时候要么放弃流式结构化要么用方案 B 自己解析。我的经验是对格式要求极高的场景宁可牺牲一点流式体验也要保证结构化稳定对体验要求高的场景流式展示和结构化解析分成两次调用。5.3 封装 SSE 调用逻辑时的几个避坑点封装成通用工具函数时有几个点必须处理。第一是中断处理用户点了“停止生成”前端要调reader.cancel()或者AbortController.abort()后端要能感知到连接断开并停止模型调用不然白白烧 token。第二是重连SSE 原生支持retry字段自动重连但 AI 对话场景重连会导致重复内容所以一般禁用自动重连由业务层决定。第三是多路复用一个页面可能同时有多个流比如多个 Agent 并行要给每个流分配唯一 id前端按 id 分发别混在一起。实操心得后端检测客户端断开可以用await request.is_disconnected()轮询或者在生成器里捕获asyncio.CancelledError。后者更可靠因为连接断开时 Starlette 会取消对应的任务。6. 进阶把流式结构化输出用到 Agent 场景6.1 Agent 中间步骤的流式可视化现在做 Agent 应用的人越来越多基于 LangGraph 或者类似框架的多步推理用户最想知道的是“它现在在干嘛”。这时候 SSE 就不只是吐最终答案还要吐中间步骤正在调用哪个工具、工具返回了什么、下一步准备做什么。做法是在 Agent 的每个节点里往同一个流里写事件前端根据type字段渲染不同的 UI 组件思考中、工具调用、结果。这里的关键是事件类型设计要提前规划好别等到前端写一半发现类型不够用。我一般会定义token、tool_start、tool_end、step、done、error这几类每类带自己的 payload 结构。这样前端可以做成一个可扩展的事件渲染器加新类型不用改核心逻辑。6.2 结构化输出在 Agent 决策中的应用Agent 的每一步决策其实都可以用结构化输出来约束。比如让模型决定“下一步调用哪个工具”与其让它输出自然语言再解析不如直接定义一个ToolDecision的 Pydantic 模型字段是tool_name和arguments。这样决策结果直接可用不用解析。LangChain 的 Agent 内部其实也是这么做的理解这一点你自己手写 Agent 循环时就能少走弯路。6.3 性能与成本的平衡取舍流式 结构化这套组合性能开销主要在结构化解析那一步。如果每次都调两次模型一次流式展示、一次结构化成本直接翻倍。我的优化思路是能一次调用解决的绝不用两次。用astream_events在流式过程中攒完整文本流结束后本地做结构化解析如果模型输出本身就是 JSON 格式这样只有一次模型调用。只有当模型输出格式不可控时才退化成两次调用。另外结构化输出的 schema 别设计得太复杂嵌套层级越深模型填错的概率越高。我一般控制在两层以内复杂结构拆成多次调用。这是用稳定性换来的经验不是理论推导。7. 我踩过的坑和最后想说的这套方案我从头搭到尾前后重构过三次。第一次是没处理消息分片线上偶发 JSON 解析错误查了两天才定位到 buffer 拼接问题。第二次是没加X-Accel-Buffering: no本地好好的上线打字机效果消失被产品追着问。第三次是结构化输出和流式硬凑在一起模型时不时不触发 function calling最后改成流式展示和结构化解析分离才稳定下来。如果让我给刚上手的人一句建议先把 SSE 流式跑通确认打字机效果没问题再叠加结构化输出。别一上来就追求“流式结构化一步到位”那个复杂度对新手不友好而且很多场景根本不需要。分阶段来每一步都验证清楚比什么都强。最后分享一个小技巧调试 SSE 的时候用curl -N直接看原始流比在浏览器里看 Network 面板清楚得多。curl -N -X POST http://localhost:8000/chat/stream?prompt你好-N关闭缓冲能实时看到每条消息格式对不对一眼就知道。这个命令帮我省了无数调试时间。