1. 为什么流式输出和结构化输出总是打架做过大模型应用的人大概率都遇到过这个场景前端要打字机效果一个字一个字往外蹦体验拉满后端却想要一个干净的 JSON直接塞进数据库或者喂给下一个函数。这两件事天然是矛盾的——流式输出的本质是边生成边推送你拿到的是碎片结构化输出的本质是完整、合法、可解析你要的是整体。SSEServer-Sent Events把这种矛盾放大了它基于 HTTP 长连接服务端持续往客户端推data:事件一旦中间某个 chunk 把 JSON 截断了前端JSON.parse直接炸。LangChain 这套东西的价值就在这里。它没有回避矛盾而是用OutputParser把模型吐出来的自由文本翻译成程序能用的结构再用ToolCall把结构化结果直接变成可执行动作。整条链路是SSE 负责传输层的实时性OutputParser 负责语义层的结构化ToolCall 负责应用层的执行力。三者串起来才是一个能上生产的 Agent 骨架。这篇东西适合谁看如果你已经跑通过最简单的llm.invoke()但一上流式就不知道怎么接结构化或者接了 ToolCall 之后发现流式全乱了那这篇就是给你写的。我会把三种主流 OutputParser 的适用边界、SSE 流式下怎么保住结构化、ToolCall 和流式怎么共存全部拆开讲附上能直接抄的参数和踩过的坑。2. 三种 OutputParser 的选型逻辑与底层原理2.1 PydanticOutputParser强类型场景的首选PydanticOutputParser 是我在需要严格字段校验时第一个想到的方案。它的工作方式是你定义一个 Pydantic 模型parser 会自动生成一段格式说明format instructions塞进 prompt 里告诉模型你必须按这个 schema 输出然后模型返回的文本再被 parser 解析回 Pydantic 对象。为什么选它因为 Pydantic 本身带类型校验。模型如果漏了字段、类型写错比如把 int 写成字符串parser 会直接抛ValidationError而不是悄悄给你一个残缺的 dict。这在生产环境里太重要了——宁可报错重试也不要脏数据流进下游。from langchain_core.output_parsers import PydanticOutputParser from pydantic import BaseModel, Field class ProductInfo(BaseModel): name: str Field(description产品名称) price: float Field(description价格单位元) tags: list[str] Field(description标签列表) parser PydanticOutputParser(pydantic_objectProductInfo) prompt ChatPromptTemplate.from_template( 提取产品信息。\n{format_instructions}\n\n输入{input} ).partial(format_instructionsparser.get_format_instructions())这里有个细节很多人忽略get_format_instructions()生成的说明里包含 JSON schematoken 消耗不小。如果你的模型上下文紧张可以考虑精简 Field 的 description但别删——description 是模型理解字段含义的主要依据。2.2 JsonOutputParser轻量、宽容、适合流式JsonOutputParser 是 PydanticOutputParser 的轻量版。它不要求你定义模型直接返回 dict也不做严格类型校验。听起来好像更弱但它在流式场景下有个杀手锏支持增量解析。流式输出时模型吐出来的是{name: 苹这种半截 JSON。JsonOutputParser 内部用了parse_partial_json能在 JSON 还没闭合的时候尽量解析出已经完整的部分字段。这意味着你可以在 SSE 推送过程中前端就能逐步渲染出name 已经有了price 还在等的效果而不是等整个 JSON 结束才一次性显示。from langchain_core.output_parsers import JsonOutputParser parser JsonOutputParser() chain prompt | llm | parser # 流式场景 async for chunk in chain.astream({input: ...}): # chunk 是逐步补全的 dict print(chunk)注意JsonOutputParser 的宽容是有代价的。如果模型输出的是{a: 1, b:它能给你{a: 1}但如果你依赖b字段做后续逻辑就会拿到 None。所以流式阶段只适合做 UI 渲染最终落库前一定要等完整结果再校验一次。2.3 StructuredOutputParser多字段但不想写模型的折中StructuredOutputParser 介于前两者之间。你用ResponseSchema声明字段名和描述它生成格式说明返回 dict。相比 Pydantic 少了类型系统相比 Json 多了 schema 约束。适合那种字段不多、类型简单、不想引入 Pydantic 依赖的场景。from langchain.output_parsers import StructuredOutputParser, ResponseSchema schemas [ ResponseSchema(namesummary, description一句话摘要), ResponseSchema(namesentiment, description情感倾向正面/负面/中性), ] parser StructuredOutputParser.from_response_schemas(schemas)三种 parser 的选型我一般这么判断场景推荐 Parser理由需要类型校验、字段多PydanticOutputParser强类型错误早暴露流式增量渲染JsonOutputParser支持 partial 解析字段少、快速原型StructuredOutputParser无需定义模型输出要直接触发函数配合 ToolCall结构化后转工具调用2.4 格式说明到底是怎么教模型的很多人以为 parser 只是解析其实它一半的工作在约束生成。get_format_instructions()返回的文本大致长这样The output should be formatted as a JSON instance that conforms to the JSON schema below. ...这段文本被拼进 prompt 后模型是在被提示的情况下生成结构化内容的。所以 prompt 里 format_instructions 的位置很关键——放在最后、紧挨着用户输入约束效果最好。我试过放在最前面模型经常看着看着就忘了尤其是长上下文场景。3. SSE 流式链路下如何保住结构化输出3.1 SSE 的本质与 LangChain 的 astream 对接SSE 就是服务端往客户端单向推文本流每条消息以data:开头以两个换行结束。LangChain 的astream返回的是异步生成器每个 chunk 是一个AIMessageChunk。你要做的就是把 chunk 的内容转成 SSE 事件推出去。from fastapi import FastAPI from fastapi.responses import StreamingResponse app FastAPI() async def event_generator(input_text: str): async for chunk in chain.astream({input: input_text}): # chunk 可能是 dictparser 后或 AIMessageChunkparser 前 yield fdata: {json.dumps(chunk, ensure_asciiFalse)}\n\n app.get(/stream) async def stream(q: str): return StreamingResponse(event_generator(q), media_typetext/event-stream)关键点media_type必须是text/event-stream每条消息必须以\n\n结尾否则浏览器 EventSource 不认。3.2 流式 结构化先流文本再补结构最稳的架构不是边流边结构化而是双通道一条通道流原始文本给前端做打字机效果另一条通道在流结束后做结构化解析。为什么因为结构化解析需要完整文本流式过程中强行解析只会得到残缺结果。但如果你确实想要边流边结构化的体验可以用 JsonOutputParser 的 partial 能力前端拿到部分字段就先渲染等finish_reason到了再补全。我实测下来这种方案在字段之间有依赖关系时容易出问题——比如total price * quantityprice 先到了但 quantity 没到前端算出来的 total 是错的。所以我的建议是UI 层用 partial 做占位渲染业务逻辑层等完整结果。3.3 处理 idle timeout 与断流重连热词里有个很典型的报错stream disconnected before completion: idle timeout waiting for sse。这是 SSE 长连接在中间层网关、负载均衡被判定为空闲而掐断。模型生成慢的时候两个 chunk 之间可能隔十几秒中间层就以为连接死了。解决办法有三个层次心跳保活每隔 10-15 秒推一个注释行: keepalive\n\nEventSource 会忽略注释但连接保持活跃。调整中间层超时把网关的 idle timeout 调到大于模型最长生成时间。客户端重连EventSource 自带重连但重连后要从头开始所以服务端最好支持Last-Event-ID做断点续传。async def event_generator(input_text: str): last_heartbeat time.time() async for chunk in chain.astream({input: input_text}): if time.time() - last_heartbeat 10: yield : keepalive\n\n last_heartbeat time.time() yield fdata: {json.dumps(chunk, ensure_asciiFalse)}\n\n提示心跳间隔别设太短太频繁会占用带宽也别太长超过中间层超时就白搭。10-15 秒是实测比较稳的区间。3.4 前端 EventSource 的正确用法前端这块坑也不少。EventSource 只支持 GET不支持自定义 header所以鉴权只能靠 query 参数或者 cookie。另外 EventSource 收到非data:开头的行会忽略收到event:行会触发对应的事件监听。const es new EventSource(/stream?q${encodeURIComponent(query)}); es.onmessage (e) { const data JSON.parse(e.data); // 增量渲染 }; es.onerror (err) { // EventSource 会自动重连但要注意重连风暴 es.close(); };重连风暴是个隐蔽的坑如果服务端一直返回错误EventSource 会不停重连把服务端打挂。所以onerror里最好加个退避逻辑或者服务端在错误时返回特定事件让前端主动 close。4. ToolCall 与流式的共存方案4.1 ToolCall 的本质结构化输出的特例ToolCall 说白了就是模型输出一个结构化的函数调用请求。它和 OutputParser 的关系是OutputParser 把文本转成结构ToolCall 把结构转成动作。模型返回的tool_calls字段本身就是结构化的包含name和argumentsJSON 字符串。from langchain_core.tools import tool tool def get_weather(city: str) - str: 查询指定城市的天气 return f{city} 今天晴25度 llm_with_tools llm.bind_tools([get_weather]) response llm_with_tools.invoke(北京天气怎么样) # response.tool_calls [{name: get_weather, args: {city: 北京}, id: ...}]4.2 流式下 ToolCall 参数是分片到达的这是最容易踩的坑。流式模式下tool_calls的arguments不是一次性给你的而是分片拼接的。第一个 chunk 可能只有name后面的 chunk 逐步补arguments的字符串片段。你必须自己累积拼接等finish_reason tool_calls才能解析。tool_call_buffer {} async for chunk in llm_with_tools.astream(北京天气): for tc in chunk.tool_calls: idx tc[index] if idx not in tool_call_buffer: tool_call_buffer[idx] {name: , args: } tool_call_buffer[idx][name] tc.get(name, ) tool_call_buffer[idx][args] tc.get(args, ) # 流结束后再 json.loads(args)我见过有人直接在流式循环里json.loads(tc[args])结果必然报JSONDecodeError因为参数还没拼完。这个坑几乎每个新手都会踩一次。4.3 流式 ToolCall 的完整执行链路一个完整的流式 ToolCall 链路是这样的模型流式返回累积 tool_calls 分片流结束解析出完整的工具名和参数执行工具函数拿到结果把工具结果作为ToolMessage塞回消息历史再次调用模型生成最终自然语言回复最终回复同样可以流式推给前端第 5 步的二次调用也可以流式这样用户看到的是先显示正在调用工具再流式显示最终答案。体验上比转圈等半天好太多。4.4 并行 ToolCall 的处理模型可能一次返回多个 tool_calls比如查北京和上海两个城市的天气。这时候要并行执行用asyncio.gatherimport asyncio async def execute_tools(tool_calls): tasks [execute_single(tc) for tc in tool_calls] return await asyncio.gather(*tasks)注意工具函数本身要支持异步或者用run_in_executor包一层。同步工具在异步链路里会阻塞事件循环SSE 的心跳就发不出去了又回到 idle timeout 那个坑。5. 常见问题排查与避坑速查5.1 结构化解析失败的排查顺序模型输出解析失败别急着改 prompt按这个顺序排查看原始输出先把 parser 去掉打印模型原始返回确认是不是模型根本没按格式来看 format_instructions 是否进了 prompt用prompt.format()打印最终 prompt看模型能力小模型7B 以下对复杂 schema 的遵循能力很弱换大模型试试看是否被截断max_tokens设太小JSON 没输出完就断了5.2 常见问题速查表现象可能原因解决方向JSONDecodeError流式参数未拼完等 finish_reason 再解析ValidationError 缺字段模型漏输出加 few-shot 示例idle timeout心跳缺失加 keepalive 注释行前端重连风暴服务端持续报错onerror 加退避ToolCall 参数为空分片未累积按 index 累积拼接中文乱码未设 ensure_asciijson.dumps 加参数5.3 我踩过的三个真实坑第一个坑ensure_asciiFalse忘了加中文全变成\uXXXX前端显示一堆乱码。这个参数在json.dumps里默认是 True必须显式关掉。第二个坑Pydantic 模型的 Field description 写得太简略模型理解偏了。比如price: float没写单位模型有时返回25元这种带单位的字符串直接校验失败。description 一定要写清楚格式和单位。第三个坑ToolCall 的id字段在流式分片里可能只在第一个 chunk 出现后面 chunk 没有。如果你每个 chunk 都新建一个 buffer 项会导致同一个 tool_call 被拆成多个。正确做法是用index做 keyid只在首次出现时记录。5.4 性能与成本上的取舍结构化输出会显著增加 token 消耗因为 format_instructions 本身占几百 token模型还要思考怎么填 schema。如果 QPS 高这笔成本不小。我的做法是对格式要求不严的场景用 JsonOutputParser说明短对强类型场景才上 Pydantic并且把 schema 精简到必要字段。流式方面SSE 本身开销很小但长连接会占用服务端连接数。如果并发高考虑用连接池或者把 SSE 网关独立部署别和主业务抢资源。6. 一套可直接复用的最小骨架把上面的东西串起来一个能跑的最小骨架大概是这样FastAPI 提供 SSE 端点LangChain 负责 chain 编排JsonOutputParser 做流式增量解析ToolCall 做动作触发心跳保活防断流。这套骨架我在几个内部项目里复用改改 prompt 和工具就能上。真正上生产还要补的东西错误重试、限流、日志埋点、断点续传。但骨架对了后面都是加法。我个人的体会是SSE 和结构化输出的矛盾不是靠某个库解决的而是靠分层——传输层管实时解析层管结构执行层管动作各司其职别让一层干三层的活。想清楚这个分层剩下的都是细节问题。