企业官网建设流程全解析

上周调一个基于 FastAPI＋LangChain 的 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_object=WeatherReport)

然后你在构造 Prompt 时一定要把parser.get_format_instructions()塞进去，否则模型根本不知道要返回什么结构。注意这里有个很容易被忽略的细节：get_format_instructions()生成的说明可能会占几百 token，模型每次回答都要先把这堆规则“读一遍”。拿小模型跑的时候，规则越多越容易顾此失彼，经常出现把 field 名改写成自然语言的情况。

它最大的优点是类型安全。解析成功后你拿到的就是WeatherReport对象，字段类型、默认值、描述都有保障。我一般只在 schema 明确、字段不算太多、对下游强耦合的场景用它，比如调用数据库写入、触发第三方 API。缺点也很直接：如果 LLM 生成的不是合法 JSON，PydanticOutputParser 会抛OutputParserException，不会帮你修补。生产环境必须有配套的重试或修复逻辑。

2.2 StructuredOutputParser：轻量多字段场景还行，字段一多就露怯

StructuredOutputParser 的定位比 Pydantic 轻很多。它不要求你定义完整的 Pydantic 模型，只需要用ResponseSchema列出字段名和描述：

from langchain_core.output_parsers import StructuredOutputParser, ResponseSchema response_schemas = [ ResponseSchema(name="city", description="城市名"), ResponseSchema(name="temperature", description="当前温度，单位摄氏度"), ResponseSchema(name="condition", description="天气状况"), ResponseSchema(name="advice", 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_object=WeatherReport)

用它的前提是：你已经通过 Prompt 把期望结构说清楚了，它不会像 PydanticOutputParser 那样自动生成 format instructions，也不会对字段做严格的 Pydantic 校验。它适合两类场景：一类是模型本身就比较强，不给格式说明也能输出正确 JSON；另一类是你要的结果比较动态，今天返回两个字段，明天返回四个字段，用固定 Pydantic 模型反而绑手绑脚。

不要把它当成银弹。它实际返回的结果可能只是一个 dict，字段缺失、类型错误都需要你手动处理。如果你在 Prompt 里没有写明“必须输出 JSON”，模型给你一段带前缀的自然语言，JsonOutputParser 也有一定概率解析失败。我通常在它外面再包一层OutputFixingParser，让解析失败时把错误信息回喂给模型，让它重新修一次。这个包装对线上场景很管用，但会额外消耗一次模型调用，需要自己在成本和成功率之间做权衡。

2.4 三大 parser 对比与选型表

直接给结论，下面这张表方便大家抄作业：

如果你只能记一句话：下游代码需要强约束时用 PydanticOutputParser，只想快速拿个 dict 时用 JsonOutputParser，StructuredOutputParser 适合那种字段少到不会出错的简单场景。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, strict=False)

注意，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_object=WeatherReport) fix_parser = OutputFixingParser.from_llm(parser=parser, llm=llm)

第二层是业务兜底。如果修复仍然失败，不要把整个流式接口返回 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 f"data: {json.dumps({'type': 'token', 'content': token}, ensure_ascii=False)}\n\n" count = 0 else: count += 1 if count >= 15: # 每 15 秒没有 token 就发心跳 yield ": keep-alive\n\n" count = 0

这个做法的意图是：如果模型有输出，立刻推给前端；如果模型在思考，就靠心跳维持连接。网关看到的是“这个响应一直有数据在流动”，不会再判定为空闲。心跳间隔我建议设置在网关空闲超时的一半以内，比如代理超时 60 秒，心跳就 20 到 30 秒发一次。

另外还要注意 HTTP 层代理缓冲。很多反向代理默认会缓冲响应，直到攒够一定字节才发给客户端，这会让 SSE 实时性大打折扣。接口这边可以主动加两个 header：

headers = { "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_tools：

from langchain_openai import ChatOpenAI model = ChatOpenAI(model="gpt-4o-mini", temperature=0) 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。有些经由第三方网关封装的老模型，只有一个大的 AIMessageChunk，tool_call_chunks为空，但resp.tool_calls在最终结果里有。如果发现这种情况，建议降级为“先做完推理，再一次性输出结构化对象”，不要强求流式工具参数。

第二，工具调用可能不止一个。LLM 在一次请求里返回多个 tool_call 是很正常的，尤其是“帮我同时查上海和北京天气”这类多实体需求。上面代码里用index区分不同调用，别只取第 0 个就完事。

第三，工具参数拼接好之后，一定要用 Pydantic 再做一次校验：

report = WeatherReport(**json.loads(raw_args))

如果模型生成了多余字段或类型错误，这里会非常明确地告诉你哪里不对。别省这一步，否则脏数据会直接进入你的业务逻辑。

5. 完整链路封装：FastAPI 发 SSE，LangChain 算流式，Vue/Python 收结果

前面几节分别讲了 OutputParser、增量解析和 ToolCall，现在把它们合成一条可运行的全链路。前端是 Vue，后端是 FastAPI，中间用 SSE 在每次 agent 执行中同时传递 token 和最终结构化结果。

5.1 后端：StreamingResponse + SSE 事件协议

FastAPI 里最直接的方式是返回一个StreamingResponse，media 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 f"data: {json.dumps(event, ensure_ascii=False)}\n\n" elif event["type"] == "tool_call": yield f"data: {json.dumps(event, ensure_ascii=False)}\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_type="text/event-stream", headers={ "Cache-Control": "no-cache", "X-Accel-Buffering": "no", "Connection": "keep-alive", }, )

stream_agent内部可以接 LangGraph 的节点，也可以接普通的prompt | model | parser链。重要的是统一事件协议：type=token表示普通正文流，type=tool_call表示工具调用参数，type=done表示最终结构化结果。三者并存，前端才既能展示打字机效果，又能在最后拿到完整对象。

这里提醒一句：不要把asyncio.sleep(0)省掉。在同步阻塞的链路上，一个长耗时工具调用会把整个事件循环卡死，心跳也不会发。凡是工具调用这类慢操作，都要用asyncio.to_thread或run_in_executor包一层，不能直接阻塞在 async 函数里。

5.2 前端 Vue：EventSource 的不足与 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": "上海天气怎么样"}, timeout=httpx.Timeout(connect=30.0, read=120.0, write=30.0, pool=30.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 事件，再聊性能优化的事。