流式输出和结构化输出这两个词放在一起本身就有点矛盾。SSE 追求的是边生成边推送一个字一个字往外蹦用户体感好而结构化输出要求的是整段 JSON 完整可解析字段一个不能少、类型一个不能错。我在实际项目里踩过最典型的坑就是前端用 EventSource 接流后端用 LangChain 的 PydanticOutputParser 做解析结果流式返回的 JSON 是分片的解析器拿到半截字符串直接抛异常整个链路崩掉。后来才搞明白这两件事需要分开处理——流式负责传输体验结构化负责数据落地中间得靠 OutputParser 和 ToolCall 做桥接。这篇内容适合正在用 LangChain 做 AI 应用、被流式输出和结构化解析同时折磨的开发者。不管你是刚接触 LangChain 的新手还是已经写过几版 Agent 但总在解析环节翻车的老手下面这些从实战里抠出来的细节应该都能对上你的痛点。我会把 SSE 流式的底层机制、LangChain 三大 OutputParser 的适用边界、ToolCall 的结构化方案以及它们怎么串起来用全部拆开讲清楚。1. 先搞清楚 SSE 流式到底在传什么很多人一上来就写EventSource接流但根本没想过 SSE 协议层到底在传什么格式的数据。这就像你开车不看仪表盘油没了才知道慌。1.1 SSE 的数据帧格式与 LangChain 的流式适配SSE 的全称是 Server-Sent Events它本质上就是一个长连接服务端往客户端持续推送文本。每一条消息的格式是固定的event: message data: {content: 你} data: {content: 好}注意那个空行它是消息分隔符。没有空行浏览器端的 EventSource 就不会触发onmessage。我在早期项目里自己手写 SSE 服务端时就是因为忘了加空行前端一直收不到消息排查了两个小时才发现是格式问题。LangChain 的流式输出最终也是要转成这个格式。它的astream方法返回的是一个异步生成器每次 yield 出来的是一个AIMessageChunk对象。你需要自己把它序列化成 SSE 帧。常见的做法是用 FastAPI 的StreamingResponsefrom fastapi import FastAPI from fastapi.responses import StreamingResponse from langchain_openai import ChatOpenAI app FastAPI() llm ChatOpenAI(modelgpt-4o-mini, streamingTrue) async def event_generator(): async for chunk in llm.astream(用一句话介绍 LangChain): if chunk.content: yield fdata: {chunk.content}\n\n app.get(/stream) async def stream(): return StreamingResponse(event_generator(), media_typetext/event-stream)这里有个细节media_type必须设成text/event-stream否则浏览器不会把它当成 SSE 流来处理。另外chunk.content要做空判断因为有些 chunk 只包含元数据没有实际文本。1.2 流式场景下结构化输出的核心矛盾为什么流式和结构化输出会打架因为 JSON 的语法结构决定了它不能像自然语言那样随意分片。你想想{name: 张这半截 JSON 是没法解析的解析器会直接报JSONDecodeError。我见过有人试图在流式过程中不断尝试解析累积的字符串一旦解析成功就返回。这个思路理论上可行但实际用起来问题很多第一JSON 的嵌套结构可能导致部分解析成功但字段不完整第二每次 chunk 到达都做一次完整解析性能开销随文本长度线性增长第三如果模型输出的 JSON 格式有轻微偏差累积解析会一直失败直到超时。提示流式场景下不要试图对每个 chunk 做结构化解析正确的做法是流式传输原始文本等流结束后再做一次完整的结构化解析。那有没有办法既流式又结构化有但需要换思路。LangChain 提供了几种方案下面会逐一拆解。1.3 前端消费 SSE 流的常见坑前端用 EventSource 接流时有几个坑几乎每个人都会踩第一个坑是自动重连。EventSource 默认在连接断开后会自动重连如果你的接口不支持断点续传重连后模型会重新生成一遍用户看到重复内容。解决办法是在服务端发送完最后一条消息后主动发送一个event: done的事件前端收到后调用eventSource.close()。第二个坑是中文乱码。SSE 默认使用 UTF-8 编码但如果你的服务端没有正确设置charsetutf-8中文会变成乱码。FastAPI 的StreamingResponse默认就是 UTF-8但如果你用的是 Flask 或其他框架需要手动指定。第三个坑是代理缓冲。某些反向代理会缓冲 SSE 流导致前端收到消息的延迟很大。解决办法是在响应头里加上X-Accel-Buffering: no告诉代理不要缓冲。const eventSource new EventSource(/stream); let buffer ; eventSource.onmessage (event) { buffer event.data; document.getElementById(output).textContent buffer; }; eventSource.addEventListener(done, () { eventSource.close(); // 流结束后把 buffer 发给后端做结构化解析 parseStructured(buffer); });这个模式是我目前用得最顺手的流式阶段只做展示流结束后再调一次结构化解析接口。用户体验和数据准确性都能兼顾。2. LangChain 三大 OutputParser 的适用边界LangChain 的 OutputParser 是结构化输出的核心工具但很多人只知道 PydanticOutputParser不知道另外两个的存在更不知道它们各自的适用场景。选错了 Parser轻则解析失败率飙升重则整个链路不可用。2.1 PydanticOutputParser强类型场景的首选PydanticOutputParser 是三个里面最严格的它要求模型输出的 JSON 必须完全符合你定义的 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}) ]) chain prompt | llm | parser result chain.invoke({ input: 张三今年28岁会Python和Java, format_instructions: parser.get_format_instructions() })get_format_instructions()这个方法很关键它会把 Pydantic 模型的 JSON Schema 转成一段自然语言描述塞进 prompt 里告诉模型该怎么输出。很多人忘了加这个结果模型输出的字段名跟模型定义对不上解析直接失败。PydanticOutputParser 适合什么场景适合字段固定、类型明确、不允许有额外字段的场景。比如从简历里提取姓名、年龄、技能从合同里提取甲方、乙方、金额。这些场景下数据质量要求高宁可解析失败也不能出错。但它也有明显的短板对模型能力要求高。如果你用的是小参数量的本地模型它经常输出不符合 Schema 的 JSON解析失败率可能超过 30%。这时候就需要考虑下面两种方案。2.2 StructuredOutputParser灵活性与结构化的折中StructuredOutputParser 不像 PydanticOutputParser 那样要求严格的类型定义它只需要你告诉它有哪些字段、每个字段是什么类型然后它会生成一个 ResponseSchema 列表。from langchain.output_parsers import StructuredOutputParser, ResponseSchema schemas [ ResponseSchema(nameproduct_name, description产品名称), ResponseSchema(nameprice, description产品价格数字类型), ResponseSchema(nametags, description产品标签逗号分隔的字符串) ] parser StructuredOutputParser.from_response_schemas(schemas) format_instructions parser.get_format_instructions() prompt ChatPromptTemplate.from_template( 从以下描述中提取产品信息\n{input}\n\n{format_instructions} ) chain prompt | llm | parser result chain.invoke({ input: 这款无线蓝牙耳机售价299元主打降噪和长续航, format_instructions: format_instructions })StructuredOutputParser 的输出是一个字典字段值都是字符串类型即使你描述里写了数字类型它返回的也是字符串。这个特性在某些场景下反而是优势——你不需要提前定义严格的类型拿到结果后自己做类型转换就行。它的适用场景是字段不固定、需要动态调整、或者模型能力有限无法稳定输出严格 JSON 的情况。比如做一个通用的信息抽取工具用户自己定义要抽取哪些字段这时候用 StructuredOutputParser 就比 PydanticOutputParser 灵活得多。但要注意它的解析容错性虽然比 PydanticOutputParser 好但也不是万能的。如果模型输出的 JSON 缺少某个字段它还是会报错。我一般的做法是在 prompt 里加一句如果某个字段无法从原文中提取请填写未知这样能大幅降低解析失败率。2.3 OutputFunctionsParserToolCall 场景的底层支撑OutputFunctionsParser 是三个里面最特殊的一个它不解析 JSON 文本而是解析模型返回的 function call 参数。在 OpenAI 的 Function Calling 机制下模型不会直接输出 JSON 文本而是返回一个function_call对象里面包含函数名和参数。from langchain.output_parsers.openai_functions import OutputFunctionsParser from langchain_core.prompts import ChatPromptTemplate function_schema { name: extract_info, description: 提取人物信息, parameters: { type: object, properties: { name: {type: string, description: 姓名}, age: {type: integer, description: 年龄} }, required: [name, age] } } parser OutputFunctionsParser() prompt ChatPromptTemplate.from_template(提取以下文本中的人物信息{input}) chain prompt | llm.bind(functions[function_schema]) | parser result chain.invoke({input: 李四35岁工程师})OutputFunctionsParser 的优势在于它绕过了 JSON 文本解析这一步。模型返回的 function call 参数已经是结构化的对象了不需要再做字符串解析。这就从根本上避免了 JSON 格式错误的问题。但它的限制也很明显依赖模型支持 Function Calling。OpenAI 的 GPT 系列、Claude、以及部分开源模型支持这个能力但很多小模型不支持。另外Function Calling 的输出是全有或全无的模型要么返回完整的参数要么不返回没法做流式输出。2.4 三种 Parser 的选型对照维度PydanticOutputParserStructuredOutputParserOutputFunctionsParser类型严格度高强类型校验中字段值为字符串高由 Schema 定义模型要求高需要较强 JSON 能力中容错性较好需要支持 Function Calling流式支持不支持不支持不支持适用场景字段固定的信息提取动态字段的通用抽取工具调用、Agent 场景解析失败率较高小模型上中等低依赖模型能力选型逻辑其实很简单如果你的模型支持 Function Calling 且场景适合优先用 OutputFunctionsParser如果需要严格的类型校验且模型能力够强用 PydanticOutputParser如果字段不固定或者模型能力有限用 StructuredOutputParser。3. ToolCall 方案让模型主动输出结构化数据ToolCall工具调用是 LangChain 里做结构化输出的另一条路。它的思路跟 OutputParser 不同不是让模型输出 JSON 文本再解析而是让模型直接调用一个预定义的工具工具的参数就是结构化的数据。3.1 ToolCall 与 Function Calling 的关系先理清一个概念Function Calling 是模型层面的能力ToolCall 是 LangChain 层面的封装。OpenAI 的 API 里叫functionsLangChain 把它包装成了Tool对象。from langchain_core.tools import tool from langchain_openai import ChatOpenAI from langchain_core.prompts import ChatPromptTemplate tool def extract_person(name: str, age: int, skills: list[str]) - dict: 提取人物信息 return {name: name, age: age, skills: skills} llm_with_tools ChatOpenAI(modelgpt-4o-mini).bind_tools([extract_person]) prompt ChatPromptTemplate.from_template(从以下文本提取人物信息{input}) chain prompt | llm_with_tools result chain.invoke({input: 王五42岁擅长数据分析和机器学习}) tool_calls result.tool_calls # tool_calls[0][args] 就是结构化数据bind_tools会把工具的定义转成模型能理解的 Schema模型在生成时会判断是否需要调用工具。如果需要它会返回一个tool_calls列表每个元素包含工具名和参数。这个方案最大的好处是模型输出的参数已经是解析好的 Python 对象不需要你再做 JSON 解析。而且 LangChain 会自动校验参数类型如果模型输出的参数类型不对会直接报错而不是静默失败。3.2 用 ToolCall 做结构化输出的完整链路实际项目里ToolCall 做结构化输出通常需要配合一个执行器来跑通完整链路。因为模型只是决定调用工具真正执行工具的是你的代码。from langchain_core.runnables import RunnablePassthrough from langchain_core.output_parsers import JsonOutputParser def execute_tool_calls(message): results [] for tool_call in message.tool_calls: tool_name tool_call[name] tool_args tool_call[args] # 这里可以根据 tool_name 分发到不同的处理逻辑 results.append({tool: tool_name, args: tool_args}) return results chain prompt | llm_with_tools | execute_tool_calls result chain.invoke({input: 赵六30岁会前端开发和UI设计})这个链路的关键在于execute_tool_calls这个函数。它接收模型返回的AIMessage提取tool_calls字段然后做后续处理。你可以把结果存数据库、发给前端、或者触发其他业务流程。注意ToolCall 返回的参数虽然已经是结构化对象但不代表它一定符合你的业务校验规则。比如模型可能返回一个负数年龄或者一个不存在的技能名称。业务层面的校验还是得自己做。3.3 ToolCall 在 Agent 场景下的结构化优势在 Agent 场景下ToolCall 的优势更加明显。Agent 需要根据用户输入决定调用哪个工具、传什么参数这本身就是结构化输出的过程。from langchain.agents import AgentExecutor, create_openai_tools_agent tools [extract_person, search_database, send_email] agent create_openai_tools_agent(llm, tools, prompt) agent_executor AgentExecutor(agentagent, toolstools) result agent_executor.invoke({input: 帮我查一下张三的信息然后发邮件给他})Agent 会自动规划先调用search_database查张三拿到结果后调用send_email发邮件。每一步的工具调用参数都是结构化的不需要你手动解析。我实测下来ToolCall 方案在 Agent 场景下的结构化输出成功率明显高于 OutputParser 方案。原因很简单模型在 Function Calling 模式下经过了专门的训练输出格式的稳定性远高于自由文本生成 JSON。3.4 ToolCall 方案的局限与应对ToolCall 不是银弹它有几个硬性限制第一模型必须支持 Function Calling。如果你用的是本地部署的小模型比如 Qwen 7B 或 Llama 3 8B它们可能不支持或者支持得不好。这种情况下只能退回 OutputParser 方案。第二ToolCall 不支持流式输出。模型要么返回完整的工具调用参数要么不返回。你没法像文本生成那样一个字一个字往外推。如果业务要求流式展示需要做特殊处理——比如流式展示正在分析...的提示等工具调用完成后再一次性展示结果。第三工具定义会增加 token 消耗。每个工具的定义都会作为 system prompt 的一部分发给模型工具越多token 消耗越大。我一般建议单次绑定的工具不超过 10 个超过的话考虑做工具分组或动态绑定。4. 流式与结构化的桥接方案前面分别讲了 SSE 流式、OutputParser 和 ToolCall现在到了最关键的部分怎么把它们串起来既保证流式体验又拿到结构化数据。4.1 两阶段方案流式展示 流后解析这是我最推荐的方案也是实际项目里用得最多的。核心思路是流式阶段只负责把模型的原始输出推给前端展示流结束后再把完整文本发给解析器做结构化处理。async def stream_and_parse(user_input: str): full_text # 第一阶段流式输出 async for chunk in llm.astream(user_input): if chunk.content: full_text chunk.content yield fdata: {chunk.content}\n\n # 第二阶段结构化解析 yield fevent: done\ndata: {json.dumps({status: stream_end})}\n\n # 解析完整文本 try: structured parser.parse(full_text) yield fevent: structured\ndata: {json.dumps(structured.dict())}\n\n except Exception as e: yield fevent: error\ndata: {json.dumps({error: str(e)})}\n\n这个方案的好处是职责清晰流式负责体验解析负责数据。前端收到done事件后知道流结束了收到structured事件后拿到结构化数据。但有个细节要注意解析失败的处理。如果模型输出的 JSON 格式有问题解析会失败。这时候不能直接报错给用户而是要有降级方案。我的做法是解析失败时把原始文本返回给前端同时记录日志后续人工排查。4.2 流式 JSON 的增量解析思路如果你确实需要在流式过程中就拿到部分结构化数据可以考虑增量解析。这个方案的核心是维护一个累积的 JSON 字符串每次收到新 chunk 后尝试解析如果解析成功就返回部分结果。import json class IncrementalJSONParser: def __init__(self): self.buffer self.parsed_keys set() def feed(self, chunk: str): self.buffer chunk # 尝试补全 JSON 并解析 try: # 简单场景下尝试在 buffer 末尾补全括号 candidate self.buffer open_braces candidate.count({) - candidate.count(}) candidate } * open_braces data json.loads(candidate) # 提取新增的字段 new_fields {k: v for k, v in data.items() if k not in self.parsed_keys} self.parsed_keys.update(new_fields.keys()) return new_fields except json.JSONDecodeError: return None这个方案在简单场景下能用但有几个明显的限制第一只适合扁平 JSON嵌套结构补全逻辑很复杂第二如果模型输出的 JSON 有语法错误增量解析会一直失败第三频繁的 JSON 解析有性能开销。我个人的建议是除非业务强需求否则不要用增量解析。两阶段方案已经能满足 90% 的场景而且稳定性和可维护性都好得多。4.3 用 ToolCall 做流式场景下的结构化兜底还有一个折中方案流式阶段用普通文本生成流结束后用 ToolCall 做结构化提取。这样流式体验有了结构化数据的准确性也有保障。async def stream_with_toolcall_fallback(user_input: str): # 第一阶段流式生成 full_text async for chunk in llm.astream(user_input): if chunk.content: full_text chunk.content yield fdata: {chunk.content}\n\n # 第二阶段用 ToolCall 做结构化提取 extraction_prompt f从以下文本中提取结构化信息\n{full_text} result await llm_with_tools.ainvoke(extraction_prompt) if result.tool_calls: structured_data result.tool_calls[0][args] yield fevent: structured\ndata: {json.dumps(structured_data)}\n\n这个方案相当于做了两次模型调用第一次生成内容第二次提取结构。成本翻倍但准确性最高。适合对数据质量要求极高的场景比如合同信息提取、财务数据录入。4.4 三种桥接方案的对比与选型方案流式体验结构化准确性成本适用场景两阶段方案好中低通用场景增量解析好低低简单扁平 JSONToolCall 兜底好高高高准确性要求选型建议先用两阶段方案跑通如果解析失败率超过 10%再考虑 ToolCall 兜底。增量解析除非有特殊需求否则不建议用。5. 实战中踩过的坑与排查链路这一节不讲理论只讲我在实际项目里踩过的坑和排查过程。这些经验在官方文档里找不到但每一个都让我多花了好几个小时。5.1 SSE 流中断idle timeout 的完整排查过程有一次线上环境用户反馈流式输出经常中断前端报错stream disconnected before completion: idle timeout waiting for sse。这个报错信息很明确SSE 连接因为空闲超时被断开了。排查第一步确认超时时间。我查了 Nginx 的配置proxy_read_timeout默认是 60 秒。如果模型生成一个长回复超过 60 秒没有新数据推送Nginx 就会断开连接。排查第二步确认模型是否真的有空闲期。我加了日志发现模型在生成过程中确实有超过 60 秒没有输出任何 chunk 的情况。原因是模型在处理复杂推理时会在内部思考很久期间不产生任何输出。排查第三步解决方案。有两个方向一是调大 Nginx 的超时时间二是让服务端定期发送心跳。我两个都做了location /stream { proxy_pass http://backend; proxy_read_timeout 300s; proxy_set_header Connection ; proxy_http_version 1.1; chunked_transfer_encoding off; proxy_buffering off; }async def event_generator(): last_heartbeat time.time() async for chunk in llm.astream(user_input): if chunk.content: yield fdata: {chunk.content}\n\n last_heartbeat time.time() elif time.time() - last_heartbeat 15: # 超过15秒没有内容发送心跳 yield f: heartbeat\n\n last_heartbeat time.time()心跳的格式是: heartbeat\n\n以冒号开头的行在 SSE 协议里是注释前端不会触发onmessage但能保持连接活跃。提示心跳间隔建议设为超时时间的 1/3 到 1/2。比如超时 60 秒心跳设 20-30 秒一次。5.2 PydanticOutputParser 解析失败的三种典型情况PydanticOutputParser 的解析失败是我遇到最多的坑总结下来有三种典型情况情况一模型输出了 markdown 代码块包裹的 JSON。模型经常输出json\n{...}\n这种格式解析器拿到带反引号的字符串直接报错。解决办法是在 prompt 里明确说直接输出 JSON不要用代码块包裹或者在解析前做一次字符串清洗def clean_json_string(text: str) - str: text text.strip() if text.startswith(json): text text[7:] if text.startswith(): text text[3:] if text.endswith(): text text[:-3] return text.strip()情况二模型输出的字段名跟定义不一致。比如你定义的是user_name模型输出的是userName。这种问题在 prompt 里加一句字段名必须严格使用以下名称能缓解但不能完全避免。更稳妥的做法是在 Pydantic 模型里加 aliasfrom pydantic import BaseModel, Field class UserInfo(BaseModel): user_name: str Field(aliasuserName) class Config: populate_by_name True情况三模型输出的类型不对。比如你定义age: int模型输出了28字符串。Pydantic 默认会尝试做类型转换但有些情况下会失败。解决办法是在 Field 里加strictFalse或者在 prompt 里强调类型。5.3 ToolCall 参数校验的边界情况ToolCall 虽然比 OutputParser 稳定但也不是没有坑。我遇到过几种边界情况情况一模型返回了工具定义里不存在的参数。比如工具定义只有name和age模型却返回了name、age、email。LangChain 默认会忽略多余参数但如果你开了严格模式会直接报错。情况二必填参数缺失。模型有时候会漏掉某个必填参数。这种情况下 LangChain 会报ValidationError。我的做法是在工具函数里给所有参数设默认值然后在业务逻辑里判断是否为空。情况三参数类型不匹配。比如定义age: int模型返回了二十八。这种在 Function Calling 模式下比较少见但一旦出现就是硬错误。解决办法是在工具函数里做类型转换和校验。5.4 流式与结构化并发的资源竞争问题最后一个坑比较隐蔽当多个用户同时请求流式接口时如果每个请求都创建一个新的 LLM 连接资源消耗会很大。我一开始没注意这个问题上线后发现并发稍微高一点就出现连接超时。解决办法是用连接池 信号量控制并发import asyncio semaphore asyncio.Semaphore(10) # 最多10个并发 async def stream_handler(user_input: str): async with semaphore: async for chunk in llm.astream(user_input): yield fdata: {chunk.content}\n\n信号量的值根据你的模型服务承载能力来定。如果是调 OpenAI 的 API一般 10-20 个并发没问题如果是本地部署的模型要根据 GPU 显存来调整。另外LangChain 的astream方法在连接断开时不会自动清理资源需要在finally块里手动关闭async def stream_handler(user_input: str): try: async for chunk in llm.astream(user_input): yield fdata: {chunk.content}\n\n finally: await llm.async_client.close()这些坑每一个都让我在深夜排查过希望你看完能少走点弯路。流式和结构化输出本身不复杂复杂的是它们之间的衔接和边界情况的处理。把上面这些方案和排查思路吃透大部分场景都能覆盖了。