尧图网站建设 
1. 为什么流式输出和结构化输出总是打架做过大模型应用的人多半遇到过这个场景前端用 SSE 接收流式内容一个字一个字往外蹦体验很顺滑可一旦业务需要拿到结构化的 JSON比如要提取字段、要触发下游工具调用流式就变成了麻烦。你没法在流还没结束时解析 JSON因为 JSON 是残缺的可如果等流结束再解析那流式的意义又没了一半。这个矛盾在 LangChain 里体现得特别明显。LangChain 的 OutputParser 体系天生是为拿到完整输出再解析设计的而 SSE 流式是边生成边消费。两者要配合就得在中间做文章。再加上 ToolCall 这一层——模型不只是输出文本还要决定调用哪个工具、传什么参数——结构化输出的要求就更高了因为工具调用的参数必须是合法 JSON错一个括号整个调用就废了。我最近在一个基于 FastAPI LangChain 的项目里把这三块彻底捋了一遍SSE 流式接口怎么封装、OutputParser 怎么选、ToolCall 怎么和流式共存。踩的坑不少有些是 LangChain 版本差异导致的有些是 SSE 协议本身的特性决定的。这篇就把整个实战过程拆开讲从最基础的流式封装讲到 ToolCall 的流式解析尽量把每个为什么这么设计说清楚。适合的读者是已经用过 LangChain 基础功能、想把它接到真实 Web 服务里的人。如果你还在纠结 LangChain 怎么入门建议先把 Chain 和 Prompt 跑通再来看这篇不然有些细节会显得突兀。2. SSE 流式接口的封装逻辑与常见断流问题2.1 SSE 到底是个什么东西为什么大模型场景偏爱它SSE 全称 Server-Sent Events是 HTTP 协议下的一种单向推送机制。服务端保持一个长连接持续往客户端写data: xxx\n\n格式的文本块客户端用 EventSource 或者 fetch 的 ReadableStream 来读。它和 WebSocket 的区别在于SSE 是单向的服务端到客户端基于纯 HTTP不需要额外的协议升级浏览器原生支持自动重连。大模型场景偏爱 SSE 的原因很实际。第一大模型的输出本身就是单向的用户发一次请求模型吐一串 token不需要双向通信。第二SSE 基于 HTTP能直接穿过大多数网关和负载均衡部署成本低。第三SSE 的文本格式天然适合传 token每个 token 包一个data:就行。但 SSE 有个容易被忽略的点它是纯文本协议没有二进制帧的概念。这意味着你不能在里面塞复杂的二进制结构所有东西都得序列化成字符串。这一点在后面讲结构化输出时会变成一个关键约束。2.2 用 FastAPI 封装 SSE 接口的最小可用骨架先给一个能跑的最小骨架。FastAPI 里返回 SSE 用StreamingResponse媒体类型设成text/event-stream。from fastapi import FastAPI from fastapi.responses import StreamingResponse import asyncio app FastAPI() async def event_generator(): for i in range(5): yield fdata: chunk-{i}\n\n await asyncio.sleep(0.5) yield data: [DONE]\n\n app.get(/stream) async def stream(): return StreamingResponse( event_generator(), media_typetext/event-stream, headers{ Cache-Control: no-cache, Connection: keep-alive, X-Accel-Buffering: no, }, )这里有几个细节值得说。X-Accel-Buffering: no这个头是给 Nginx 看的告诉它不要缓冲这个响应。如果你部署在 Nginx 后面又没加这个头会发现流式变成了攒一批再发前端看起来就是卡顿的。Cache-Control: no-cache防止中间层缓存。Connection: keep-alive保持长连接。data:后面跟内容然后必须有两个换行\n\n这是 SSE 协议的帧分隔符。少一个换行客户端就认为这一帧还没结束会一直等。这个坑我见过不止一次尤其是拼接字符串时手滑。2.3 stream disconnected before completion 的根因排查热词里有个很典型的报错stream disconnected before completion: idle timeout waiting for sse。这个错误的字面意思是SSE 连接在完成前断开了因为等待 SSE 时空闲超时了。根因通常有三个方向。第一是服务端或中间层设了空闲超时。Nginx 默认的proxy_read_timeout是 60 秒如果模型生成慢60 秒内没有任何数据写出连接就被掐了。解决办法是在 Nginx 配置里调大这个值或者更优雅的做法是定期发送心跳包。async def event_generator_with_heartbeat(chain, inputs): queue asyncio.Queue() async def produce(): async for chunk in chain.astream(inputs): await queue.put(chunk) await queue.put(None) producer asyncio.create_task(produce()) while True: try: chunk await asyncio.wait_for(queue.get(), timeout15) except asyncio.TimeoutError: yield : heartbeat\n\n continue if chunk is None: break yield fdata: {chunk}\n\n await producer心跳包用:开头这是 SSE 的注释语法客户端会忽略它但它能让连接保持活跃避免空闲超时。15 秒发一次是个比较稳妥的值比大多数默认超时都短。第二个方向是客户端主动断开。用户切了页面、关了标签EventSource 会断开服务端如果还在往一个已断开的连接写数据就会抛异常。这个要在生成器里捕获asyncio.CancelledError或者ConnectionResetError做好清理。第三个方向是代理层。有些云厂商的负载均衡对长连接有硬性时长限制比如 5 分钟强制断开。这种只能靠客户端重连 服务端记录断点来续传或者干脆改成轮询。选型时要先确认部署环境的限制。2.4 前端消费 SSE 的两种姿势与踩坑前端消费 SSE 有两种主流方式。一种是用原生EventSource简单但有局限它只支持 GET 请求不能自定义请求头没法传 Authorization。另一种是用fetchReadableStream灵活但需要自己解析帧。async function consumeSSE(url, body) { const resp await fetch(url, { method: POST, headers: { Content-Type: application/json }, body: JSON.stringify(body), }); const reader resp.body.getReader(); const decoder new TextDecoder(); let buffer ; while (true) { const { done, value } 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: )) { const data part.slice(6); if (data [DONE]) return; handleChunk(data); } } } }这里的关键是buffer的处理。网络传输是分片的一个 SSE 帧可能被切成两半到达所以不能假设每次read()拿到的就是完整帧。正确做法是累积到 buffer按\n\n切分最后一段可能不完整留在 buffer 里等下一片。这个逻辑写错了就会出现 JSON 解析到一半报错的情况而且报错是间歇性的很难复现。decoder.decode(value, { stream: true })里的stream: true也很重要。UTF-8 一个汉字占 3 字节如果分片正好切在汉字中间不加这个参数会解码出乱码。加了之后 TextDecoder 会保留不完整的字节序列等下一片拼上再解码。3. LangChain 三大 OutputParser 的选型与实战差异3.1 PydanticOutputParser强类型场景的首选LangChain 的 OutputParser 家族里PydanticOutputParser 是最重的一个也是结构化要求最高时的首选。它的工作方式是你定义一个 Pydantic 模型它自动生成一段格式说明塞进 prompt模型按这个格式输出解析器再把文本转回 Pydantic 对象。from langchain_core.output_parsers import PydanticOutputParser from langchain_core.prompts import ChatPromptTemplate from pydantic import BaseModel, Field class PersonInfo(BaseModel): name: str Field(description人物姓名) age: int Field(description年龄整数) skills: list[str] Field(description技能列表) parser PydanticOutputParser(pydantic_objectPersonInfo) prompt ChatPromptTemplate.from_messages([ (system, 从文本中提取人物信息。\n{format_instructions}), (human, {input}), ]).partial(format_instructionsparser.get_format_instructions()) chain prompt | llm | parser result chain.invoke({input: 张三今年 28 岁会 Python 和 Go。})get_format_instructions()会生成一段类似请输出符合以下 JSON Schema 的内容的说明。这段说明的质量直接决定解析成功率。Pydantic 的Field(description...)会变成 JSON Schema 里的 description模型靠这个理解每个字段的含义。所以字段描述写得越清楚解析越稳。PydanticOutputParser 的优势是类型安全。解析出来的对象有完整的类型校验age是 int 不是 strskills是 list 不是字符串。下游代码可以直接用属性访问不用手动转换。缺点是它对格式要求严格模型稍微跑偏就解析失败而且失败时抛的是OutputParserException需要自己捕获处理。3.2 JsonOutputParser轻量灵活但需要自己兜底JsonOutputParser 比 Pydantic 轻它不要求你定义模型直接输出 dict。适合结构不固定、或者你懒得定义模型的场景。from langchain_core.output_parsers import JsonOutputParser parser JsonOutputParser() prompt ChatPromptTemplate.from_messages([ (system, 输出 JSON 格式。\n{format_instructions}), (human, {input}), ]).partial(format_instructionsparser.get_format_instructions()) chain prompt | llm | parser result chain.invoke({input: ...}) # result 是 dict它的解析逻辑比 Pydantic 宽松一些内部会尝试从文本里提取 JSON 块即使模型在 JSON 前后加了废话也能捞出来。但宽松也意味着不可控模型可能输出一个字段名拼错的 JSONJsonOutputParser 照样解析成功错误要到下游用的时候才暴露。我的经验是如果结构固定优先用 PydanticOutputParser把校验前移。如果结构确实灵活用 JsonOutputParser但一定要在拿到 dict 后自己再做一层校验别直接信任。3.3 StructuredOutputParser多字段场景的折中方案StructuredOutputParser 是介于两者之间的方案。你用ResponseSchema定义字段它生成格式说明解析出来是 dict。from langchain.output_parsers import StructuredOutputParser, ResponseSchema schemas [ ResponseSchema(namesummary, description内容摘要), ResponseSchema(namesentiment, description情感倾向positive/negative/neutral), ResponseSchema(namekeywords, description关键词逗号分隔), ] parser StructuredOutputParser.from_response_schemas(schemas)它和 JsonOutputParser 的区别在于StructuredOutputParser 会强制要求模型按指定的字段名输出格式说明更明确。JsonOutputParser 更自由模型可能自己发明字段名。三个解析器的选型可以总结成一张表解析器类型安全格式约束适用场景解析失败率PydanticOutputParser强严格结构固定、需要类型校验中JsonOutputParser弱宽松结构灵活、快速原型低但错误隐蔽StructuredOutputParser中中等多字段、字段名固定中3.4 解析失败时的重试与修复策略不管用哪个解析器解析失败都是常态。模型不是编译器它输出的 JSON 偶尔会多一个逗号、少一个引号、或者把true写成True。处理策略有三层。第一层是 prompt 层面预防。在格式说明里明确写只输出 JSON不要有任何其他文字并且给一两个示例。示例比说明管用模型模仿示例的能力很强。第二层是解析器层面兜底。LangChain 提供了OutputFixingParser它包一层 LLM解析失败时把错误信息和原始输出一起丢给模型让它修复。from langchain.output_parsers import OutputFixingParser base_parser PydanticOutputParser(pydantic_objectPersonInfo) fixing_parser OutputFixingParser.from_llm(parserbase_parser, llmllm)代价是多一次 LLM 调用延迟和成本都上去了。适合对成功率要求高、能接受额外开销的场景。第三层是业务层面降级。解析失败时返回一个默认结构或者把原始文本透传给前端让用户自己看。这层最土但最稳生产环境一定要有。4. ToolCall 与流式输出的共存方案4.1 ToolCall 的本质模型输出的是结构化参数ToolCall 看起来是个新概念本质上还是结构化输出。模型决定调用某个工具时它输出的是一段 JSON包含工具名和参数。LangChain 把这部分封装成tool_calls字段挂在 AIMessage 上。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(北京天气怎么样) print(response.tool_calls) # [{name: get_weather, args: {city: 北京}, id: call_xxx}]bind_tools会把工具的定义名字、描述、参数 schema转成模型能理解的格式塞进请求。模型返回的tool_calls里args就是结构化参数。所以 ToolCall 的可靠性本质上还是结构化输出的可靠性。4.2 流式场景下 ToolCall 参数是分片到达的这是最容易踩坑的地方。流式模式下tool_calls的参数不是一次性给你的而是一片一片拼出来的。第一片可能是{name: get_weather, args: }第二片是{args: {\ci}第三片是{args: ty\: \北}以此类推。如果你在流式过程中直接读chunk.tool_calls[0][args]拿到的是残缺的 JSON 字符串json.loads必然报错。正确做法是累积所有分片等流结束后再解析。async def stream_with_tools(chain, inputs): tool_call_buffer {} async for chunk in chain.astream(inputs): if chunk.tool_call_chunks: for tc in chunk.tool_call_chunks: idx tc.get(index, 0) if idx not in tool_call_buffer: tool_call_buffer[idx] {name: , args: , id: } if tc.get(name): tool_call_buffer[idx][name] tc[name] if tc.get(args): tool_call_buffer[idx][args] tc[args] if tc.get(id): tool_call_buffer[idx][id] tc[id] if chunk.content: yield fdata: {chunk.content}\n\n # 流结束后统一解析 for idx, tc in tool_call_buffer.items(): args json.loads(tc[args]) # 执行工具...index字段很关键。模型可能一次返回多个工具调用每个有自己的 index必须按 index 分组累积不能混在一起。这个细节在文档里不显眼但不处理就会串参数。4.3 流式过程中如何给前端反馈工具调用状态用户等工具调用时是焦虑的因为界面上什么都没发生。好的做法是把工具调用的状态也通过 SSE 推给前端让用户知道模型正在查天气。可以自定义事件类型SSE 支持event:字段yield fevent: tool_start\ndata: {json.dumps({name: tc[name]})}\n\n # 工具执行完 yield fevent: tool_end\ndata: {json.dumps({name: tc[name], result: result})}\n\n # 正常文本 yield fevent: message\ndata: {chunk.content}\n\n前端按event字段分流处理tool_start时显示正在调用 XX 工具tool_end时显示结果message时追加文本。这样整个交互过程是透明的用户不会觉得卡住了。4.4 工具执行结果回灌模型时的格式陷阱工具执行完结果要作为 ToolMessage 回灌给模型让它继续生成。这里有个格式陷阱ToolMessage 必须带上对应的tool_call_id否则模型不知道这个结果对应哪个调用。from langchain_core.messages import ToolMessage tool_messages [] for tc in response.tool_calls: result execute_tool(tc[name], tc[args]) tool_messages.append( ToolMessage(contentstr(result), tool_call_idtc[id]) )tool_call_id必须和模型返回的id完全一致。如果模型一次调了多个工具每个结果都要配对正确的 id配错了模型会混乱。这个在单工具场景下不容易出错多工具并发时就要小心。另外工具返回的内容如果是复杂结构建议序列化成 JSON 字符串再塞进content。ToolMessage 的 content 是字符串类型直接塞 dict 会被转成 Python 的 repr 格式模型读起来别扭。5. 把三者串起来一个完整的流式结构化输出链路5.1 整体架构SSE 层、Chain 层、Parser 层怎么分工把前面几块拼起来一个完整的链路是这样的FastAPI 提供 SSE 端点接收请求后启动一个 LangChain ChainChain 内部可能触发 ToolCall最终输出经过 OutputParser 结构化。SSE 层负责把过程中的每个事件文本 token、工具调用、最终结构推给前端。分工上SSE 层只管传输不关心内容语义Chain 层负责编排 LLM 和工具Parser 层负责把最终文本转成结构化对象。三层解耦的好处是换模型不影响 SSE 层换传输方式不影响 Chain 层。5.2 流式过程中做增量结构化解析的可行性有人会想能不能在流式过程中就做增量解析边收边出结构化字段技术上可行但很麻烦。JSON 是上下文相关的一个字段的值没结束时你无法确定它是什么类型。比如{age: 2后面可能是8}也可能是8.5}前者是 int 后者是 float。有个取巧的办法是用支持流式解析的库比如ijson它能处理不完整的 JSON。但收益有限因为大部分业务场景下用户要的是最终结果中间过程用文本展示就够了。我的建议是流式阶段只做文本展示和工具状态反馈结构化解析放到流结束后统一做。这样逻辑简单出错也好排查。5.3 完整代码从请求到结构化结果的端到端实现import json import asyncio from fastapi import FastAPI from fastapi.responses import StreamingResponse from langchain_core.prompts import ChatPromptTemplate from langchain_core.output_parsers import PydanticOutputParser from langchain_core.messages import ToolMessage from pydantic import BaseModel, Field app FastAPI() class ExtractResult(BaseModel): title: str Field(description标题) tags: list[str] Field(description标签列表) parser PydanticOutputParser(pydantic_objectExtractResult) prompt ChatPromptTemplate.from_messages([ (system, 提取信息。\n{format_instructions}), (human, {input}), ]).partial(format_instructionsparser.get_format_instructions()) async def run_chain(input_text: str): chain prompt | llm_with_tools tool_buffer {} full_text async for chunk in chain.astream({input: input_text}): if chunk.content: full_text chunk.content yield fevent: message\ndata: {json.dumps({text: chunk.content})}\n\n if chunk.tool_call_chunks: for tc in chunk.tool_call_chunks: idx tc.get(index, 0) buf tool_buffer.setdefault(idx, {name: , args: , id: }) buf[name] tc.get(name) or buf[args] tc.get(args) or buf[id] tc.get(id) or buf[id] yield fevent: tool_delta\ndata: {json.dumps({name: buf[name]})}\n\n # 处理工具调用 if tool_buffer: tool_messages [] for idx, tc in tool_buffer.items(): args json.loads(tc[args]) result execute_tool(tc[name], args) tool_messages.append(ToolMessage(contentstr(result), tool_call_idtc[id])) yield fevent: tool_end\ndata: {json.dumps({name: tc[name], result: str(result)})}\n\n # 回灌模型继续生成 follow_up await llm_with_tools.ainvoke( prompt.format_messages(inputinput_text) tool_messages ) full_text follow_up.content yield fevent: message\ndata: {json.dumps({text: full_text})}\n\n # 最终结构化解析 try: parsed parser.parse(full_text) yield fevent: result\ndata: {parsed.model_dump_json()}\n\n except Exception as e: yield fevent: error\ndata: {json.dumps({msg: str(e)})}\n\n yield event: done\ndata: [DONE]\n\n app.post(/extract) async def extract(payload: dict): return StreamingResponse( run_chain(payload[input]), media_typetext/event-stream, headers{X-Accel-Buffering: no, Cache-Control: no-cache}, )这段代码把前面讲的所有点都串起来了心跳没加可以按需补、工具参数按 index 累积、工具结果带 tool_call_id 回灌、最终统一解析。实际用时还要加上异常处理和日志。5.4 实测中的性能与稳定性观察实测下来有几个观察值得分享。第一SSE 的首字节延迟主要取决于模型的首 token 时间和 SSE 封装本身关系不大。如果首字节慢先查模型而不是查 SSE 代码。第二工具调用会显著拉长总时长因为多了一次 LLM 往返。如果工具执行本身也慢用户等待时间会很难看。建议给工具执行加超时超时就返回一个查询超时的结果让模型继续而不是一直挂着。第三结构化解析的失败率在流式场景下比非流式略高。原因是流式拼接的文本偶尔会有细微的截断问题尤其是网络抖动时。所以流式场景下更要做好解析失败的兜底。6. 几个容易翻车的细节和我的处理习惯6.1 中文编码在 SSE 分片时的乱码问题前面提过 TextDecoder 的stream: true这里再强调一次服务端侧。Python 里如果手动拼接字节再 decode同样要注意。用yield fdata: {text}\n\n这种字符串 yieldFastAPI 会自己处理编码一般没问题。但如果你在中间做了text.encode().decode()之类的操作就可能把多字节字符切断。我的习惯是全程用 str不碰 bytes让框架处理编码。只有在明确需要控制字节流时才手动处理且一定用增量解码器。6.2 工具参数里嵌套对象时的解析顺序工具参数如果是嵌套结构比如{filter: {city: 北京, date: 2024-01-01}}流式分片可能在任何位置切断。累积的时候是按字符串拼所以顺序天然是对的只要 index 分组正确就没问题。但解析时要注意json.loads要求完整 JSON所以必须等所有分片到齐。判断到齐的标志是流结束不是某个特定字段出现。6.3 多工具并发调用时的 index 管理模型一次返回多个工具调用时tool_call_chunks里的每个 chunk 都带 index。不同模型的 index 行为不完全一致有的从 0 开始连续有的可能跳号。稳妥的做法是用 dict 按 index 存不要用 list 按顺序 append否则跳号时会错位。6.4 生产环境必须加的几道保险第一道是超时。整个请求要有总超时工具执行要有单独超时LLM 调用也要有超时。任何一层没超时都可能把连接挂死。第二道是限流。SSE 长连接很占资源没有限流的话并发一高就崩。按用户或 IP 做并发数限制。第三道是日志。流式请求的日志要记录请求开始、首字节时间、工具调用、解析结果、异常。出问题时这些日志是唯一的线索因为流式请求很难复现。第四道是降级。解析失败、工具超时、模型报错都要有对应的降级响应不能让前端一直转圈。这几道保险加完整个链路的稳定性会有质的提升。我自己的项目里加之前线上偶发卡死加之后基本没再出现过连接层面的问题。 LangChain流式输出与结构化输出实战：SSE、OutputParser与ToolCall整合
1. 为什么流式输出和结构化输出总是打架做过大模型应用的人多半遇到过这个场景前端用 SSE 接收流式内容一个字一个字往外蹦体验很顺滑可一旦业务需要拿到结构化的 JSON比如要提取字段、要触发下游工具调用流式就变成了麻烦。你没法在流还没结束时解析 JSON因为 JSON 是残缺的可如果等流结束再解析那流式的意义又没了一半。这个矛盾在 LangChain 里体现得特别明显。LangChain 的 OutputParser 体系天生是为拿到完整输出再解析设计的而 SSE 流式是边生成边消费。两者要配合就得在中间做文章。再加上 ToolCall 这一层——模型不只是输出文本还要决定调用哪个工具、传什么参数——结构化输出的要求就更高了因为工具调用的参数必须是合法 JSON错一个括号整个调用就废了。我最近在一个基于 FastAPI LangChain 的项目里把这三块彻底捋了一遍SSE 流式接口怎么封装、OutputParser 怎么选、ToolCall 怎么和流式共存。踩的坑不少有些是 LangChain 版本差异导致的有些是 SSE 协议本身的特性决定的。这篇就把整个实战过程拆开讲从最基础的流式封装讲到 ToolCall 的流式解析尽量把每个为什么这么设计说清楚。适合的读者是已经用过 LangChain 基础功能、想把它接到真实 Web 服务里的人。如果你还在纠结 LangChain 怎么入门建议先把 Chain 和 Prompt 跑通再来看这篇不然有些细节会显得突兀。2. SSE 流式接口的封装逻辑与常见断流问题2.1 SSE 到底是个什么东西为什么大模型场景偏爱它SSE 全称 Server-Sent Events是 HTTP 协议下的一种单向推送机制。服务端保持一个长连接持续往客户端写data: xxx\n\n格式的文本块客户端用 EventSource 或者 fetch 的 ReadableStream 来读。它和 WebSocket 的区别在于SSE 是单向的服务端到客户端基于纯 HTTP不需要额外的协议升级浏览器原生支持自动重连。大模型场景偏爱 SSE 的原因很实际。第一大模型的输出本身就是单向的用户发一次请求模型吐一串 token不需要双向通信。第二SSE 基于 HTTP能直接穿过大多数网关和负载均衡部署成本低。第三SSE 的文本格式天然适合传 token每个 token 包一个data:就行。但 SSE 有个容易被忽略的点它是纯文本协议没有二进制帧的概念。这意味着你不能在里面塞复杂的二进制结构所有东西都得序列化成字符串。这一点在后面讲结构化输出时会变成一个关键约束。2.2 用 FastAPI 封装 SSE 接口的最小可用骨架先给一个能跑的最小骨架。FastAPI 里返回 SSE 用StreamingResponse媒体类型设成text/event-stream。from fastapi import FastAPI from fastapi.responses import StreamingResponse import asyncio app FastAPI() async def event_generator(): for i in range(5): yield fdata: chunk-{i}\n\n await asyncio.sleep(0.5) yield data: [DONE]\n\n app.get(/stream) async def stream(): return StreamingResponse( event_generator(), media_typetext/event-stream, headers{ Cache-Control: no-cache, Connection: keep-alive, X-Accel-Buffering: no, }, )这里有几个细节值得说。X-Accel-Buffering: no这个头是给 Nginx 看的告诉它不要缓冲这个响应。如果你部署在 Nginx 后面又没加这个头会发现流式变成了攒一批再发前端看起来就是卡顿的。Cache-Control: no-cache防止中间层缓存。Connection: keep-alive保持长连接。data:后面跟内容然后必须有两个换行\n\n这是 SSE 协议的帧分隔符。少一个换行客户端就认为这一帧还没结束会一直等。这个坑我见过不止一次尤其是拼接字符串时手滑。2.3 stream disconnected before completion 的根因排查热词里有个很典型的报错stream disconnected before completion: idle timeout waiting for sse。这个错误的字面意思是SSE 连接在完成前断开了因为等待 SSE 时空闲超时了。根因通常有三个方向。第一是服务端或中间层设了空闲超时。Nginx 默认的proxy_read_timeout是 60 秒如果模型生成慢60 秒内没有任何数据写出连接就被掐了。解决办法是在 Nginx 配置里调大这个值或者更优雅的做法是定期发送心跳包。async def event_generator_with_heartbeat(chain, inputs): queue asyncio.Queue() async def produce(): async for chunk in chain.astream(inputs): await queue.put(chunk) await queue.put(None) producer asyncio.create_task(produce()) while True: try: chunk await asyncio.wait_for(queue.get(), timeout15) except asyncio.TimeoutError: yield : heartbeat\n\n continue if chunk is None: break yield fdata: {chunk}\n\n await producer心跳包用:开头这是 SSE 的注释语法客户端会忽略它但它能让连接保持活跃避免空闲超时。15 秒发一次是个比较稳妥的值比大多数默认超时都短。第二个方向是客户端主动断开。用户切了页面、关了标签EventSource 会断开服务端如果还在往一个已断开的连接写数据就会抛异常。这个要在生成器里捕获asyncio.CancelledError或者ConnectionResetError做好清理。第三个方向是代理层。有些云厂商的负载均衡对长连接有硬性时长限制比如 5 分钟强制断开。这种只能靠客户端重连 服务端记录断点来续传或者干脆改成轮询。选型时要先确认部署环境的限制。2.4 前端消费 SSE 的两种姿势与踩坑前端消费 SSE 有两种主流方式。一种是用原生EventSource简单但有局限它只支持 GET 请求不能自定义请求头没法传 Authorization。另一种是用fetchReadableStream灵活但需要自己解析帧。async function consumeSSE(url, body) { const resp await fetch(url, { method: POST, headers: { Content-Type: application/json }, body: JSON.stringify(body), }); const reader resp.body.getReader(); const decoder new TextDecoder(); let buffer ; while (true) { const { done, value } 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: )) { const data part.slice(6); if (data [DONE]) return; handleChunk(data); } } } }这里的关键是buffer的处理。网络传输是分片的一个 SSE 帧可能被切成两半到达所以不能假设每次read()拿到的就是完整帧。正确做法是累积到 buffer按\n\n切分最后一段可能不完整留在 buffer 里等下一片。这个逻辑写错了就会出现 JSON 解析到一半报错的情况而且报错是间歇性的很难复现。decoder.decode(value, { stream: true })里的stream: true也很重要。UTF-8 一个汉字占 3 字节如果分片正好切在汉字中间不加这个参数会解码出乱码。加了之后 TextDecoder 会保留不完整的字节序列等下一片拼上再解码。3. LangChain 三大 OutputParser 的选型与实战差异3.1 PydanticOutputParser强类型场景的首选LangChain 的 OutputParser 家族里PydanticOutputParser 是最重的一个也是结构化要求最高时的首选。它的工作方式是你定义一个 Pydantic 模型它自动生成一段格式说明塞进 prompt模型按这个格式输出解析器再把文本转回 Pydantic 对象。from langchain_core.output_parsers import PydanticOutputParser from langchain_core.prompts import ChatPromptTemplate from pydantic import BaseModel, Field class PersonInfo(BaseModel): name: str Field(description人物姓名) age: int Field(description年龄整数) skills: list[str] Field(description技能列表) parser PydanticOutputParser(pydantic_objectPersonInfo) prompt ChatPromptTemplate.from_messages([ (system, 从文本中提取人物信息。\n{format_instructions}), (human, {input}), ]).partial(format_instructionsparser.get_format_instructions()) chain prompt | llm | parser result chain.invoke({input: 张三今年 28 岁会 Python 和 Go。})get_format_instructions()会生成一段类似请输出符合以下 JSON Schema 的内容的说明。这段说明的质量直接决定解析成功率。Pydantic 的Field(description...)会变成 JSON Schema 里的 description模型靠这个理解每个字段的含义。所以字段描述写得越清楚解析越稳。PydanticOutputParser 的优势是类型安全。解析出来的对象有完整的类型校验age是 int 不是 strskills是 list 不是字符串。下游代码可以直接用属性访问不用手动转换。缺点是它对格式要求严格模型稍微跑偏就解析失败而且失败时抛的是OutputParserException需要自己捕获处理。3.2 JsonOutputParser轻量灵活但需要自己兜底JsonOutputParser 比 Pydantic 轻它不要求你定义模型直接输出 dict。适合结构不固定、或者你懒得定义模型的场景。from langchain_core.output_parsers import JsonOutputParser parser JsonOutputParser() prompt ChatPromptTemplate.from_messages([ (system, 输出 JSON 格式。\n{format_instructions}), (human, {input}), ]).partial(format_instructionsparser.get_format_instructions()) chain prompt | llm | parser result chain.invoke({input: ...}) # result 是 dict它的解析逻辑比 Pydantic 宽松一些内部会尝试从文本里提取 JSON 块即使模型在 JSON 前后加了废话也能捞出来。但宽松也意味着不可控模型可能输出一个字段名拼错的 JSONJsonOutputParser 照样解析成功错误要到下游用的时候才暴露。我的经验是如果结构固定优先用 PydanticOutputParser把校验前移。如果结构确实灵活用 JsonOutputParser但一定要在拿到 dict 后自己再做一层校验别直接信任。3.3 StructuredOutputParser多字段场景的折中方案StructuredOutputParser 是介于两者之间的方案。你用ResponseSchema定义字段它生成格式说明解析出来是 dict。from langchain.output_parsers import StructuredOutputParser, ResponseSchema schemas [ ResponseSchema(namesummary, description内容摘要), ResponseSchema(namesentiment, description情感倾向positive/negative/neutral), ResponseSchema(namekeywords, description关键词逗号分隔), ] parser StructuredOutputParser.from_response_schemas(schemas)它和 JsonOutputParser 的区别在于StructuredOutputParser 会强制要求模型按指定的字段名输出格式说明更明确。JsonOutputParser 更自由模型可能自己发明字段名。三个解析器的选型可以总结成一张表解析器类型安全格式约束适用场景解析失败率PydanticOutputParser强严格结构固定、需要类型校验中JsonOutputParser弱宽松结构灵活、快速原型低但错误隐蔽StructuredOutputParser中中等多字段、字段名固定中3.4 解析失败时的重试与修复策略不管用哪个解析器解析失败都是常态。模型不是编译器它输出的 JSON 偶尔会多一个逗号、少一个引号、或者把true写成True。处理策略有三层。第一层是 prompt 层面预防。在格式说明里明确写只输出 JSON不要有任何其他文字并且给一两个示例。示例比说明管用模型模仿示例的能力很强。第二层是解析器层面兜底。LangChain 提供了OutputFixingParser它包一层 LLM解析失败时把错误信息和原始输出一起丢给模型让它修复。from langchain.output_parsers import OutputFixingParser base_parser PydanticOutputParser(pydantic_objectPersonInfo) fixing_parser OutputFixingParser.from_llm(parserbase_parser, llmllm)代价是多一次 LLM 调用延迟和成本都上去了。适合对成功率要求高、能接受额外开销的场景。第三层是业务层面降级。解析失败时返回一个默认结构或者把原始文本透传给前端让用户自己看。这层最土但最稳生产环境一定要有。4. ToolCall 与流式输出的共存方案4.1 ToolCall 的本质模型输出的是结构化参数ToolCall 看起来是个新概念本质上还是结构化输出。模型决定调用某个工具时它输出的是一段 JSON包含工具名和参数。LangChain 把这部分封装成tool_calls字段挂在 AIMessage 上。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(北京天气怎么样) print(response.tool_calls) # [{name: get_weather, args: {city: 北京}, id: call_xxx}]bind_tools会把工具的定义名字、描述、参数 schema转成模型能理解的格式塞进请求。模型返回的tool_calls里args就是结构化参数。所以 ToolCall 的可靠性本质上还是结构化输出的可靠性。4.2 流式场景下 ToolCall 参数是分片到达的这是最容易踩坑的地方。流式模式下tool_calls的参数不是一次性给你的而是一片一片拼出来的。第一片可能是{name: get_weather, args: }第二片是{args: {\ci}第三片是{args: ty\: \北}以此类推。如果你在流式过程中直接读chunk.tool_calls[0][args]拿到的是残缺的 JSON 字符串json.loads必然报错。正确做法是累积所有分片等流结束后再解析。async def stream_with_tools(chain, inputs): tool_call_buffer {} async for chunk in chain.astream(inputs): if chunk.tool_call_chunks: for tc in chunk.tool_call_chunks: idx tc.get(index, 0) if idx not in tool_call_buffer: tool_call_buffer[idx] {name: , args: , id: } if tc.get(name): tool_call_buffer[idx][name] tc[name] if tc.get(args): tool_call_buffer[idx][args] tc[args] if tc.get(id): tool_call_buffer[idx][id] tc[id] if chunk.content: yield fdata: {chunk.content}\n\n # 流结束后统一解析 for idx, tc in tool_call_buffer.items(): args json.loads(tc[args]) # 执行工具...index字段很关键。模型可能一次返回多个工具调用每个有自己的 index必须按 index 分组累积不能混在一起。这个细节在文档里不显眼但不处理就会串参数。4.3 流式过程中如何给前端反馈工具调用状态用户等工具调用时是焦虑的因为界面上什么都没发生。好的做法是把工具调用的状态也通过 SSE 推给前端让用户知道模型正在查天气。可以自定义事件类型SSE 支持event:字段yield fevent: tool_start\ndata: {json.dumps({name: tc[name]})}\n\n # 工具执行完 yield fevent: tool_end\ndata: {json.dumps({name: tc[name], result: result})}\n\n # 正常文本 yield fevent: message\ndata: {chunk.content}\n\n前端按event字段分流处理tool_start时显示正在调用 XX 工具tool_end时显示结果message时追加文本。这样整个交互过程是透明的用户不会觉得卡住了。4.4 工具执行结果回灌模型时的格式陷阱工具执行完结果要作为 ToolMessage 回灌给模型让它继续生成。这里有个格式陷阱ToolMessage 必须带上对应的tool_call_id否则模型不知道这个结果对应哪个调用。from langchain_core.messages import ToolMessage tool_messages [] for tc in response.tool_calls: result execute_tool(tc[name], tc[args]) tool_messages.append( ToolMessage(contentstr(result), tool_call_idtc[id]) )tool_call_id必须和模型返回的id完全一致。如果模型一次调了多个工具每个结果都要配对正确的 id配错了模型会混乱。这个在单工具场景下不容易出错多工具并发时就要小心。另外工具返回的内容如果是复杂结构建议序列化成 JSON 字符串再塞进content。ToolMessage 的 content 是字符串类型直接塞 dict 会被转成 Python 的 repr 格式模型读起来别扭。5. 把三者串起来一个完整的流式结构化输出链路5.1 整体架构SSE 层、Chain 层、Parser 层怎么分工把前面几块拼起来一个完整的链路是这样的FastAPI 提供 SSE 端点接收请求后启动一个 LangChain ChainChain 内部可能触发 ToolCall最终输出经过 OutputParser 结构化。SSE 层负责把过程中的每个事件文本 token、工具调用、最终结构推给前端。分工上SSE 层只管传输不关心内容语义Chain 层负责编排 LLM 和工具Parser 层负责把最终文本转成结构化对象。三层解耦的好处是换模型不影响 SSE 层换传输方式不影响 Chain 层。5.2 流式过程中做增量结构化解析的可行性有人会想能不能在流式过程中就做增量解析边收边出结构化字段技术上可行但很麻烦。JSON 是上下文相关的一个字段的值没结束时你无法确定它是什么类型。比如{age: 2后面可能是8}也可能是8.5}前者是 int 后者是 float。有个取巧的办法是用支持流式解析的库比如ijson它能处理不完整的 JSON。但收益有限因为大部分业务场景下用户要的是最终结果中间过程用文本展示就够了。我的建议是流式阶段只做文本展示和工具状态反馈结构化解析放到流结束后统一做。这样逻辑简单出错也好排查。5.3 完整代码从请求到结构化结果的端到端实现import json import asyncio from fastapi import FastAPI from fastapi.responses import StreamingResponse from langchain_core.prompts import ChatPromptTemplate from langchain_core.output_parsers import PydanticOutputParser from langchain_core.messages import ToolMessage from pydantic import BaseModel, Field app FastAPI() class ExtractResult(BaseModel): title: str Field(description标题) tags: list[str] Field(description标签列表) parser PydanticOutputParser(pydantic_objectExtractResult) prompt ChatPromptTemplate.from_messages([ (system, 提取信息。\n{format_instructions}), (human, {input}), ]).partial(format_instructionsparser.get_format_instructions()) async def run_chain(input_text: str): chain prompt | llm_with_tools tool_buffer {} full_text async for chunk in chain.astream({input: input_text}): if chunk.content: full_text chunk.content yield fevent: message\ndata: {json.dumps({text: chunk.content})}\n\n if chunk.tool_call_chunks: for tc in chunk.tool_call_chunks: idx tc.get(index, 0) buf tool_buffer.setdefault(idx, {name: , args: , id: }) buf[name] tc.get(name) or buf[args] tc.get(args) or buf[id] tc.get(id) or buf[id] yield fevent: tool_delta\ndata: {json.dumps({name: buf[name]})}\n\n # 处理工具调用 if tool_buffer: tool_messages [] for idx, tc in tool_buffer.items(): args json.loads(tc[args]) result execute_tool(tc[name], args) tool_messages.append(ToolMessage(contentstr(result), tool_call_idtc[id])) yield fevent: tool_end\ndata: {json.dumps({name: tc[name], result: str(result)})}\n\n # 回灌模型继续生成 follow_up await llm_with_tools.ainvoke( prompt.format_messages(inputinput_text) tool_messages ) full_text follow_up.content yield fevent: message\ndata: {json.dumps({text: full_text})}\n\n # 最终结构化解析 try: parsed parser.parse(full_text) yield fevent: result\ndata: {parsed.model_dump_json()}\n\n except Exception as e: yield fevent: error\ndata: {json.dumps({msg: str(e)})}\n\n yield event: done\ndata: [DONE]\n\n app.post(/extract) async def extract(payload: dict): return StreamingResponse( run_chain(payload[input]), media_typetext/event-stream, headers{X-Accel-Buffering: no, Cache-Control: no-cache}, )这段代码把前面讲的所有点都串起来了心跳没加可以按需补、工具参数按 index 累积、工具结果带 tool_call_id 回灌、最终统一解析。实际用时还要加上异常处理和日志。5.4 实测中的性能与稳定性观察实测下来有几个观察值得分享。第一SSE 的首字节延迟主要取决于模型的首 token 时间和 SSE 封装本身关系不大。如果首字节慢先查模型而不是查 SSE 代码。第二工具调用会显著拉长总时长因为多了一次 LLM 往返。如果工具执行本身也慢用户等待时间会很难看。建议给工具执行加超时超时就返回一个查询超时的结果让模型继续而不是一直挂着。第三结构化解析的失败率在流式场景下比非流式略高。原因是流式拼接的文本偶尔会有细微的截断问题尤其是网络抖动时。所以流式场景下更要做好解析失败的兜底。6. 几个容易翻车的细节和我的处理习惯6.1 中文编码在 SSE 分片时的乱码问题前面提过 TextDecoder 的stream: true这里再强调一次服务端侧。Python 里如果手动拼接字节再 decode同样要注意。用yield fdata: {text}\n\n这种字符串 yieldFastAPI 会自己处理编码一般没问题。但如果你在中间做了text.encode().decode()之类的操作就可能把多字节字符切断。我的习惯是全程用 str不碰 bytes让框架处理编码。只有在明确需要控制字节流时才手动处理且一定用增量解码器。6.2 工具参数里嵌套对象时的解析顺序工具参数如果是嵌套结构比如{filter: {city: 北京, date: 2024-01-01}}流式分片可能在任何位置切断。累积的时候是按字符串拼所以顺序天然是对的只要 index 分组正确就没问题。但解析时要注意json.loads要求完整 JSON所以必须等所有分片到齐。判断到齐的标志是流结束不是某个特定字段出现。6.3 多工具并发调用时的 index 管理模型一次返回多个工具调用时tool_call_chunks里的每个 chunk 都带 index。不同模型的 index 行为不完全一致有的从 0 开始连续有的可能跳号。稳妥的做法是用 dict 按 index 存不要用 list 按顺序 append否则跳号时会错位。6.4 生产环境必须加的几道保险第一道是超时。整个请求要有总超时工具执行要有单独超时LLM 调用也要有超时。任何一层没超时都可能把连接挂死。第二道是限流。SSE 长连接很占资源没有限流的话并发一高就崩。按用户或 IP 做并发数限制。第三道是日志。流式请求的日志要记录请求开始、首字节时间、工具调用、解析结果、异常。出问题时这些日志是唯一的线索因为流式请求很难复现。第四道是降级。解析失败、工具超时、模型报错都要有对应的降级响应不能让前端一直转圈。这几道保险加完整个链路的稳定性会有质的提升。我自己的项目里加之前线上偶发卡死加之后基本没再出现过连接层面的问题。