1. 为什么流式输出和结构化输出总是打架做过大模型应用的人大概率都遇到过这个场景前端页面已经用上了 SSE 流式打字机效果用户看着文字一个个蹦出来体验很好结果后端突然需要从模型回复里提取一个 JSON 字段做后续业务处理整个链路就卡住了。要么等流式全部结束再解析打字机效果变成转圈三秒然后一次性出结果要么在流式过程中硬解析结果 JSON 被切成好几段解析器疯狂报错。这个矛盾的本质在于流式输出追求的是尽快把 token 吐给用户而结构化输出追求的是拿到一个完整合法的数据结构。前者是时间维度上的增量交付后者是空间维度上的完整性校验两者天然存在张力。LangChain 在这个问题上给出的答案是一整套 OutputParser 体系加上 ToolCall 机制。我前后在三个生产项目里用过这套东西踩过的坑不算少今天就把 SSE 流式、三大 OutputParserPydanticOutputParser、JsonOutputParser、StructuredOutputParser以及 ToolCall 方案的实际使用经验完整梳理一遍。不管你是刚接触 LangChain 的新手还是已经在做 Agent 二次开发的老手这篇内容应该都能帮你少走一些弯路。先明确一下这篇文章覆盖的范围从最基础的 OutputParser 选型到流式场景下如何做增量解析再到 ToolCall 的结构化输出方案最后是 SSE 接口封装和常见故障排查。代码以 Python 为主前端会简单提一下 Vue 侧的对接方式。2. 三大 OutputParser 的选型逻辑与核心差异2.1 OutputParser 到底在解决什么问题很多人第一次接触 OutputParser 会觉得多此一举模型输出的不就是字符串吗我自己写个正则或者json.loads不就行了这个想法在小规模实验阶段没问题但一旦进入生产环境就会暴露三个致命问题。第一模型输出不稳定。你让模型返回 JSON它可能给你返回json包裹的代码块可能在前面加一句好的以下是结果可能在末尾多一个逗号。自己写解析逻辑意味着你要处理无穷无尽的边界情况。第二格式错误无法自愈。当解析失败时OutputParser 可以配合OutputFixingParser自动把错误信息发回给模型让它重新生成这套重试机制自己实现起来很繁琐。第三缺少格式指令注入。OutputParser 的get_format_instructions()方法会生成一段格式说明你可以把它塞进 prompt 里告诉模型你必须按这个格式输出。这是保证输出质量的第一道防线。理解了这三点再看三大 Parser 的差异就很清晰了。2.2 PydanticOutputParser类型安全优先PydanticOutputParser是我在需要严格类型校验时首选方案。它的工作方式是你定义一个 Pydantic 模型Parser 根据模型生成 JSON Schema注入到 prompt 中模型返回后再用 Pydantic 做校验和反序列化。from langchain_core.output_parsers import PydanticOutputParser from pydantic import BaseModel, Field from typing import List class ProductReview(BaseModel): product_name: str Field(description产品名称) rating: float Field(description评分0到5之间, ge0, le5) pros: List[str] Field(description优点列表) cons: List[str] Field(description缺点列表) summary: str Field(description一句话总结) parser PydanticOutputParser(pydantic_objectProductReview)它的优势在于校验能力最强。ge0, le5这种约束会在解析时自动生效模型如果返回了 6 分直接抛ValidationError你可以捕获后触发重试。字段描述description也会被写进格式指令相当于给模型做了字段级别的提示。代价是灵活性差。模型必须严格按 Schema 输出字段少一个就报错。而且 Pydantic 的校验是全有或全无流式场景下没法做增量解析——你必须等完整 JSON 才能校验。2.3 JsonOutputParser流式场景的救星JsonOutputParser是我在流式场景下用得最多的。它不依赖 Pydantic 模型直接解析 JSON而且支持流式增量解析——这是它和 PydanticOutputParser 最本质的区别。from langchain_core.output_parsers import JsonOutputParser parser JsonOutputParser() # 流式场景下的用法 for chunk in chain.stream({query: ...}): # chunk 是逐步累积的部分 JSON print(chunk)它的内部实现用了json.loads的容错版本能在 JSON 还没闭合时返回已解析的部分字段。比如模型正在输出{name: 张三, age: 25它能先返回{name: 张三}等}到了再补上 age。这个特性对于边生成边渲染的场景太重要了。但要注意JsonOutputParser 不做类型校验。它只保证是合法 JSON字段类型对不对、有没有缺字段它不管。所以我的经验是流式阶段用 JsonOutputParser 做增量展示流式结束后再用 Pydantic 模型做一次完整校验两者配合使用。2.4 StructuredOutputParser多字段场景的轻量选择StructuredOutputParser是三者里最老派的一个它通过ResponseSchema定义字段不依赖 Pydantic。from langchain.output_parsers import StructuredOutputParser, ResponseSchema schemas [ ResponseSchema(nametitle, description文章标题), ResponseSchema(nametags, description标签逗号分隔), ResponseSchema(nameword_count, description字数), ] parser StructuredOutputParser.from_response_schemas(schemas)它的定位比较尴尬论校验不如 PydanticOutputParser论流式支持不如 JsonOutputParser。我现在基本只在两种情况下用它一是项目里不想引入 Pydantic 依赖二是字段结构非常简单、不需要复杂校验。2.5 三者对比速查维度PydanticOutputParserJsonOutputParserStructuredOutputParser类型校验强Pydantic 全量校验无弱仅字段存在性流式增量解析不支持支持不支持格式指令生成自动JSON Schema自动自动依赖pydantic无无适用场景严格结构化、非流式流式展示、宽松解析简单字段、轻量项目错误自愈配合 OutputFixingParser配合 OutputFixingParser配合 OutputFixingParser选型口诀要校验用 Pydantic要流式用 Json要轻量用 Structured。实际项目里最常见的是 Json Pydantic 组合拳。3. 流式场景下的增量解析实战3.1 SSE 流式链路的基本结构先把这个链路讲清楚不然后面聊增量解析会没有上下文。一个典型的 LangChain SSE 链路是这样的前端发起请求 → 后端 FastAPI 接收 → 调用 LangChain chain 的astream→ 每个 chunk 通过 SSE 推给前端 → 前端 EventSource 接收并渲染。后端核心代码大概长这样from fastapi import FastAPI from fastapi.responses import StreamingResponse from langchain_core.prompts import ChatPromptTemplate from langchain_core.output_parsers import JsonOutputParser from langchain_openai import ChatOpenAI app FastAPI() prompt ChatPromptTemplate.from_template( 请分析以下评论返回 JSON{review} ) model ChatOpenAI(modelgpt-4o-mini, streamingTrue) parser JsonOutputParser() chain prompt | model | parser app.get(/analyze) async def analyze(review: str): async def event_generator(): async for chunk in chain.astream({review: review}): yield fdata: {json.dumps(chunk, ensure_asciiFalse)}\n\n yield data: [DONE]\n\n return StreamingResponse(event_generator(), media_typetext/event-stream)这里有个关键点chain 末尾挂了 JsonOutputParser所以astream吐出来的 chunk 已经是解析后的 dict 片段不是原始 token。这是 LangChain 帮我们做掉的一层封装。3.2 增量解析的坑字段顺序和部分 JSONJsonOutputParser 的增量解析有个容易被忽略的细节它只能解析已经完整闭合的字段。假设模型按顺序输出{summary: 还不错, rating: 4, tags: [性价比, 物流快]}解析过程会是这样收到{summary: 还不错→ 返回{}summary 还没闭合收到{summary: 还不错,→ 返回{summary: 还不错}收到{summary: 还不错, rating: 4→ 返回{summary: 还不错}收到{summary: 还不错, rating: 4,→ 返回{summary: 还不错, rating: 4}...以此类推所以前端拿到的 chunk 是累积的、逐步补全的 dict不是增量 diff。前端渲染时要用覆盖而不是追加的逻辑。这一点我在第一个项目里搞错了导致页面上字段重复显示排查了半天。3.3 让模型按固定字段顺序输出既然增量解析依赖字段闭合顺序那控制模型的输出顺序就很有价值。有两个办法第一个是在 prompt 里明确要求顺序。比如请严格按照 summary、rating、tags 的顺序输出。实测下来模型对顺序要求的遵守率大概在 80% 左右不是 100%。第二个更可靠用 Pydantic 模型的字段定义顺序。Pydantic 生成 JSON Schema 时字段顺序就是定义顺序模型看到 Schema 后倾向于按这个顺序输出。所以把最重要的、想优先展示的字段放在模型定义的最前面。class Review(BaseModel): summary: str Field(description一句话总结) # 放最前优先展示 rating: float Field(description评分) tags: List[str] Field(description标签)3.4 流式 校验的两段式方案前面提过JsonOutputParser 不做校验。我的生产方案是两段式第一段流式用 JsonOutputParser 做增量解析前端实时渲染。这个阶段容忍字段缺失和类型错误目的是让用户看到东西在动。第二段流式结束拿到完整输出后用 Pydantic 模型做一次严格校验。校验失败就触发一次修复重试或者返回降级结果。async def stream_with_validation(review: str): full_output {} async for chunk in chain.astream({review: review}): full_output.update(chunk) yield fdata: {json.dumps(chunk, ensure_asciiFalse)}\n\n # 流式结束做严格校验 try: validated Review(**full_output) yield fdata: {json.dumps({validated: validated.model_dump()}, ensure_asciiFalse)}\n\n except ValidationError as e: yield fdata: {json.dumps({error: str(e)}, ensure_asciiFalse)}\n\n yield data: [DONE]\n\n这个方案的好处是用户体验和业务严谨性兼顾。用户看到的是流畅的流式输出业务侧拿到的是校验过的数据。注意full_output.update(chunk)这里用 update 而不是赋值是因为 chunk 是累积的 dict后一个 chunk 包含前一个的所有字段。用 update 能保证拿到最终完整结果。4. ToolCall 方案结构化输出的另一条路4.1 为什么需要 ToolCall 做结构化输出OutputParser 的思路是让模型输出文本我再解析文本。ToolCall 的思路完全不同把结构化输出定义成一个工具的参数让模型通过调用工具来输出结构化数据。这个思路的优势在于模型在训练时就专门优化过工具调用的参数生成格式遵守率比纯文本输出高得多。尤其是复杂嵌套结构ToolCall 的稳定性明显优于 prompt 里塞 JSON Schema。LangChain 里用with_structured_output就能一行搞定from langchain_openai import ChatOpenAI from pydantic import BaseModel, Field class Review(BaseModel): summary: str Field(description一句话总结) rating: float Field(description评分) tags: list[str] Field(description标签) model ChatOpenAI(modelgpt-4o-mini) structured_model model.with_structured_output(Review) result structured_model.invoke(分析这条评论东西不错物流很快) # result 直接就是 Review 实例 print(result.rating)4.2 ToolCall 和 OutputParser 的取舍两者不是替代关系而是各有适用场景。我整理了一个决策表场景推荐方案理由需要流式展示 结构化JsonOutputParserToolCall 不支持流式增量复杂嵌套结构ToolCall格式遵守率高需要严格类型校验ToolCall 或 PydanticOutputParser都支持 Pydantic多工具 Agent 场景ToolCall天然集成简单字段、快速原型StructuredOutputParser代码量最少核心结论ToolCall 不适合流式场景。因为工具调用的参数是作为一个整体生成的模型不会把参数拆成 token 流式吐出来。如果你既要流式又要结构化还是得回到 JsonOutputParser。4.3 ToolCall 在 Agent 里的实战在 Agent 场景下ToolCall 的价值更明显。比如做一个评论分析 Agent它需要先调用工具查商品信息再输出结构化分析结果from langchain_core.tools import tool tool def get_product_info(product_id: str) - dict: 根据商品ID查询商品信息 # 实际项目里这里查数据库 return {name: 某商品, category: 数码} tools [get_product_info] model_with_tools model.bind_tools(tools)这里有个经验工具的描述docstring质量直接决定模型调用准确率。我见过太多项目工具描述写得含糊导致模型该调不调、不该调乱调。描述要写清楚这个工具做什么、参数是什么、什么时候用。4.4 流式 ToolCall 的混合方案如果业务既需要流式展示思考过程又需要最终的结构化结果可以用混合方案流式输出文本部分ToolCall 输出结构化部分。具体做法是让模型先流式输出一段自然语言分析然后触发工具调用输出结构化数据。前端先渲染文本流工具调用结果到了再渲染结构化卡片。这个方案在AI 分析报告类产品里很常见。5. SSE 接口封装的工程细节5.1 后端 SSE 封装的三个关键点第一心跳保活。SSE 连接如果长时间没有数据推送中间的网络设备负载均衡、网关可能会主动断开。我遇到过最典型的问题就是stream disconnected before completion: idle timeout waiting for sse排查下来是 Nginx 的proxy_read_timeout默认 60 秒而模型思考时间超过了这个值。解决办法是定期发送心跳注释async def event_generator(): last_heartbeat time.time() async for chunk in chain.astream(...): yield fdata: {json.dumps(chunk)}\n\n last_heartbeat time.time() # 如果长时间没数据发心跳 # 实际实现里通常用 asyncio.wait_for 配合超时SSE 协议里以:开头的行是注释前端会忽略但能保持连接活跃: heartbeat第二错误处理。流式过程中模型报错、网络中断、解析失败都要通过 SSE 事件推给前端而不是直接断连。我通常定义几种事件类型data正常数据、error错误、done结束。第三[DONE]标记。前端需要知道流什么时候结束。约定一个data: [DONE]作为结束标记是最常见的做法OpenAI 的 API 也是这么干的。5.2 前端 Vue 侧的对接Vue 里对接 SSE 有两种方式原生EventSource和fetchReadableStream。EventSource更简单但它只支持 GET 请求且不能自定义 header。如果需要 POST 或者带认证 header就得用 fetch 方案。// EventSource 方案 const es new EventSource(/analyze?reviewxxx) es.onmessage (e) { if (e.data [DONE]) { es.close() return } const chunk JSON.parse(e.data) // 注意chunk 是累积的用覆盖不是追加 Object.assign(result.value, chunk) } es.onerror (err) { console.error(SSE error, err) es.close() }// fetch ReadableStream 方案支持 POST const response await fetch(/analyze, { method: POST, headers: { Content-Type: application/json }, body: JSON.stringify({ review }) }) const reader response.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 }) // 按 \n\n 分割事件 const events buffer.split(\n\n) buffer events.pop() // 最后一段可能不完整留到下次 for (const event of events) { if (event.startsWith(data: )) { const data event.slice(6) if (data [DONE]) return Object.assign(result.value, JSON.parse(data)) } } }这里有个必须注意的细节buffer的处理。SSE 数据是按\n\n分隔的事件但网络传输可能把一个事件切成两半。所以每次读取后要保留最后一段不完整的等下次数据到了再拼接。我见过不少项目因为没处理这个导致偶发的 JSON 解析失败。5.3 封装成可复用的 SSE 调用逻辑生产项目里我会把 SSE 调用封装成一个通用的 composableexport function useSSE(url, options {}) { const data ref({}) const loading ref(false) const error ref(null) async function start(body) { loading.value true error.value null try { const response await fetch(url, { method: POST, headers: { Content-Type: application/json }, body: JSON.stringify(body) }) const reader response.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 events buffer.split(\n\n) buffer events.pop() for (const event of events) { if (!event.startsWith(data: )) continue const payload event.slice(6) if (payload [DONE]) return Object.assign(data.value, JSON.parse(payload)) } } } catch (e) { error.value e } finally { loading.value false } } return { data, loading, error, start } }这个封装把 buffer 处理、错误捕获、结束标记都包进去了业务组件里只需要const { data, start } useSSE(/analyze)然后start({ review })就行。6. 常见问题与排查技巧实录6.1 流式解析报错速查表报错信息根本原因解决方案stream disconnected before completion: idle timeout waiting for sse网关/负载均衡超时加心跳、调大 proxy_read_timeoutJSONDecodeError: Expecting valuechunk 被切断或模型输出非 JSON用 JsonOutputParser 而非 json.loadsValidationError: field required模型漏字段用 OutputFixingParser 重试前端字段重复显示把累积 chunk 当增量追加了改用 Object.assign 覆盖流式一直不结束没发 [DONE] 标记检查 event_generator 末尾中文乱码编码问题确保 ensure_asciiFalse 且前端 UTF-86.2 模型不按格式输出的排查思路这是最高频的问题。模型明明收到了格式指令还是输出一堆废话加 JSON。我的排查顺序是第一步检查格式指令有没有真的进 prompt。用parser.get_format_instructions()打印出来看看有时候 prompt 模板里忘了插{format_instructions}变量。第二步检查模型能力。小模型7B 以下对复杂 JSON Schema 的遵守率确实差。如果业务允许换大一点的模型或者简化 Schema。第三步加 few-shot 示例。在 prompt 里给一两个输入输出示例遵守率能提升一大截。这是最有效的办法。第四步用 ToolCall 替代。如果 prompt 怎么调都不行直接上with_structured_output让模型走工具调用通道。6.3 流式场景下的重试策略流式场景的重试比较麻烦因为已经推给前端的数据没法撤回。我的策略是分级处理解析失败但能修复流式结束后用 OutputFixingParser 修复把修复结果作为最后一个事件推给前端前端做一次覆盖更新。解析失败且无法修复推一个 error 事件前端展示结果格式异常请重试。模型调用失败如果还没开始推数据直接重试如果已经推了部分数据推 error 事件让用户重新发起。实操心得流式场景下不要做自动重试因为用户已经看到了部分输出自动重试会导致内容跳变体验很差。宁可让用户手动重试。6.4 性能优化的几个点第一prompt 缓存。如果格式指令很长且固定用模型的 prompt caching 功能能省不少 token 和时间。OpenAI 和部分国产模型都支持。第二并行解析。流式结束后如果要做多个字段的校验用asyncio.gather并行处理。第三前端节流渲染。chunk 来得太快会导致 Vue 频繁重渲染用requestAnimationFrame或者节流函数控制渲染频率。我一般设 50ms 一次肉眼看起来依然流畅。第四合理设置 max_tokens。结构化输出的长度是可预估的设一个合理的上限能避免模型啰嗦。7. 一个完整的实战案例串起来把前面所有东西串成一个完整案例做一个商品评论分析接口要求流式展示分析过程最终输出结构化结果。后端from fastapi import FastAPI from fastapi.responses import StreamingResponse from langchain_core.prompts import ChatPromptTemplate from langchain_core.output_parsers import JsonOutputParser from langchain_openai import ChatOpenAI from pydantic import BaseModel, Field import json app FastAPI() class ReviewAnalysis(BaseModel): summary: str Field(description一句话总结) rating: float Field(description评分0-5, ge0, le5) tags: list[str] Field(description标签列表) suggestion: str Field(description改进建议) parser JsonOutputParser(pydantic_objectReviewAnalysis) prompt ChatPromptTemplate.from_template( 分析以下商品评论按格式输出\n{format_instructions}\n\n评论{review} ).partial(format_instructionsparser.get_format_instructions()) model ChatOpenAI(modelgpt-4o-mini, streamingTrue) chain prompt | model | parser app.post(/analyze) async def analyze(payload: dict): review payload[review] async def gen(): accumulated {} try: async for chunk in chain.astream({review: review}): accumulated.update(chunk) yield fdata: {json.dumps(chunk, ensure_asciiFalse)}\n\n # 流式结束严格校验 validated ReviewAnalysis(**accumulated) yield fdata: {json.dumps({final: validated.model_dump()}, ensure_asciiFalse)}\n\n except Exception as e: yield fdata: {json.dumps({error: str(e)}, ensure_asciiFalse)}\n\n yield data: [DONE]\n\n return StreamingResponse(gen(), media_typetext/event-stream)前端用前面封装的useSSEdata里会逐步出现 summary、rating、tags、suggestion最后出现 final 字段。这个案例覆盖了JsonOutputParser 流式增量解析、Pydantic 严格校验、SSE 封装、错误处理、结束标记。基本是一个可以直接抄作业的模板。8. 几个我踩过的坑和对应经验坑一以为 JsonOutputParser 会做类型转换。模型返回rating: 4字符串JsonOutputParser 原样返回字符串不会转成数字。类型转换是 Pydantic 的活。所以流式阶段前端拿到的 rating 可能是字符串渲染时要自己处理。坑二在 chain 里同时挂 Parser 和 with_structured_output。这两个是互斥的with_structured_output返回的模型已经内置了结构化输出能力再挂 Parser 会报错。二选一。坑三SSE 的 Content-Type 写错。必须是text/event-stream写成application/json前端 EventSource 会直接报错。这个错误很低级但很常见。坑四忘了处理客户端断开。用户关掉页面时后端的流式生成还在跑浪费资源。要在event_generator里监听request.is_disconnected()断开就停止生成。坑五prompt 里的 format_instructions 没转义。JSON Schema 里有花括号如果 prompt 模板用{}做变量占位会冲突。用.partial()或者双花括号转义。最后分享一个我最近才用上的技巧用 LangSmith 追踪流式链路。流式场景的调试比普通调用难得多因为你看不到完整的中间状态。LangSmith 能把每个 chunk 的解析过程都记录下来排查增量解析问题时特别有用。免费额度对个人项目完全够用。这套方案我在两个生产项目里跑了半年多日均调用量几万次稳定性没问题。核心就是记住那个口诀要校验用 Pydantic要流式用 Json要稳定用 ToolCall三者按需组合。