最近在推进一个基于 FastAPI + LangChain 的 Agent 项目，前端用的 Vue，需求本身并不稀奇：聊天界面要像 ChatGPT 一样逐字吐字，但后端返回的内容又不只是文本——还有结构化 JSON、工具调用参数、思考过程、联网状态这些混合数据。一开始我天真地以为，只要把 LLM 的输出用流式接口推给前端就完事了。结果第一次联调就发现了那个经典问题：SSE 流式推过来的是一段段破碎的 token，前端拿到手根本没法直接解析成 JSON；而等到把整个流收完再统一解析，流式又变成了“假流式”，用户看着就是转圈等待。这篇文章就把我在这个项目里从 SSE 流式接入，到 LangChain 三大 OutputParser 选型，再到 ToolCall 参数流式解析的完整过程写出来，包括踩过的坑和最终沉淀下来的工程方案。适合正在做 LangChain 集成、搞 AI Agent 后端、或者被“流式输出但又要结构化数据”这个问题卡住的朋友参考。
1. 流式与结构化：看似矛盾的两个需求是怎么凑到一起的
1.1 先搞清楚 SSE 在 LangChain 项目里到底扮演什么角色
SSE 全称 Server-Sent Events，是一种基于 HTTP 的长连接方案，服务端可以持续往同一个连接里推送数据，客户端用原生的 EventSource 或者 fetch 就能接收。在 LLM 项目里选 SSE 而不是 WebSocket，大多数情况是因为 LLM 服务本身只支持单向流式输出，而且 SSE 天生就在 HTTP 协议上，走负载均衡、网关、日志那一套链路不需要额外处理，前端也不用引入额外的 WebSocket 客户端库。
项目里最常见的架构是 Python 后端（FastAPI）通过 LangChain 调用大模型接口，拿到 token 流之后，再通过 SSE 转发给前端 Vue 页面。这里面有一个关键点：LangChain 的 LLM 对象本身就支持astream或astream_events，后端做的只是把 token 流“翻译”成 SSE 格式，本质上是不需要额外存储的管道转发。
但真正的复杂度不在传输层，而在内容层。LLM 流式返回的 token 是极不稳定的中间产物，比如一个 JSON 对象{"name": "张三"}在流式过程中会被拆成{"、nam、e":、"张、三"}这种碎片。也就是说，流式传输保证了“快”，但结构化解析要求“完整”，这两个需求天然是矛盾的。所以问题就变成了：解析工作到底应该放在哪一个环节。
1.2 LangChain 流式输出的真实形态：不是纯文本，而是一串事件
很多人入门 LangChain 时只看同步调用，chain.invoke()一把梭，跑通 Demo 就以为会了。真正做工程化的时候，必须切换到事件流视角。LangChain 的astream_events在版本v2下会产出大量事件，比如on_chat_model_start、on_chat_model_stream、on_llm_end等等。
对我这个项目来说，核心只关注两个事件：
on_chat_model_stream：模型每生成一个 token 块就会触发一次，数据在event["data"]["chunk"]里，可能是文本、可能带tool_call_chunks。on_chain_end：整条链跑完，可以在这里拿到完整的结构化结果。
后端要做的事情就是监听这些事件，把chunk里的文本、工具调用参数、状态信息分别抽出来，封装成不同的 SSE 事件推给前端。这里最容易犯的错误是直接在事件回调里做 JSON 解析——每个 chunk 都是不完整的，你会收获一堆Expecting value: line 1 column 1的报错。
正确的思路是：流式阶段只负责缓冲和转发，到达某个语义断点（比如一条完整的工具调用参数已经收齐）后再做解析。后面 ToolCall 部分我会详细演示这个缓冲怎么写。
2. 后端 SSE 落地细节：FastAPI 的 EventStream 工程化
2.1 StreamingResponse 与 SSE 格式的基石
FastAPI 里做 SSE 非常简单，核心就两个点：StreamingResponse的media_type="text/event-stream"，以及每次yield出去的数据必须是 SSE 协议格式。
SSE 协议格式长这样：
- 普通数据：
data: {...}\n\n - 命名事件：
event: text\ndata: {...}\n\n - 注释/心跳：
: ping\n\n
下面是我项目里最简版本的后端路由：
from fastapi import FastAPI from fastapi.responses import StreamingResponse import json app = FastAPI() async def event_generator(query: str): # chain 是构建好的 LangChain 可运行对象 async for event in chain.astream_events( {"input": query}, version="v2" ): if event["event"] == "on_chat_model_stream": chunk = event["data"]["chunk"] token = chunk.content if token: payload = json.dumps( {"type": "text", "content": token}, ensure_ascii=False, ) yield f"data: {payload}\n\n" yield "event: done\ndata: {}\n\n" @app.get("/api/chat") async def chat(query: str): return StreamingResponse( event_generator(query), media_type="text/event-stream", headers={ "Cache-Control": "no-cache", "Connection": "keep-alive", "X-Accel-Buffering": "no", }, )注意X-Accel-Buffering: no这个 header，如果你用 Nginx 反向代理，默认是开启缓冲的，不加这个 header，SSE 会被 Nginx 攒到一大坨才推给前端，流式就名存实亡了。这是我实际项目里排查了半天才发现的。
2.2 踩坑实录：stream disconnected before completion 的根因
项目联调时，前端那边一直报一个错误：stream disconnected before completion: idle timeout waiting for sse。从字面看是“空闲超时”，意思是连接建立后，在规定时间内没有收到任何 SSE 数据，中间的网络链路主动断开了连接。
为什么 LLM 已经在干活了还会“空闲”？因为大模型在流式生成之前经常有一段“思考期”，尤其接了 Agent 或者复杂 Prompt 的时候，模型内部可能在推理、在准备调用工具，这个阶段不会产生任何 token。后端这边没有数据可推，前端那边收不到流，整个连接在链路层面被判定为“空闲”，于是断开。
解决办法是在思考阶段持续推送心跳包。SSE 支持纯注释行作为心跳，它不会被前端当成业务数据触发渲染逻辑：
async def event_generator(query: str): # 等待首个 token 时持续发送心跳 import asyncio heartbeat_task = asyncio.create_task(send_heartbeat()) async for event in chain.astream_events(...): ...心跳发送用一个独立协程，每隔 10 秒往连接里写一条: ping：
import asyncio async def send_heartbeat(): while True: await asyncio.sleep(10) yield ": heartbeat\n\n"这个方案在我这边的实测效果是：以前超过 30 秒的思考期必断，加心跳之后稳定保持连接。要提醒的是，心跳间隔要根据你链路里的超时配置来定，一般是超时时间的三分之一到二分之一，太频繁了浪费带宽，太慢了等于没加。
2.3 单流多事件设计：区分“思考”“文本”和“工具调用”
SSE 的event字段可以用来区分不同类型的数据。我在项目里设计了这样几类事件：
| 事件类型 | 用途 | 前端处理方式 |
|---|---|---|
thinking | 思考过程、状态提示 | 单独区域滚动展示 |
text | 正常文本 token | 追加到聊天内容区 |
tool_call | 工具调用参数或结果 | 渲染成结构化卡片 |
done | 流结束 | 关闭 loading 状态 |
这样做的好处是，前端拿到不同类型的数据可以做差异化渲染，不至于把所有东西都塞进同一个气泡里。比如工具调用，前端可以渲染一个“正在调用搜索工具…”的折叠卡片，等流式参数解析完成后，把参数和结果填入卡片。用户看到的是一个完整的执行链路，而不是一串乱码。
3. 三大 OutputParser 拆解：什么时候用哪个
3.1 PydanticOutputParser：最硬核的固定 Schema 方案
PydanticOutputParser 是 LangChain 里结构化输出的“正统方案”，适合字段固定、逻辑严谨的业务场景。它的工作方式是：你先定义一个 Pydantic 模型，解析器会生成一段format instructions附加到 Prompt 里，告诉 LLM 必须按什么样的 JSON 结构输出，最后返回的文本会被解析成 Pydantic 对象，带类型校验。
from pydantic import BaseModel, Field from langchain.output_parsers import PydanticOutputParser class OrderInfo(BaseModel): order_id: str = Field(description="订单号") amount: float = Field(description="订单金额") status: str = Field(description="订单状态") parser = PydanticOutputParser(pydantic_object=OrderInfo) prompt = PromptTemplate( template="提取订单信息。\n{format_instructions}\n内容：{input}", input_variables=["input"], partial_variables={"format_instructions": parser.get_format_instructions()}, )跑通之后你会发现，这个方案的优点是稳定，模型一旦遵守指令，返回的就是完整、可校验的结构化对象。缺点是死板，字段一变就得改代码，而且它通常需要完整文本才能解析，不适合做流式增量解析。所以我一般把它用在“非流式的离线批量处理”场景，比如从历史日志里抽取结构化字段入库。
3.2 JsonOutputParser：轻量灵活的通用兜底
JsonOutputParser 不强制绑定 Pydantic 模型，它只是要求 LLM 输出 JSON 对象，然后解析。相比 PydanticOutputParser，它的格式说明更简短，模型更容易跟随，尤其在大模型能力参差不齐的情况下，越简短的约束越不容易出错。
from langchain.output_parsers import JsonOutputParser parser = JsonOutputParser() prompt = PromptTemplate( template="输出 JSON，包含 name 和 score。\n{format_instructions}\n问题：{input}", input_variables=["input"], partial_variables={"format_instructions": parser.get_format_instructions()}, )使用注意：JsonOutputParser 解析出来的是纯 Python 字典，没有校验。字段类型不对、缺字段，它都直接放过。所以它适合字段不多、容错要求不高的场景，适合做兜底，比如前端只需要展示，不需要入库。
而且 JsonOutputParser 对我来说最大的价值是它有parse_partial_json能力——传入一段不完整的 JSON 文本，能解析出当前已经“长出来”的字段。这个能力在流式场景里非常有用，比如前端可以实时看到工具调用的参数名和参数值一个个蹦出来。后面 ToolCall 那节会用到它。
3.3 StructuredOutputParser（ResponseSchema）与 StrOutputParser 的对比
还有一个容易被忽略的是 StructuredOutputParser，它基于ResponseSchema定义输出结构。它和 Pydantic 的区别是：它没有强类型校验，生成的指令也更偏向“列表式输出”，但在某些模型的听话程度上反而比 Pydantic 好。
from langchain.output_parsers import StructuredOutputParser, ResponseSchema response_schemas = [ ResponseSchema(name="title", description="标题"), ResponseSchema(name="summary", description="摘要"), ] parser = StructuredOutputParser.from_response_schemas(response_schemas)配合 StrOutputParser 一起看会更清楚。StrOutputParser 是 LangChain 的默认输出解析器，它做的事情只有一件：把流式 chunk 累积成完整字符串。它不解析 JSON、不做校验，是最纯粹的文本流解析器。
我用一张表把这几个方案放在一起对比：
| 解析器 | Schema 定义 | 类型校验 | 支持部分 JSON | 适用场景 |
|---|---|---|---|---|
| StrOutputParser | 无 | 无 | 无 | 纯文本流式拼接 |
| JsonOutputParser | 无强制 | 无 | 支持 | 轻量 JSON、流式增量展示 |
| PydanticOutputParser | Pydantic 模型 | 强校验 | 不支持 | 离线批量、入库强校验 |
| StructuredOutputParser | ResponseSchema | 弱校验 | 不支持 | 结构简单、通用展示 |
3.4 我的选型标准：一句话说清楚
做了好几个项目之后，我现在的选型逻辑非常直接：如果能接受稍长的 Prompt 开销并且字段固定，用 PydanticOutputParser；如果只想让模型输出 JSON 且前端要做增量展示，用 JsonOutputParser；如果只是把整条链跑通、返回文本也没关系，默认 StrOutputParser。StructuredOutputParser 我用得少了，主要是它的输出格式在部分模型上容易与 JSON 模式打架，调试成本高于收益。但你 Model 比较便宜、响应比较随意的时候，它的鲁棒性反而更好。
4. ToolCall 方案实战：让 AI 真的“下地干活”
4.1 bind_tools 和 with_structured_output：两种接入姿势
当 Agent 需要调用外部工具时，LLM 的输出就不再是纯文本，而是一个“工具调用请求”，里面包含工具名和参数。LangChain 里接入这个能力有两条路线。
一条是bind_tools，给模型绑定工具定义，模型在生成过程中会返回一个或多个tool_call，其中参数部分是一段 JSON 字符串：
from langchain_openai import ChatOpenAI from langchain_core.tools import tool @tool def get_weather(city: str) -> str: """查询城市天气""" return f"{city} 今天晴，25℃" llm = ChatOpenAI(model="gpt-4o", temperature=0) llm_with_tools = llm.bind_tools([get_weather])另一条是with_structured_output，本质上是把“要求模型以结构化的 JSON 形式回答”封装成标准方法，传入一个 Pydantic 模型即可，LangChain 内部会帮你处理是走function_calling还是json_mode：
class WeatherResponse(BaseModel): city: str temperature: float condition: str structured_llm = llm.with_structured_output(WeatherResponse)这两条路线的区别在于定位：bind_tools强调的是“触发工具执行”，结果仍然是复杂的工具调用对象；with_structured_output强调的是“让模型按 schema 回答”，结果就是干净的 Pydantic 对象。如果你需要的是 Agent 自主决策调用工具，选前一条；如果你的目标是纯结构化抽取，选后一条。实际项目里，Agent 用bind_tools，数据抽取用with_structured_output，各干各的活。
4.2 流式 tool_call_chunks 的正确累积姿势
这里是最容易写错的环节。模型在流式模式下，工具调用的参数不是一个完整的 JSON 一次性返回的，而是一块块分发，每一块叫一个tool_call_chunk。一个关键点是index字段，它标识这段 chunk 属于第几个工具调用。
我最初的错误写法是直接把每个 chunk 单独json.loads，结果就是一片报错。正确写法是维护一个按 index 索引的缓冲字典，把 id、name、args 分段拼接，等完整后再解析：
tool_call_chunks = {} async for event in llm_with_tools.astream_events( messages, version="v2" ): if event["event"] == "on_chat_model_stream": chunk = event["data"]["chunk"] for tc_chunk in chunk.tool_call_chunks: idx = tc_chunk["index"] if idx not in tool_call_chunks: tool_call_chunks[idx] = { "id": "", "name": "", "args": "", } if tc_chunk["id"]: tool_call_chunks[idx]["id"] += tc_chunk["id"] if tc_chunk["name"]: tool_call_chunks[idx]["name"] += tc_chunk["name"] if tc_chunk["args"]: tool_call_chunks[idx]["args"] += tc_chunk["args"] # 流结束后解析完整参数 import json for idx, acc in tool_call_chunks.items(): parsed_args = json.loads(acc["args"])注意：id和name字段在流式过程中通常只在第一块 chunk 出现，后面都是空字符串，所以用if tc_chunk["id"]这种条件追加是安全的。args是逐段拼接的 JSON 字符串碎片，直到流结束才是一个完整 JSON。
4.3 中途做增量解析：让参数实时呈现出来
如果不想等到流结束才看到参数，可以配合前面提到的parse_partial_json做增量解析：
from langchain.output_parsers import JsonOutputParser partial_parser = JsonOutputParser() # 每次拼接完一块 args 就尝试解析一次，捕获失败则忽略 try: partial_args = partial_parser.parse_partial_json(acc["args"]) # 推送 partial_args 给前端，渲染成实时变化的卡片 except Exception: pass这样前端在工具参数流式到达过程中，就能看到一个对象不断“长出”新字段，体验比等全部完成再渲染要好很多。但这个能力不是每个模型都稳定支持，实测在部分模型上parse_partial_json偶尔会解析出半截结构，所以推送时我会打个is_partial标记，前端知道当前展示的是“不完整预览”，最终以完整结果为准。
4.4 Agent 执行器：工具结果如何回流给模型
工具调用解析出来之后，需要把执行结果作为新的消息喂回给模型，Agent 才会继续生成最终回答。LangChain 有现成的 AgentExecutor 和 LangGraph 可以处理这个循环，但如果你只想自己掌控流式过程，手动实现也不复杂：
from langchain_core.messages import AIMessage, ToolMessage # 假设 tool_name 和 tool_args 已解析 result = tools_map[tool_name].invoke(tool_args) messages.extend([ AIMessage( content="", tool_calls=[{"name": tool_name, "args": tool_args, "id": tool_call_id}] ), ToolMessage(content=str(result), tool_call_id=tool_call_id), ]) # 继续下一轮模型生成 async for event in llm_with_tools.astream_events(messages, version="v2"): ...这个循环是 Agent 的核心，跑通这一步，“让 AI 下地干活”才真正落地。我项目里还基于这个模式做了一套小型的工具注册表，通过装饰器把 Python 函数挂载进去，模型侧只用关心工具名和参数，执行细节全部在后端完成。
5. 前后端对接完整链路：Vue 侧怎么解析 SSE
5.1 fetch + ReadableStream：不依赖 EventSource 的原因
很多人会直接用浏览器原生的EventSource接收 SSE，但它有一个致命限制：只能 GET 请求，而且没法自定义 header。我的项目里需要给后端传 token 鉴权，所以选择了fetch+ReadableStream手工解析。
核心代码如下：
const res = await fetch("/api/chat?query=" + encodeURIComponent(query), { headers: { Authorization: `Bearer ${token}` }, }); const reader = res.body!.getReader(); const decoder = new TextDecoder("utf-8"); let buffer = ""; while (true) { const { done, value } = await reader.read(); buffer += decoder.decode(value, { stream: true }); const parts = buffer.split("\n\n"); buffer = parts.pop() ?? ""; for (const part of parts) { handleSSEChunk(part); } if (done) break; }handleSSEChunk里面按event:和data:两行拆分，然后把data部分JSON.parse后分发到对应的渲染函数。这个方案兼容性很好，占用的代码量也不大。
5.2 按事件分流渲染：思考区、文本区、结构化卡片区
前端的渲染逻辑建议拆成三块：
- 聊天主区域：只接收
text事件，逐字追加； - 顶部状态条/侧边区：接收
thinking事件，展示当前正在执行的动作，比如“正在搜索资料”“正在分析订单”； - 工具调用卡片区：接收
tool_call事件，流式更新参数。
每一块互不干扰，用户可以清晰地看到 AI 先想了什么、调了什么工具、最后输出了什么。这个交互模式现在已经成为我所有 Agent 项目的标配。
5.3 断线重连与断点续推
项目上线一个月后，真实用户网络环境下频繁出现连接中断。原因不完全是后端超时，也有移动网络切换、网关策略等。客户端需要做两件事：
一是自动重连。我采用了指数退避策略：第 1 次失败等 500ms，之后翻倍，最多 5 次就不再重试，改提示用户刷新。
二是断点续推。我给后端接口加了一个last_offset参数，后端缓存最近 N 条已推送的 token，客户端重连时带上最后收到的 offset，后端从断点继续推，而不是重新生成一遍。注意，这里并不是把整个生成上下文都缓存，而是缓存“已经推送给前端的内容”，保证前端界面能续上。
这个设计实测下来效果很好，但缓存会占内存，我的做法是只保留最近 500 条 token，超过就丢弃，极端情况下最多丢失一小段内容，前端展示一个“内容可能缺失”的提示，整体可用性远大于不处理。
6. 实战踩坑汇总与我这边的最终搭配
6.1 问题清单：大部分人都可能在联调时遇到
下面这些坑是我在这个项目里实际遇到并逐一解决的，列成表格方便你排查：
| 现象 | 根因 | 解决方案 |
|---|---|---|
| 前端收到的流一坨一坨地来 | Nginx 缓冲未关闭 | 加X-Accel-Buffering: no |
| 长时间思考后连接断开 | 链路空闲超时 | SSE 心跳注释行保活 |
Expecting value解析错误 | 对 chunk 单独 JSON 解析 | 用缓冲累积，到了断点再解析 |
| 工具参数丢字段 | 多轮 ToolMessage 没关联正确 id | 注意tool_call_chunks按 index 维护 |
| 断线后重连重复生成 | 无断点续推 | 引入last_offset参数 |
| 模型输出偶尔多一个逗号 | 自定义格式指令过长 | 换更短小的format_instructions |
6.2 当前项目里的最终组合
折腾完一轮之后，我现在的固定搭配是这样：
- 纯文本聊天走
StrOutputParser，主打简单可靠； - 结构化输出优先用
with_structured_output(PydanticModel)，省心，校验强； - 需要流式展示工具参数时，用
bind_tools+tool_call_chunks累积，配合JsonOutputParser.parse_partial_json做增量预览； - 统一用
astream_events的v2事件协议做底，前端只认四类 SSE 事件。
这个组合的好处是每一层都有明确的职责边界，流式的归流式、解析的归解析、校验的归校验。改一个环节不会牵动其他地方，后续加新工具、新任务类型，只需要扩展事件类型和工具注册表，核心链路基本不用动。
最后再分享一个小经验。很多人一上来就追求“全链路流式”，但流式本身是有成本的——缓冲、状态管理、断线续传，每一块都要额外写代码。如果你的业务只需要最终结果，那老老实实非流式接invoke就好，别给自己找麻烦。真正需要流式的场景，一定是用户对等待时间敏感、或者想看过程反馈的产品。想清楚这一点，再决定要不要啃流式这条链路，比我接下来讲的所有细节都重要。
