上周调一个基于 FastAPILangChain 的 Agent 接口时前端又抛出来一行非常“经典”的报错stream disconnected before completion: idle timeout waiting for sse。这行字我这两年见过太多几乎每个做 AI Agent 的团队都会在某个版本里撞上它。表面看是网关把 SSE 长连接掐了实际是流式传输和结构化输出这对需求一直没有被放到同一张设计图里。这篇文章不扯概念只讲我实际跑过的链路LangChain 里三大 OutputParser 怎么选ToolCall 方案为什么能省掉一半解析痛苦以及最后如何用 FastAPI 把 SSE 流口子封装好喂给 Vue/Python 的客户端收结构化结果。刚入门 LangChain 的同学可以通读做 LangGraph / Deep Agents 这类长链路项目的同行更建议直接跳到 ToolCall 和事件协议两段。1. 那个让前端白屏的 idle timeout背后是两种“数据节奏”在打架1.1 流式是“边写边说”结构化是“写完再交”先回忆一下 SSE 到底做了什么。SSE 是一种服务器向浏览器单向推送的 HTTP 技术连接建立后服务器可以不断往同一个响应里追加内容客户端不用反复发请求。它适合做 LLM token 流因为模型每次吐一个字服务端就往连接里塞一段数据前端拿到就渲染阅读体验比干等十几秒再一次性返回要好得多。但问题在于SSE 本身并不知道“模型正在思考”和“模型已经死了”之间的区别。它对连接健康状态的判断只有一条这一段时间内有没有数据包经过。如果服务端超过一定时间没写任何字节代理层、网关或中间负载均衡就会认为这条连接已经空转主动断开客户端就会看到类似idle timeout waiting for sse的错误。很多团队一开始怀疑是代码 bug实际只是把“流式”和“结构化”这两件事的节奏搞混了。流式是“边写边说”模型每生成一个 token 就推给前端结构化输出是“写完再交”模型把完整答案组织好交给 OutputParser 做校验和转换。这两种节奏天然不同token 流是高频小包结构化输出是低频大包。如果你从头到尾只做了 token 流没有考虑模型在长链路里可能长时间不出包那么空闲超时几乎是必然的。1.2 流式长思考期没有任何数据SSE 空闲超时必然发生举一个我实际遇到的场景。项目里接了一个 LangChain 的 Agent用户在对话框发一句“帮我把昨天所有未处理的工单汇总成表格”。这句话表面简单但链路里有几件事要做先走工具调用查工单可能还要等数据库慢查询再让模型组织摘要最后生成 Markdown。整个过程模型大部分时间都在“想”并不输出 token。前端那边EventSource已经连上了页面也在等待。可是从网关的视角看这条 SSE 连接已经几十秒没有生产数据如果代理默认空闲超时是 60 秒到点一刀切。客户端日志里就留下stream disconnected before completion这种模棱两可的记录既不像超时错误也不像服务端异常排查起来非常迷惑。如果项目又用了 LangGraph 这类框架情况会更复杂。LangGraph 把一次任务拆成多个节点每个节点内部都可能调用模型、工具或外部服务。你对外暴露一条 SSE 接口时实际上是把一条多级流水线压在一个 HTTP 响应里。某个节点在等人审批、某个工具在重试、某个模型在规划都可能产生几十秒的空窗。我之前有一个项目还加入了类似 Agent Inbox 的人工确认环节用户点确认前信号一直挂起那个阶段如果不发心跳前端必断。所以写这篇文章的第一个目的就是先把这条认知掰正SSE 流式接口不能只关心“有没有 token 出来”还必须主动管理“没有 token 时的连接保活”。后面第 3 节会讲具体手段。2. LangChain 三大 OutputParser 横评Pydantic、Structured、JSON 的真实差距LangChain 文档里 OutputParser 的数量远不止三个有逗号分隔列表、XML、CSV、dataclass 等等但实际生产项目中我反复用到的其实就是 PydanticOutputParser、StructuredOutputParser 和 JsonOutputParser 这三个。它们都不会改变模型本身的生成能力区别在于用什么样的提示词约束模型输出以及拿到文本后怎么转换成 Python 对象。选错一个轻则多消耗 token重则在生产环境频繁报格式错误。2.1 PydanticOutputParser类型安全代价是提示词膨胀PydanticOutputParser 是 LangChain 里最“正经”的解析器。你用 Pydantic 定义一个模型它会把 schema 转换成大段格式说明附加到 Prompt 里要求 LLM 返回符合 schema 的 JSON然后解析成 Pydantic 对象。我们常用的代码大概是这样from pydantic import BaseModel, Field from langchain_core.output_parsers import PydanticOutputParser class WeatherReport(BaseModel): city: str Field(description城市名) temperature: float Field(description当前温度单位摄氏度) condition: str Field(description天气状况如晴、多云、雨) advice: str Field(description一条简短出行建议) parser PydanticOutputParser(pydantic_objectWeatherReport)然后你在构造 Prompt 时一定要把parser.get_format_instructions()塞进去否则模型根本不知道要返回什么结构。注意这里有个很容易被忽略的细节get_format_instructions()生成的说明可能会占几百 token模型每次回答都要先把这堆规则“读一遍”。拿小模型跑的时候规则越多越容易顾此失彼经常出现把 field 名改写成自然语言的情况。它最大的优点是类型安全。解析成功后你拿到的就是WeatherReport对象字段类型、默认值、描述都有保障。我一般只在 schema 明确、字段不算太多、对下游强耦合的场景用它比如调用数据库写入、触发第三方 API。缺点也很直接如果 LLM 生成的不是合法 JSONPydanticOutputParser 会抛OutputParserException不会帮你修补。生产环境必须有配套的重试或修复逻辑。2.2 StructuredOutputParser轻量多字段场景还行字段一多就露怯StructuredOutputParser 的定位比 Pydantic 轻很多。它不要求你定义完整的 Pydantic 模型只需要用ResponseSchema列出字段名和描述from langchain_core.output_parsers import StructuredOutputParser, ResponseSchema response_schemas [ ResponseSchema(namecity, description城市名), ResponseSchema(nametemperature, description当前温度单位摄氏度), ResponseSchema(namecondition, description天气状况), ResponseSchema(nameadvice, description出行建议), ] parser StructuredOutputParser.from_response_schemas(response_schemas)在 LangChain 旧版本里它生成的格式说明是一段“你必须输出符合 JSON 格式的结果”的文本并列出 key。模型需要把这些 key 完整保留在输出 JSON 里。这种方式的优点是轻、快字段少时效果不错缺点是嵌套结构很难表达一旦你想让某个字段是对象或数组用ResponseSchema描述起来非常别扭。还有一个实际坑是它对 JSON 里多余字段的处理很宽松模型偶尔会额外返回一个 key你的代码如果直接按固定 key 去取值可能踩到KeyError。我现在更倾向于把它用在一些内部小工具上比如让模型返回“是/否/原因”这种三字段决策而不是复杂业务对象。字段一旦超过五六个我会直接切到 PydanticOutputParser 或 ToolCall。2.3 JsonOutputParser没有强 schema但对于自由 JSON 够用JsonOutputParser 是三个里最“自由”的一个。它可以传入一个 Pydantic 模型作为期望结构但核心行为非常简单把模型输出的 JSON 文本转成 Python 的 dict 或对象。from langchain_core.output_parsers import JsonOutputParser parser JsonOutputParser(pydantic_objectWeatherReport)用它的前提是你已经通过 Prompt 把期望结构说清楚了它不会像 PydanticOutputParser 那样自动生成 format instructions也不会对字段做严格的 Pydantic 校验。它适合两类场景一类是模型本身就比较强不给格式说明也能输出正确 JSON另一类是你要的结果比较动态今天返回两个字段明天返回四个字段用固定 Pydantic 模型反而绑手绑脚。不要把它当成银弹。它实际返回的结果可能只是一个 dict字段缺失、类型错误都需要你手动处理。如果你在 Prompt 里没有写明“必须输出 JSON”模型给你一段带前缀的自然语言JsonOutputParser 也有一定概率解析失败。我通常在它外面再包一层OutputFixingParser让解析失败时把错误信息回喂给模型让它重新修一次。这个包装对线上场景很管用但会额外消耗一次模型调用需要自己在成本和成功率之间做权衡。2.4 三大 parser 对比与选型表直接给结论下面这张表方便大家抄作业维度PydanticOutputParserStructuredOutputParserJsonOutputParserSchema 定义方式Pydantic 模型ResponseSchema 列表不强制也可传 Pydantic自动生成格式说明会会不会类型校验强度强弱弱偏 dict 转换嵌套结构支持好差一般看模型发挥推荐场景强业务耦合、下游入库少量字段的轻量决策自由格式 JSON、快速原型如果你只能记一句话下游代码需要强约束时用 PydanticOutputParser只想快速拿个 dict 时用 JsonOutputParserStructuredOutputParser 适合那种字段少到不会出错的简单场景。ToolCall 方案出现后很多 PydanticOutputParser 的用法都可以被替代但并没有完全淘汰——模型不一定会调用工具有些需求本质上仍需要“自然语言回答结构化字段”同时输出。3. 把 OutputParser 用进 SSE 流式的三条自救路径在纯同步链路里OutputParser 的用法非常简单模型生成完整文本解析器解析返回对象。但一旦接上 SSE事情就变了。你总不能等模型把完整 JSON 吐完才开始流式因为那样前端会长时间空白。下面是我实测可用的三条路径。3.1 增量解析partial JSON 也能解析LLM 流式输出时很可能前一个 chunk 是{city: 上下一个 chunk 才是海}。如果每收到一个 chunk 就执行json.loads几乎必然失败。更好的方式是维护一个 buffer每次追加后尝试用parse_partial_json解析import json try: from langchain_core.utils.json import parse_partial_json except ImportError: def parse_partial_json(s): # langchain 不同版本入口不同找不到就给一个轻量实现 return json.loads(s, strictFalse)注意parse_partial_json并不是万能的。它能处理末尾缺少括号、缺少引号这类情况但如果模型输出结构改变比如漏了一个 key它同样失败。增量解析的目的不是替代最终解析而是让你在流式过程中提前拿到大致的结构把它当作前端预览数据。真正的最终对象还是要等完整输出后再用正式 OutputParser 跑一遍。我在实际项目里的做法是后端维护一个stream_buffer每收到一段 token 就追加并尝试解析。能解析出部分字段就先通过 SSE 推一个类型为partial的事件给前端解析失败也不慌张继续等待后续 chunk。这样即使中间被网关掐断前端至少渲染出了部分内容用户不会面对一个纯白屏。3.2 解析失败时用修复链路兜底而不是直接报错OutputParser 的报错信息对终端用户没有任何意义。用户只看到“接口内部错误”你则需要在日志里翻半天。更实际的办法是给解析套一层修复链路。第一层是用OutputFixingParser。它内部会拿着解析失败的文本和错误信息让模型重新生成一份符合 schema 的输出from langchain_core.output_parsers import OutputFixingParser parser PydanticOutputParser(pydantic_objectWeatherReport) fix_parser OutputFixingParser.from_llm(parserparser, llmllm)第二层是业务兜底。如果修复仍然失败不要把整个流式接口返回 500。我的经验是先保证 token 流正常发完前端该展示的展示然后在done事件里附一个structured: null和失败原因。前端拿到后可以降级为纯文本渲染至少用户还能看到模型说了什么而不是整体崩溃。这里还要提醒一下修复链路会调用一次模型如果流式通道本身已经因为空闲超时断掉了修复结果没地方送。所以修复动作最好放在后端完成之后再尝试把事件写回连接如果发现连接已经关闭就返回一个长轮询补拉接口前端主动来取。3.3 空闲超时的工程解法心跳、事件格式与超时缓存既然空闲超时的本质是“没有数据包经过”最直接的解法就是定期发心跳包。SSE 协议里有一类以冒号开头的注释行专门用来探测连接不会触发前端的数据解析。在 FastAPI 的异步生成器里可以这样async def event_generator(): count 0 while True: token await model_stream.get() # 阻塞等待新 token if token: yield fdata: {json.dumps({type: token, content: token}, ensure_asciiFalse)}\n\n count 0 else: count 1 if count 15: # 每 15 秒没有 token 就发心跳 yield : keep-alive\n\n count 0这个做法的意图是如果模型有输出立刻推给前端如果模型在思考就靠心跳维持连接。网关看到的是“这个响应一直有数据在流动”不会再判定为空闲。心跳间隔我建议设置在网关空闲超时的一半以内比如代理超时 60 秒心跳就 20 到 30 秒发一次。另外还要注意 HTTP 层代理缓冲。很多反向代理默认会缓冲响应直到攒够一定字节才发给客户端这会让 SSE 实时性大打折扣。接口这边可以主动加两个 headerheaders { Cache-Control: no-cache, X-Accel-Buffering: no, Connection: keep-alive, }X-Accel-Buffering是 Nginx 专有的不设置的话Nginx 可能把 SSE 存在缓冲里导致前端迟迟收不到数据。这已经是我见过第 N 次“报错排查了半天结果是缓冲吞包”的情况了。4. ToolCall让模型“原生”产出结构化参数比解析输出更省事如果说 OutputParser 是“让模型写作文再从作文里归纳出要点”那 ToolCall 就是“让模型直接填一张表格”。同样是拿到结构化数据ToolCall 走的路径完全不同可靠性也高出不少。4.1 ToolCall 与文本解析的本质差异调用 OpenAI、Anthropic 这类大模型 API 时如果给模型绑定工具模型并不是把工具参数写在正文里而是放进一个专门的tool_calls字段。它在生成层就已经是结构化的 JSON 参数不经过“文本输出-再解析”的折腾。在 LangChain 里绑定工具的典型写法是把 Pydantic schema 直接传给bind_toolsfrom langchain_openai import ChatOpenAI model ChatOpenAI(modelgpt-4o-mini, temperature0) model_with_tools model.bind_tools([WeatherReport]) resp model_with_tools.invoke(上海今天天气怎么样) print(resp.tool_calls)resp.tool_calls是一个列表每个元素包含工具名称、参数 dict 和调用 id。正因为工具名和参数是模型 API 主动输出的你不需要在 Prompt 里写“必须以 JSON 格式输出”token 消耗更少格式出错的概率也更低。4.2 LangChain 中 bind_tools 与 streaming 实战流式场景下ToolCall 的优势更明显。LangChain 的AIMessageChunk对象上有一个tool_call_chunks字段专门承载增量参数。注意这里的参数是按片段返回的你不能直接拿去json.loads要先把同一 index 下的args片段拼起来async def stream_tool_call(prompt: str): args_by_index {} names_by_index {} async for chunk in model_with_tools.astream(prompt): for tc in chunk.tool_call_chunks or []: idx tc[index] args_by_index.setdefault(idx, ) names_by_index.setdefault(idx, ) args_by_index[idx] tc[args] or names_by_index[idx] tc[name] or if chunk.content: yield {type: token, content: chunk.content} # 流结束后拼接出的 args 才是完整 JSON for idx, raw_args in args_by_index.items(): yield {type: tool_call, name: names_by_index[idx], args_json: raw_args}这段代码的核心思路是不阻塞等待完整tool_calls而是先让模型正文 token 继续流式到前端同时在后端累积工具参数。等流结束后你手里已经有一份完整的工具调用参数可以立刻执行真正的工具函数再返回工具结果。如果不想写这么多细节其实 LangChain 还提供了一个with_structured_output方法专门服务于“我要结构化输出”这个需求structured_model model.with_structured_output(WeatherReport) result await structured_model.ainvoke(上海今天天气怎么样)它的底层大多还是走函数调用好处是你只需要关心最终WeatherReport对象坏处是在流式事件里默认拿不到中间过程。如果你想做“先流式展示 token再给结构化结果”建议还是用bind_tools自己控制。4.3 ToolCall 与 OutputParser 的选型边界我在这两种方案之间来回切换过几次总结下来的边界是这样的。OutputParser 适合“回答问题 抽取字段”的场景。比如用户问“上海天气怎么样”你既想让模型用一句自然语言回答“今天 22 度多云”又想把城市、温度、天气状况抽出来存库。这种情况下 ToolCall 也可以做但你需要额外约定一个“不被渲染进聊天区”的工具参数前端要判断哪部分是正文、哪部分是工具调用稍微绕一点。ToolCall 更适合“让模型替你去调用函数”。比如用户说“订一张明天下午去杭州的高铁票”你需要拿到日期、时间、出发地、目的地、座位类型这些参数才能去调用订票接口。用 OutputParser 解析这些参数也不是不行但模型自己生成的自然语言里经常夹带解释比如“明天下午”可能被翻译成“明天 14:00”稍有偏差解析就断。ToolCall 直接逼着模型生成{date: ..., time: ...}语义更干净。一句话总结如果结构化数据是给程序用的优先 ToolCall如果结构化数据只是自然语言回答的附属品用 OutputParser 更符合直觉。4.4 ToolCall 流式场景下要注意的边界第一不是所有模型在流式时都会返回tool_call_chunks。有些经由第三方网关封装的老模型只有一个大的 AIMessageChunktool_call_chunks为空但resp.tool_calls在最终结果里有。如果发现这种情况建议降级为“先做完推理再一次性输出结构化对象”不要强求流式工具参数。第二工具调用可能不止一个。LLM 在一次请求里返回多个 tool_call 是很正常的尤其是“帮我同时查上海和北京天气”这类多实体需求。上面代码里用index区分不同调用别只取第 0 个就完事。第三工具参数拼接好之后一定要用 Pydantic 再做一次校验report WeatherReport(**json.loads(raw_args))如果模型生成了多余字段或类型错误这里会非常明确地告诉你哪里不对。别省这一步否则脏数据会直接进入你的业务逻辑。5. 完整链路封装FastAPI 发 SSELangChain 算流式Vue/Python 收结果前面几节分别讲了 OutputParser、增量解析和 ToolCall现在把它们合成一条可运行的全链路。前端是 Vue后端是 FastAPI中间用 SSE 在每次 agent 执行中同时传递 token 和最终结构化结果。5.1 后端StreamingResponse SSE 事件协议FastAPI 里最直接的方式是返回一个StreamingResponsemedia type 设为text/event-stream。下面是一个最小后端模型from fastapi import FastAPI from fastapi.responses import StreamingResponse import asyncio, json app FastAPI() async def event_generator(prompt: str): async for event in stream_agent(prompt): if event[type] token: yield fdata: {json.dumps(event, ensure_asciiFalse)}\n\n elif event[type] tool_call: yield fdata: {json.dumps(event, ensure_asciiFalse)}\n\n # 如果一直没有新内容由 stream_agent 内部保证心跳 await asyncio.sleep(0) app.post(/chat/stream) async def chat_stream(payload: dict): return StreamingResponse( event_generator(payload[prompt]), media_typetext/event-stream, headers{ Cache-Control: no-cache, X-Accel-Buffering: no, Connection: keep-alive, }, )stream_agent内部可以接 LangGraph 的节点也可以接普通的prompt | model | parser链。重要的是统一事件协议typetoken表示普通正文流typetool_call表示工具调用参数typedone表示最终结构化结果。三者并存前端才既能展示打字机效果又能在最后拿到完整对象。这里提醒一句不要把asyncio.sleep(0)省掉。在同步阻塞的链路上一个长耗时工具调用会把整个事件循环卡死心跳也不会发。凡是工具调用这类慢操作都要用asyncio.to_thread或run_in_executor包一层不能直接阻塞在 async 函数里。5.2 前端 VueEventSource 的不足与 readable stream 方案原生EventSource只能发 GET 请求没法带 JSON body也没办法自定义请求头。很多项目为了图省事把对模型的调用改成了 GET query param这种做法在 prompt 很短时还能用prompt 一长就非常尴尬。我在 Vue 项目里更推荐直接用 fetch ReadableStream自己解析 SSE 分帧。async function requestStream(prompt) { const resp await fetch(/chat/stream, { method: POST, headers: { Content-Type: application/json }, body: JSON.stringify({ prompt }), }); const reader resp.body.getReader(); const decoder new TextDecoder(); let buffer ; while (true) { const { value, done } 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 raw part.slice(5).trim(); if (!raw) continue; const event JSON.parse(raw); handleStreamEvent(event); } } }这种写法把判断协议、连接生命周期、心跳都交给你自己控制。优点是 POST 请求随便传 body断开后可以自己决定多久重连不需要依赖浏览器内置行为。缺点是代码量比EventSource多不少。如果你不想手写可以用vue-sse这类库但底层依然是控制 ReadableStream库只是帮你封装好了自动重连和事件分流。5.3 Python 客户端httpx SSE 解析与超时控制除了浏览器我们经常还需要一个 Python 脚本去调用同一个 SSE 接口用途可能是测试、后端对后端的调用、或者做数据回流。用 httpx 的流式接口比较省事import json import httpx with httpx.stream( POST, http://localhost:8000/chat/stream, json{prompt: 上海天气怎么样}, timeouthttpx.Timeout(connect30.0, read120.0, write30.0, pool30.0), ) as response: for line in response.iter_lines(): if not line.startswith(data:): continue payload json.loads(line[5:].strip()) print(payload)这里我把read超时设到了 120 秒并且依赖服务端心跳。如果不设这个值httpx 默认会在一定时间内等不到数据就抛超时。注意这个超时不是越短越好如果读超时小于模型单次推理可能耗时的上限即使服务端一直很健康客户端也会自己把自己断开。iter_lines()是按行读的SSE 事件之间由空行分隔但这种写法要求服务端每条事件都独占一行data:。如果一条事件被拆成多行数据你可以进一步用sseclient库来处理标准 SSE 分帧。我自己的偏好是先统一服务端协议保证每条事件都是单行 JSON这样客户端代码能保持最简。5.4 端到端事件格式约定让结构化输出和流式 token 共存全链路里最容易被忽略的是“事件协议约定”。我建议所有事件都长成一个统一结构{type: token, content: 上海} {type: token, content: 今天} {type: tool_call, name: WeatherReport, args_json: {\city\: \上海\} {type: done, data: {city: 上海, temperature: 22, condition: 多云}}前端按type分流token直接追加到对话气泡tool_call用来画一个“正在调用工具”的组件done触发业务状态更新比如把表单字段填进去。所有事件里除了token需要即时渲染其它都是后置处理就算中途丢了几帧用户在体验上也不会发现明显问题。我之前踩过一个坑把结构化结果也当作 token 推给前端结果前端把 JSON 花括号直接渲染到聊天框里用户看到一大段 “{city...}”又丑又难用。所以事件协议一定要在设计阶段就分好前端不做“猜测式解析”。真实跑完这套链路之后最深的体会是OutputParser 和 ToolCall 并不是对立的它们服务的是两种不同的输出目标。SSE 真正的难点也不在协议本身而在于你要同时兼顾“实时的阅读体验”和“最终的程序可用性”。如果你是第一次做 LangChain 流式接口建议先把第 3 节的心跳和第 2 节的 parser 选型跑通再考虑 ToolCall 和 LangGraph 节点编排。最后再分享一个小技巧上线前用 Python 脚本模拟一次 120 秒无输出的调用如果连接没断、前端能正常收到 done 事件再聊性能优化的事。