最近在推进一个基于 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_typetext/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}, versionv2 ): if event[event] on_chat_model_stream: chunk event[data][chunk] token chunk.content if token: payload json.dumps( {type: text, content: token}, ensure_asciiFalse, ) yield fdata: {payload}\n\n yield event: done\ndata: {}\n\n app.get(/api/chat) async def chat(query: str): return StreamingResponse( event_generator(query), media_typetext/event-stream, headers{ Cache-Control: no-cache, Connection: keep-alive, X-Accel-Buffering: no, }, )注意X-Accel-Buffering: no这个 header如果你用 Nginx 反向代理默认是开启缓冲的不加这个 headerSSE 会被 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 秒往连接里写一条: pingimport 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_objectOrderInfo) 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 StructuredOutputParserResponseSchema与 StrOutputParser 的对比还有一个容易被忽略的是 StructuredOutputParser它基于ResponseSchema定义输出结构。它和 Pydantic 的区别是它没有强类型校验生成的指令也更偏向“列表式输出”但在某些模型的听话程度上反而比 Pydantic 好。from langchain.output_parsers import StructuredOutputParser, ResponseSchema response_schemas [ ResponseSchema(nametitle, description标题), ResponseSchema(namesummary, description摘要), ] parser StructuredOutputParser.from_response_schemas(response_schemas)配合 StrOutputParser 一起看会更清楚。StrOutputParser 是 LangChain 的默认输出解析器它做的事情只有一件把流式 chunk 累积成完整字符串。它不解析 JSON、不做校验是最纯粹的文本流解析器。我用一张表把这几个方案放在一起对比解析器Schema 定义类型校验支持部分 JSON适用场景StrOutputParser无无无纯文本流式拼接JsonOutputParser无强制无支持轻量 JSON、流式增量展示PydanticOutputParserPydantic 模型强校验不支持离线批量、入库强校验StructuredOutputParserResponseSchema弱校验不支持结构简单、通用展示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(modelgpt-4o, temperature0) llm_with_tools llm.bind_tools([get_weather])另一条是with_structured_output本质上是把“要求模型以结构化的 JSON 形式回答”封装成标准方法传入一个 Pydantic 模型即可LangChain 内部会帮你处理是走function_calling还是json_modeclass 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, versionv2 ): 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(contentstr(result), tool_call_idtool_call_id), ]) # 继续下一轮模型生成 async for event in llm_with_tools.astream_events(messages, versionv2): ...这个循环是 Agent 的核心跑通这一步“让 AI 下地干活”才真正落地。我项目里还基于这个模式做了一套小型的工具注册表通过装饰器把 Python 函数挂载进去模型侧只用关心工具名和参数执行细节全部在后端完成。5. 前后端对接完整链路Vue 侧怎么解析 SSE5.1 fetch ReadableStream不依赖 EventSource 的原因很多人会直接用浏览器原生的EventSource接收 SSE但它有一个致命限制只能 GET 请求而且没法自定义 header。我的项目里需要给后端传 token 鉴权所以选择了fetchReadableStream手工解析。核心代码如下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_instructions6.2 当前项目里的最终组合折腾完一轮之后我现在的固定搭配是这样纯文本聊天走StrOutputParser主打简单可靠结构化输出优先用with_structured_output(PydanticModel)省心校验强需要流式展示工具参数时用bind_toolstool_call_chunks累积配合JsonOutputParser.parse_partial_json做增量预览统一用astream_events的v2事件协议做底前端只认四类 SSE 事件。这个组合的好处是每一层都有明确的职责边界流式的归流式、解析的归解析、校验的归校验。改一个环节不会牵动其他地方后续加新工具、新任务类型只需要扩展事件类型和工具注册表核心链路基本不用动。最后再分享一个小经验。很多人一上来就追求“全链路流式”但流式本身是有成本的——缓冲、状态管理、断线续传每一块都要额外写代码。如果你的业务只需要最终结果那老老实实非流式接invoke就好别给自己找麻烦。真正需要流式的场景一定是用户对等待时间敏感、或者想看过程反馈的产品。想清楚这一点再决定要不要啃流式这条链路比我接下来讲的所有细节都重要。