尧图网站建设 
做 LLM 应用有一阵子了从最开始“调一个 prompt 返回文本”的玩具项目到现在面对真实业务里动辄几十路并发的流式交互我踩过的坑基本都集中在同一个地方模型输出怎么从“自由文本”变成“程序能直接用的结构”。今天这篇就把我最终沉淀下来的方案完整讲一遍——怎么用 LangChain 的三大 OutputParser 做结构化解析怎么把 SSE 流式接口从后端串到前端再配合 ToolCall 让智能体真正能调工具干实事。适合正在入门 LangChain、准备做 Agent 应用、或者已经在处理流式输出但被 JSON 解析折腾到头疼的朋友。这句话说出来很直白但确实是核心矛盾大模型给的是“自然语言”程序要的是“结构化数据 流式体验”。这两个诉求往往打架——你要流式就得一段段吐字可一段段吐的字怎么保证最后拼起来还是合法 JSON你让模型调工具它给的 tool_calls 参数格式不稳定怎么办这些问题不解决Agent 应用上线就是个定时炸弹。下文的所有方案我都按“能直接抄走”的标准写环境是 Python 3.10、LangChain 0.2.x、FastAPI Vue 的组合。1. 整体设计拆解结构化输出与流式为什么必须一起考虑1.1 从真实业务反推技术选型先别急着写代码。我一般拿到需求第一件事是把用户的交互链路画出来。以我这阵子做的一个“智能客服 工单自动创建”Agent 为例链路大概是这样的用户消息 → 后端调用 LLM → 流式返回打字机效果 → 最后附加一份结构化工单摘要标题、分类、紧急度、处理人→ 前端展示完文本后自动把工单信息预填到表单里。这里就有三个硬需求全链路要流式不然用户等 5 秒才看到第一个字体验直接崩最后的“工单摘要”必须是合法的 JSON不然没法直接绑定表单中间模型可能还要查一下知识库、查一下用户历史订单这就需要 ToolCall。所以你会发现SSE 流式、OutputParser、ToolCall这三个东西不是独立的技术点而是同一条链路上的三个环节。前端消费流式增量后端在流结束时做结构化收敛模型在生成过程中有需要就发起工具调用。任何一个环节断了整个体验就塌了。我见过不少项目单独看某个环节做得挺漂亮一联调就四处漏水根源就是没把这三件事当成一个整体来设计。1.2 方案选型我为什么选这些组件关于 LLM 应用的框架选型社区里天天吵。我最终稳定下来的搭配是 FastAPI LangChain LangGraph看场景决定用不用 LangGraph。选 LangChain 不是因为它“最时髦”而是因为它的抽象层帮我做完了三件烦事儿Prompt 模板与输出格式指令的拼接、流式事件的标准化astream、工具调用的参数校验与绑定bind_tools。这三件你自己写当然也行但一旦模型换了OpenAI 换到通义、换到本地 vLLM 部署的模型LangChain 的适配层能让我少改一堆代码。至于 SSE本质上就是一个 HTTP 长连接Content-Type: text/event-stream按行推data: {...}。它比 WebSocket 轻得多又是标准 HTTP 协议前端 EventSource 或者 fetch 都能消费Nginx 网关也友好。LLM 的 token 级流式输出用 SSE 是最稳的组合。我知道有些团队直接用 WebSocket但说实话对“单向下行推送”这种场景SSE 是更简单的选择少了一堆连接管理和心跳自研的活儿。还有一个小细节要提前定所有流式事件我都用 JSON 封装成统一的{type, content, ...}结构而不是直接裸推文本。这看起来多了一层序列化实际上为你后面的“多事件类型”留了后路——你可以在一个流里同时推 token、推工具调用状态、推结束标记、推错误前端只维护一个分发函数。后面讲 Vue 消费逻辑时你会看到这个设计能让前端的代码量少一半。2. 三大 OutputParser 实战把模型输出变成可靠数据2.1 JsonOutputParser最快上手、也最容易失控JsonOutputParser 是 LangChain 里最基础的 JSON 解析器用法很简单from langchain_core.output_parsers import JsonOutputParser from langchain_core.prompts import PromptTemplate parser JsonOutputParser() prompt PromptTemplate( template分析用户咨询内容并输出 JSON。\n{format_instructions}\n用户输入{input}, input_variables[input], partial_variables{format_instructions: parser.get_format_instructions()}, ) chain prompt | llm | parser result chain.invoke({input: 我的订单三天没发货了你们怎么回事}) # result 就是一个 dict比如 {sentiment: anger, category: shipping, priority: 2}它的get_format_instructions()会自动往 Prompt 里塞一段“请严格输出 JSON 对象”的指令。底层直接把模型输出的文本做extract_json抽取然后用json.loads解析成 dict。为什么说它容易失控因为模型在流式输出过程中你拿到的每一段 chunk 都可能是不完整的 JSON。JsonOutputParser 内部有parse_partial_json的容错逻辑但它只保证“当前这一步尽量可解析”不保证最终 schema 一定是你想要的字段。换句话说它保证的是“JSON 语法合法”不保证“你的业务字段齐全”。如果你只要一个{ reply: ... }这种极简结构用它够了但如果你要给前端表单绑定 5 个字段远远不够。另外我要强烈建议让 JsonOutputParser 配合自定义校验函数使用。你可以在 chain 后面再加一步def validate(result: dict) - dict: required {sentiment, category, priority} if not required.issubset(result.keys()): raise ValueError(f缺少必要字段: {required - result.keys()}) return result chain prompt | llm | parser | validate这种做法叫“防御式解析”——不要相信模型 100% 按你的指令来在边界处加一道闸门宁可让这轮报错重试也不要让脏数据进到下游业务里。这个习惯救过我很多次尤其是换模型或者调 prompt 之后输出格式经常悄无声息地变形。2.2 PydanticOutputParser类型安全的硬核选择如果你需要严格的数据模型PydanticOutputParser 是更可靠的选择。它直接用 Pydantic 定义 schema然后让模型按这个 schema 输出解析结果直接是一个 Pydantic 模型实例天然带类型校验。from typing import Literal from langchain_core.pydantic_v1 import BaseModel, Field from langchain_core.output_parsers import PydanticOutputParser class TicketInfo(BaseModel): title: str Field(description工单标题一句话概括问题) category: Literal[shipping, refund, quality, other] Field(description问题分类) urgency: int Field(description紧急度 1-5数字越大越紧急, ge1, le5) customer_id: str Field(description客户编号如 CUS-2024-0001) action: str Field(description建议的处理动作) parser PydanticOutputParser(pydantic_objectTicketInfo) prompt PromptTemplate( template根据客服对话提取工单信息。\n{format_instructions}\n对话内容{conversation}, input_variables[conversation], partial_variables{format_instructions: parser.get_format_instructions()}, )这个方案的最强之处在于它把“输出契约”从自然语言指令上升到了程序类型系统。模型输出的 JSON 一旦出现类型错误比如 urgency 传了字符串 high解析阶段直接抛校验异常。你可以在解析出错时用OutputFixingParser自动补救——它会让模型基于原始输出和错误信息再生成一份修正过的输出from langchain_core.output_parsers import OutputFixingParser fixing_parser OutputFixingParser.from_llm(parserparser, llmllm)这一个环节至少救了我十几次生产事故。之前有个场景是模型总把枚举值输出成中文“退货”而不是 “refund”Pydantic 直接校验失败OutputFixingParser 拿到报错信息后让模型重新对照格式说明输出成功率能拉回 95% 以上。当然修复意味着多一次模型调用延迟会加几百毫秒所以我的策略是修复后返回日志里标记was_fixedTrue前端不用管后端的监控体系能看到。这里有个兼容性细节LangChain 0.2/0.3 的兼容层里老接口建议从langchain_core.pydantic_v1导入如果你确定全链路都是 pydantic v2 的新项目直接从pydantic导入也完全没问题。示例里统一用langchain_core.pydantic_v1兼容性最稳不容易跟 LangChain 内部组件打架。2.3 StructuredOutputParserSchema 驱动的轻量方案StructuredOutputParser 是介于前面两者之间的方案。不像 Pydantic 那样需要定义完整的模型类它用ResponseSchema列表来描述字段适合字段不多、类型要求不算苛刻、但需要保持灵活的场景。from langchain_core.output_parsers import StructuredOutputParser, ResponseSchema schemas [ ResponseSchema(nameanswer, description对用户问题的直接回答), ResponseSchema(namesources, description信息来源列表字符串数组), ResponseSchema(nameconfidence, description置信度 high/medium/low), ] parser StructuredOutputParser.from_response_schemas(schemas) prompt PromptTemplate( template回答问题并补充来源。\n{format_instructions}\n问题{question}, input_variables[question], partial_variables{format_instructions: parser.get_format_instructions()}, )模型生成的内容会被解析成{answer: ..., sources: [...], confidence: high}这样的 dict。它和 JsonOutputParser 的区别在于format_instructions 是按你的 ResponseSchema 生成的模型的输出会更有“章法”但它的校验仍然很弱只做基本类型转换不会像 Pydantic 那样严格。适合的场景是你不是要把数据直接灌进有强类型的业务系统而是要给前端展示 后续自己再加工。顺带提一句如果输出是单纯的逗号分隔列表LangChain 还有CommaSeparatedListOutputParser一行parser CommaSeparatedListOutputParser()就够比自己正则拆字符串稳得多。这块属于小工具遇到再试就行。2.4 三大解析器的选型对照我把三者的差异整理成一张表方便你按场景直接抄维度JsonOutputParserPydanticOutputParserStructuredOutputParser类型校验只验 JSON 语法强类型 枚举 条件约束基础字段存在性输出契约无固定 schemaPydantic 模型 契约ResponseSchema 列表出错补救需自行处理OutputFixingParser 可自动修复需自行处理上手成本最低中要写模型类低典型场景快速返回一个简单 dict需要可靠入库/绑定表单中等复杂度的展示型输出说实话我现在的默认选项基本是 PydanticOutputParser只在“返回内容非常自由”的场景才会用 JsonOutputParser。另外要提醒LangChain 0.2 之后推出了with_structured_output()这个方法它把模型厂商的 Function Calling / Structured Output 能力封装成了统一的接口很多情况下比 Prompt Parser 的组合更稳。这里先埋个伏笔第四节讲 ToolCall 时会展开因为with_structured_output的结构化输出和 tool calling 本质上是同一套底层机制。3. SSE 流式接入后端推送与前端消费的完整链路3.1 SSE 协议要点与 LangChain 流式接口对接SSE 的协议本身不复杂服务端响应头指定Content-Type: text/event-stream然后按data: 内容\n\n的格式持续写数据。浏览器端要么用EventSource自动接收要么用fetch配合ReadableStream手动解析。真正容易踩坑的往往不在协议本身而在网络基础设施。最典型的Nginx 默认会缓冲响应导致前端半天收不到数据或者网关的proxy_read_timeout太短模型思考超过 60 秒连接就被掐了。这部分我放到第五节专门讲排查这里先把 LangChain 对接方法说清楚。LangChain 的astream是对模型流式能力的抽象封装。不同模型厂商的流式事件结构不一样OpenAI 是 chunkAnthropic 是 content_block_delta通义是 output_text但 LangChain 统一成message.content的增量文本。后端封装一个异步生成器就是把增量转成 SSE 帧import json from fastapi import FastAPI, Request from fastapi.responses import StreamingResponse from langchain_openai import ChatOpenAI app FastAPI() llm ChatOpenAI(modelgpt-4o-mini, temperature0.3) app.post(/v1/chat/stream) async def chat_stream(request: Request): body await request.json() message body.get(message, ) async def event_gen(): # 先发一个开始事件前端可以借此清理输入框状态 yield fdata: {json.dumps({type: start}, ensure_asciiFalse)}\n\n # 逐 token 推送 async for chunk in llm.astream(message): content chunk.content if content: yield fdata: {json.dumps({type: token, content: content}, ensure_asciiFalse)}\n\n # 结束事件 yield fdata: {json.dumps({type: done}, ensure_asciiFalse)}\n\n return StreamingResponse( event_gen(), media_typetext/event-stream, headers{ Cache-Control: no-cache, Connection: keep-alive, X-Accel-Buffering: no, # 关掉 Nginx 缓冲 }, )注意ensure_asciiFalse这个参数我最初几次就是没加推给前端的中文全变成\u5f00\u59cb这样的转义序列前端还得额外 unescape平白多一堆 bug。3.2 心跳与多事件类型的设计生产环境里SSE 连接可能因为各种中间设备负载均衡、代理的空闲超时而断开即使你的模型还在思考。业界标准做法是服务端定期发送注释行以:开头的行作为心跳SSE 规范里这种注释帧会被客户端忽略但能“顶住”连接不被判定为 idle。用队列模式实现最清爽token 产出一个放一个进队列主循环用wait_for等队列等不到就发心跳import asyncio, json async def event_gen(): # token_queue 由另一个协程用 llm.astream 填充 while True: try: chunk await asyncio.wait_for(token_queue.get(), timeout15) yield fdata: {json.dumps({type: token, content: chunk}, ensure_asciiFalse)}\n\n except asyncio.TimeoutError: yield : heartbeat\n\n另外SSE 的event:字段和id:字段也值得利用。event: tool_call可以和data: {json}组合让前端通过addEventListener(tool_call, ...)单独监听。不过我在 Vue 项目里更习惯全用data: 内部type字段因为 fetch 手写解析时处理起来更统一少一套分支逻辑。3.3 Python 与 Vue 前端的流式消费后端把流推出来了前端怎么接这里有个最常见的坑EventSource 只支持 GET 请求。你如果想让后端通过 POST 接收消息体带长 prompt、带会话 ID浏览器原生 EventSource 就无能为力了必须用 fetch ReadableStream 手写消费逻辑。我封装了一个 Vue 侧的流式消费工具核心代码长这样async function consumeSSE(url: string, body: object, handlers: { onToken: (text: string) void onToolCall?: (data: any) void onDone?: () void onError?: (err: Error) void }) { const response await fetch(url, { method: POST, headers: { Content-Type: application/json }, body: JSON.stringify(body), }) if (!response.ok || !response.body) { throw new Error(HTTP ${response.status}) } const reader response.body.getReader() const decoder new TextDecoder(utf-8) let buffer while (true) { const { done, value } await reader.read() if (done) break buffer decoder.decode(value, { stream: true }) // SSE 帧以空行分隔每次从缓冲里切出完整帧 const frames buffer.split(\n\n) buffer frames.pop() ?? for (const frame of frames) { const line frame.trim() if (!line.startsWith(data:)) continue const dataStr line.slice(5).trim() // 心跳注释行被过滤data 为空则跳过 if (!dataStr) continue try { const payload JSON.parse(dataStr) switch (payload.type) { case token: handlers.onToken(payload.content) break case tool_call: handlers.onToolCall?.(payload) break case done: handlers.onDone?.() break } } catch (e) { // 半截帧解析失败的场景跳过等待下一次 read console.warn(parse frame failed, dataStr) } } } }这里两个细节我觉得比代码本身更重要。第一个是缓冲拆分。SSE 帧之间由空行\n\n分隔网络读取的 chunk 边界不会恰好落在帧边界上所以必须维护一个buffer把不完整的尾部留到下次。这个逻辑写错就会出现“最后一段数据永远丢”或“JSON 被截断”的诡异 bug。第二个是解码器参数。decoder.decode(value, { stream: true })里的{ stream: true }必须带上否则多字节 UTF-8 字符比如中文“你”被 TCP 分包拆成两半时字节流会在第一半就尝试解码结果出来个乱码。这个参数告诉解码器“后面还有数据攒一攒再解码”。我光这个问题就被同事拉着排查过一下午。4. ToolCall 方案让模型真正“下地干活”4.1 bind_tools 与函数调用链路前面讲 OutputParser 解决的是“文本 → 数据”ToolCall 解决的是“意图 → 动作”。在 AI Agent 场景里模型不能只回嘴皮子它得会查订单、写工单、调 API。LangChain 给我们提供了一套统一的函数调用抽象先定义工具再绑定到模型上模型在推理时如果要调用工具就会在输出里带出结构化的tool_calls。from langchain_core.tools import tool from langchain_core.pydantic_v1 import BaseModel, Field from langchain_openai import ChatOpenAI class OrderQueryInput(BaseModel): order_id: str Field(description订单号形如 ORD-2024-0815) need_detail: bool Field(defaultFalse, description是否需要物流明细) tool(args_schemaOrderQueryInput) def query_order(order_id: str, need_detail: bool False) - str: 根据订单号查询订单状态和物流信息当用户询问订单进度、物流状态时使用。 # 这里替换成你的真实业务查询逻辑 return f订单 {order_id} 状态: 已发货; 物流: 顺丰 SF123456 llm ChatOpenAI(modelgpt-4o, temperature0) llm_with_tools llm.bind_tools([query_order]) result llm_with_tools.invoke(我的订单 ORD-2024-0815 什么时候到) print(result.tool_calls) # [{name: query_order, args: {order_id: ORD-2024-0815, need_detail: True}, id: call_abc123, type: tool_call}]有几个要点我要特别强调。第一tool装饰器下面函数的 docstring 就是给模型看的“工具说明书”非常重要。模型不会读你代码注释它只读 docstring Pydantic 字段描述来决定什么时候调用、传什么参数。实践里我见过太多人 docstring 写一句“查询订单”就完事结果模型把用户消息里所有字段乱塞进参数。我的经验是 docstring 里至少写清楚工具什么时候用、什么时候不要用、每个参数的含义、返回什么格式。第二args_schema比直接在函数签名里写类型注解更稳。Pydantic 模型可以加Field(description...)这个 description 会映射到模型 API 的parametersschema 里直接影响模型生成参数的质量。不要图省事儿只写order_id: str。第三temperature记得调低。工具调用的参数生成是“确定性任务”temperature0 能明显减少参数幻觉。这个设置单独为工具绑定场景调不影响你普通对话链路的参数。4.2 ToolCall 结果的解析与路由拿到result.tool_calls之后它只是一个声明——模型“想”调用这个工具但工具到底执行没有、执行结果是什么需要你自己完成“执行 → 回填 → 再交给模型”的循环。这个循环如果用 LangGraph可以直接用ToolNode自动包一层from langgraph.prebuilt import ToolNode, tools_condition from langgraph.graph import StateGraph, START, END from langgraph.graph.message import add_messages from typing import Annotated, TypedDict class AgentState(TypedDict): messages: Annotated[list, add_messages] def build_graph(): graph StateGraph(AgentState) graph.add_node(llm, llm_with_tools) graph.add_node(tools, ToolNode([query_order])) graph.add_edge(START, llm) graph.add_conditional_edges(llm, tools_condition, {tools: tools, END: END}) graph.add_edge(tools, llm) return graph.compile()tools_condition会自动检查上一步的 AIMessage 里有没有tool_calls有就进工具节点执行完把结果以 ToolMessage 形式追加回消息列表再让模型根据结果生成最终回复。这就是所谓的“Agent 循环”。如果你不想引入 LangGraph手写循环也不复杂核心逻辑messages [{role: user, content: 我的订单什么时候到}] for _ in range(5): # 最多轮询 5 轮防止死循环 response llm_with_tools.invoke(messages) messages.append(response) if not response.tool_calls: break # 模型没有工具调用欲望输出最终回复 for tc in response.tool_calls: tool_result execute_tool(tc[name], tc[args]) messages.append({ role: tool, tool_call_id: tc[id], content: str(tool_result), })这里一个隐藏 bug 是openai 兼容接口要求 tool message 必须携带tool_call_id这个 id 必须和你收到的 tool_calls 里那个 id 一一对应。很多人自己手写循环时忘了带导致第二次请求直接 400。用 LangGraph 的 ToolNode 就没这个问题所以我后来基本都直接上 LangGraph。另外如果 Agent 流程需要人工审批human-in-the-loopLangGraph 的interrupt机制配合 Agent Inbox 模式可以在 ToolNode 执行前挂起把待确认的 tool_call 发给用户确认后再继续。这类场景在工单创建、付款确认类 Agent 里特别常见值得单独玩玩。4.3 with_structured_output结构化输出与工具调用的统一入口前面我在解析器那节埋了个伏笔现在展开。LangChain 0.2 里有一个非常实用的方法with_structured_output()。它底层做的事情其实是“把你的 Pydantic 模型变成工具定义强制模型走 tool calling / function calling 通道来输出”。class TicketInfo(BaseModel): title: str Field(description工单标题) category: Literal[shipping, refund, quality, other] Field(description分类) urgency: int Field(description1-5, ge1, le5) action: str Field(description处理动作) structured_llm llm.with_structured_output(TicketInfo) result structured_llm.invoke(用户说订单三天没到要求马上退款) # 直接得到 TicketInfo 实例类型已校验这比 “Prompt 拼接 format_instructions 解析器” 稳得多因为模型厂商对 function calling 有专门的微调结构化输出的可靠性普遍比“从自由文本里解析 JSON”高一个量级。我的经验里同一个模型用with_structured_output做分类任务的字段正确率比 PydanticOutputParser 高约 5~8 个百分点虽然不算巨大但在生产环境已经很值了。那什么时候用 OutputParser、什么时候用 with_structured_output我的判断标准是用的是 OpenAI / Anthropic / 通义等有原生 function calling 支持的模型就用with_structured_output用的是本地 vLLM 部署的模型且没有开 function calling或者输出格式要求特别“软”比如只要一个 JSON、schema 每天都在变就用 OutputParser两者也可以组合ToolCall 本身就用bind_tools最终回答需要结构化给前端时再用with_structured_output生成一个附带的结构化摘要。5. 常见问题与排查实录5.1 stream disconnected before completion: idle timeout这个报错我在生产环境至少见过三种形态Nginx 层的proxy_read_timeout、云厂商负载均衡的连接空闲限制、以及浏览器对 fetch 流的内部超时。表象都是一个流没推完连接被服务端或中间层掐断。排查路径我建议按这个顺序走先确认是不是你的后端程序主动崩了日志里有没有异常再确认是不是 Nginx/网关层超时看 access log 的 upstream_status出现 499 或 502 基本就是这层问题最后确认是不是前端断开的浏览器 Network 面板看响应是否中断。解决方案分三层后端在生成器里加心跳注释帧: keep-alive\n\n15 秒一条能顶住绝大多数 idle 超时Nginx显式配置proxy_read_timeout 300s; proxy_buffering off;同时加上X-Accel-Buffering: no头前端对 fetch 流做异常重连捕获 TypeError网络中断会抛这个后根据断点位置决定是重发整请求还是提示用户重试。其中续传这个事目前没有完美方案我一般策略是断路器 用户手动重试并明确记录已经拼接的文本。流式应用对于断线要有一个共识宁可让用户点一下重试也不能自动重发导致重复内容。5.2 流式 JSON 解析失败与半截帧做流式聊天时最大痛点你在流式输出过程中想提前把 JSON 里的title字段抽出来做实时展示但模型还只输出到一半JSON 就是个残缺品。这里我的方案是“分两路推进”第一路展示文本流直接用 token 拼接不解析 第二路用一个partial_json_parser持续尝试把已收到的文本解析成“尽可能完整的 JSON”把能提取的字段实时更新到前端状态里。一旦最终流结束再走一次完整 Pydantic 校验覆盖掉中间值。LangChain 的 JsonOutputParser 本质上就是在做这件事它内部有parse_partial_json容错逻辑所以如果你用astream时每一帧都对 parse 前的文本调用parser.parse_partial_json(text)就能拿到“当前最接近完整”的 dict。注意要 catch 异常因为极端情况下解析器也救不回来。另外一个特别 dirty 的场景模型输出的 JSON 里带了多余的前缀或后缀比如“好的以下是结果{...}”。好在 LangChain 的extract_json会做智能抽取能容忍这种情况。但如果你用原生json.loads去解析八成会踩坑。所以我建议解析一律走 LangChain 的 parser不要自己图省事直接 load。5.3 SSE 中文字符与编码问题这个坑来得毫无防备。FastAPI 的 StreamingResponse 默认会帮你做 UTF-8 编码但如果你用sse-starlette之类的库或者自己拼data:字符串时忘了ensure_asciiFalse中文就会以\uXXXX形式推给前端。前端拿到合法 JSON 却显示成“\u5f00\u59cb”字符串而不是“开始”。解决办法就一条服务端生成 SSE 帧时统一json.dumps(data, ensure_asciiFalse)前端解码统一TextDecoder(utf-8)JSON.parse两边都保持“JSON 内不转义、传输层不二次转义”。还有一个隐藏点如果服务端用yield fdata: {json_str}\n\n而json_str本身含换行会破坏 SSE 帧格式。因为 SSE 帧的 data 不能包含裸换行解决办法是json_str.replace(\n, \\n)。这个我见过有人踩坑前端疯狂解析错误。5.4 ToolCall 参数幻觉与兜底最后聊聊 ToolCall 最常见的问题参数幻觉。模型“觉得”用户想查订单就编了一个order_id: ORD-2024-0815但用户根本没提供订单号。这类问题在小模型上尤其严重。我的兜底方案Pydantic 模型里给参数加min_length之类的约束或者用自定义校验检查订单号前缀工具执行前做二次校验参数不合法直接返回工具错误信息让模型知道“参数错了请向用户要正确的订单号”而不是崩溃工具执行结果统一返回结构化文本不要返回 Python 对象避免模型在下一轮推理时被非文本对象搞晕。第二个方案我特别推荐把“工具参数错误”当成一次正常的工具结果返回给模型模型会很自然地生成“抱歉我需要确认您的订单号”这类回复。这比让程序抛异常、把整个 Agent 循环打挂体验好 10 倍。最后再分享一个我的实战习惯。上面这套链路我每次接到新项目都不会急着写业务代码而是先把“一条消息从进来到出结果”的完整链路用最小 demo 跑通然后把五个关键节点打上日志入参、LLM 原始输出、解析器结果、ToolCall 参数、最终结构化输出。这五个节点的日志打全了后面上线排查基本不用动脑子直接看日志点就能定位是哪一环出了幺蛾子。工期紧的时候这个习惯帮我把联调时间压缩了至少一半你也试试。 LangChain三大OutputParser与SSE流式接入实战指南
做 LLM 应用有一阵子了从最开始“调一个 prompt 返回文本”的玩具项目到现在面对真实业务里动辄几十路并发的流式交互我踩过的坑基本都集中在同一个地方模型输出怎么从“自由文本”变成“程序能直接用的结构”。今天这篇就把我最终沉淀下来的方案完整讲一遍——怎么用 LangChain 的三大 OutputParser 做结构化解析怎么把 SSE 流式接口从后端串到前端再配合 ToolCall 让智能体真正能调工具干实事。适合正在入门 LangChain、准备做 Agent 应用、或者已经在处理流式输出但被 JSON 解析折腾到头疼的朋友。这句话说出来很直白但确实是核心矛盾大模型给的是“自然语言”程序要的是“结构化数据 流式体验”。这两个诉求往往打架——你要流式就得一段段吐字可一段段吐的字怎么保证最后拼起来还是合法 JSON你让模型调工具它给的 tool_calls 参数格式不稳定怎么办这些问题不解决Agent 应用上线就是个定时炸弹。下文的所有方案我都按“能直接抄走”的标准写环境是 Python 3.10、LangChain 0.2.x、FastAPI Vue 的组合。1. 整体设计拆解结构化输出与流式为什么必须一起考虑1.1 从真实业务反推技术选型先别急着写代码。我一般拿到需求第一件事是把用户的交互链路画出来。以我这阵子做的一个“智能客服 工单自动创建”Agent 为例链路大概是这样的用户消息 → 后端调用 LLM → 流式返回打字机效果 → 最后附加一份结构化工单摘要标题、分类、紧急度、处理人→ 前端展示完文本后自动把工单信息预填到表单里。这里就有三个硬需求全链路要流式不然用户等 5 秒才看到第一个字体验直接崩最后的“工单摘要”必须是合法的 JSON不然没法直接绑定表单中间模型可能还要查一下知识库、查一下用户历史订单这就需要 ToolCall。所以你会发现SSE 流式、OutputParser、ToolCall这三个东西不是独立的技术点而是同一条链路上的三个环节。前端消费流式增量后端在流结束时做结构化收敛模型在生成过程中有需要就发起工具调用。任何一个环节断了整个体验就塌了。我见过不少项目单独看某个环节做得挺漂亮一联调就四处漏水根源就是没把这三件事当成一个整体来设计。1.2 方案选型我为什么选这些组件关于 LLM 应用的框架选型社区里天天吵。我最终稳定下来的搭配是 FastAPI LangChain LangGraph看场景决定用不用 LangGraph。选 LangChain 不是因为它“最时髦”而是因为它的抽象层帮我做完了三件烦事儿Prompt 模板与输出格式指令的拼接、流式事件的标准化astream、工具调用的参数校验与绑定bind_tools。这三件你自己写当然也行但一旦模型换了OpenAI 换到通义、换到本地 vLLM 部署的模型LangChain 的适配层能让我少改一堆代码。至于 SSE本质上就是一个 HTTP 长连接Content-Type: text/event-stream按行推data: {...}。它比 WebSocket 轻得多又是标准 HTTP 协议前端 EventSource 或者 fetch 都能消费Nginx 网关也友好。LLM 的 token 级流式输出用 SSE 是最稳的组合。我知道有些团队直接用 WebSocket但说实话对“单向下行推送”这种场景SSE 是更简单的选择少了一堆连接管理和心跳自研的活儿。还有一个小细节要提前定所有流式事件我都用 JSON 封装成统一的{type, content, ...}结构而不是直接裸推文本。这看起来多了一层序列化实际上为你后面的“多事件类型”留了后路——你可以在一个流里同时推 token、推工具调用状态、推结束标记、推错误前端只维护一个分发函数。后面讲 Vue 消费逻辑时你会看到这个设计能让前端的代码量少一半。2. 三大 OutputParser 实战把模型输出变成可靠数据2.1 JsonOutputParser最快上手、也最容易失控JsonOutputParser 是 LangChain 里最基础的 JSON 解析器用法很简单from langchain_core.output_parsers import JsonOutputParser from langchain_core.prompts import PromptTemplate parser JsonOutputParser() prompt PromptTemplate( template分析用户咨询内容并输出 JSON。\n{format_instructions}\n用户输入{input}, input_variables[input], partial_variables{format_instructions: parser.get_format_instructions()}, ) chain prompt | llm | parser result chain.invoke({input: 我的订单三天没发货了你们怎么回事}) # result 就是一个 dict比如 {sentiment: anger, category: shipping, priority: 2}它的get_format_instructions()会自动往 Prompt 里塞一段“请严格输出 JSON 对象”的指令。底层直接把模型输出的文本做extract_json抽取然后用json.loads解析成 dict。为什么说它容易失控因为模型在流式输出过程中你拿到的每一段 chunk 都可能是不完整的 JSON。JsonOutputParser 内部有parse_partial_json的容错逻辑但它只保证“当前这一步尽量可解析”不保证最终 schema 一定是你想要的字段。换句话说它保证的是“JSON 语法合法”不保证“你的业务字段齐全”。如果你只要一个{ reply: ... }这种极简结构用它够了但如果你要给前端表单绑定 5 个字段远远不够。另外我要强烈建议让 JsonOutputParser 配合自定义校验函数使用。你可以在 chain 后面再加一步def validate(result: dict) - dict: required {sentiment, category, priority} if not required.issubset(result.keys()): raise ValueError(f缺少必要字段: {required - result.keys()}) return result chain prompt | llm | parser | validate这种做法叫“防御式解析”——不要相信模型 100% 按你的指令来在边界处加一道闸门宁可让这轮报错重试也不要让脏数据进到下游业务里。这个习惯救过我很多次尤其是换模型或者调 prompt 之后输出格式经常悄无声息地变形。2.2 PydanticOutputParser类型安全的硬核选择如果你需要严格的数据模型PydanticOutputParser 是更可靠的选择。它直接用 Pydantic 定义 schema然后让模型按这个 schema 输出解析结果直接是一个 Pydantic 模型实例天然带类型校验。from typing import Literal from langchain_core.pydantic_v1 import BaseModel, Field from langchain_core.output_parsers import PydanticOutputParser class TicketInfo(BaseModel): title: str Field(description工单标题一句话概括问题) category: Literal[shipping, refund, quality, other] Field(description问题分类) urgency: int Field(description紧急度 1-5数字越大越紧急, ge1, le5) customer_id: str Field(description客户编号如 CUS-2024-0001) action: str Field(description建议的处理动作) parser PydanticOutputParser(pydantic_objectTicketInfo) prompt PromptTemplate( template根据客服对话提取工单信息。\n{format_instructions}\n对话内容{conversation}, input_variables[conversation], partial_variables{format_instructions: parser.get_format_instructions()}, )这个方案的最强之处在于它把“输出契约”从自然语言指令上升到了程序类型系统。模型输出的 JSON 一旦出现类型错误比如 urgency 传了字符串 high解析阶段直接抛校验异常。你可以在解析出错时用OutputFixingParser自动补救——它会让模型基于原始输出和错误信息再生成一份修正过的输出from langchain_core.output_parsers import OutputFixingParser fixing_parser OutputFixingParser.from_llm(parserparser, llmllm)这一个环节至少救了我十几次生产事故。之前有个场景是模型总把枚举值输出成中文“退货”而不是 “refund”Pydantic 直接校验失败OutputFixingParser 拿到报错信息后让模型重新对照格式说明输出成功率能拉回 95% 以上。当然修复意味着多一次模型调用延迟会加几百毫秒所以我的策略是修复后返回日志里标记was_fixedTrue前端不用管后端的监控体系能看到。这里有个兼容性细节LangChain 0.2/0.3 的兼容层里老接口建议从langchain_core.pydantic_v1导入如果你确定全链路都是 pydantic v2 的新项目直接从pydantic导入也完全没问题。示例里统一用langchain_core.pydantic_v1兼容性最稳不容易跟 LangChain 内部组件打架。2.3 StructuredOutputParserSchema 驱动的轻量方案StructuredOutputParser 是介于前面两者之间的方案。不像 Pydantic 那样需要定义完整的模型类它用ResponseSchema列表来描述字段适合字段不多、类型要求不算苛刻、但需要保持灵活的场景。from langchain_core.output_parsers import StructuredOutputParser, ResponseSchema schemas [ ResponseSchema(nameanswer, description对用户问题的直接回答), ResponseSchema(namesources, description信息来源列表字符串数组), ResponseSchema(nameconfidence, description置信度 high/medium/low), ] parser StructuredOutputParser.from_response_schemas(schemas) prompt PromptTemplate( template回答问题并补充来源。\n{format_instructions}\n问题{question}, input_variables[question], partial_variables{format_instructions: parser.get_format_instructions()}, )模型生成的内容会被解析成{answer: ..., sources: [...], confidence: high}这样的 dict。它和 JsonOutputParser 的区别在于format_instructions 是按你的 ResponseSchema 生成的模型的输出会更有“章法”但它的校验仍然很弱只做基本类型转换不会像 Pydantic 那样严格。适合的场景是你不是要把数据直接灌进有强类型的业务系统而是要给前端展示 后续自己再加工。顺带提一句如果输出是单纯的逗号分隔列表LangChain 还有CommaSeparatedListOutputParser一行parser CommaSeparatedListOutputParser()就够比自己正则拆字符串稳得多。这块属于小工具遇到再试就行。2.4 三大解析器的选型对照我把三者的差异整理成一张表方便你按场景直接抄维度JsonOutputParserPydanticOutputParserStructuredOutputParser类型校验只验 JSON 语法强类型 枚举 条件约束基础字段存在性输出契约无固定 schemaPydantic 模型 契约ResponseSchema 列表出错补救需自行处理OutputFixingParser 可自动修复需自行处理上手成本最低中要写模型类低典型场景快速返回一个简单 dict需要可靠入库/绑定表单中等复杂度的展示型输出说实话我现在的默认选项基本是 PydanticOutputParser只在“返回内容非常自由”的场景才会用 JsonOutputParser。另外要提醒LangChain 0.2 之后推出了with_structured_output()这个方法它把模型厂商的 Function Calling / Structured Output 能力封装成了统一的接口很多情况下比 Prompt Parser 的组合更稳。这里先埋个伏笔第四节讲 ToolCall 时会展开因为with_structured_output的结构化输出和 tool calling 本质上是同一套底层机制。3. SSE 流式接入后端推送与前端消费的完整链路3.1 SSE 协议要点与 LangChain 流式接口对接SSE 的协议本身不复杂服务端响应头指定Content-Type: text/event-stream然后按data: 内容\n\n的格式持续写数据。浏览器端要么用EventSource自动接收要么用fetch配合ReadableStream手动解析。真正容易踩坑的往往不在协议本身而在网络基础设施。最典型的Nginx 默认会缓冲响应导致前端半天收不到数据或者网关的proxy_read_timeout太短模型思考超过 60 秒连接就被掐了。这部分我放到第五节专门讲排查这里先把 LangChain 对接方法说清楚。LangChain 的astream是对模型流式能力的抽象封装。不同模型厂商的流式事件结构不一样OpenAI 是 chunkAnthropic 是 content_block_delta通义是 output_text但 LangChain 统一成message.content的增量文本。后端封装一个异步生成器就是把增量转成 SSE 帧import json from fastapi import FastAPI, Request from fastapi.responses import StreamingResponse from langchain_openai import ChatOpenAI app FastAPI() llm ChatOpenAI(modelgpt-4o-mini, temperature0.3) app.post(/v1/chat/stream) async def chat_stream(request: Request): body await request.json() message body.get(message, ) async def event_gen(): # 先发一个开始事件前端可以借此清理输入框状态 yield fdata: {json.dumps({type: start}, ensure_asciiFalse)}\n\n # 逐 token 推送 async for chunk in llm.astream(message): content chunk.content if content: yield fdata: {json.dumps({type: token, content: content}, ensure_asciiFalse)}\n\n # 结束事件 yield fdata: {json.dumps({type: done}, ensure_asciiFalse)}\n\n return StreamingResponse( event_gen(), media_typetext/event-stream, headers{ Cache-Control: no-cache, Connection: keep-alive, X-Accel-Buffering: no, # 关掉 Nginx 缓冲 }, )注意ensure_asciiFalse这个参数我最初几次就是没加推给前端的中文全变成\u5f00\u59cb这样的转义序列前端还得额外 unescape平白多一堆 bug。3.2 心跳与多事件类型的设计生产环境里SSE 连接可能因为各种中间设备负载均衡、代理的空闲超时而断开即使你的模型还在思考。业界标准做法是服务端定期发送注释行以:开头的行作为心跳SSE 规范里这种注释帧会被客户端忽略但能“顶住”连接不被判定为 idle。用队列模式实现最清爽token 产出一个放一个进队列主循环用wait_for等队列等不到就发心跳import asyncio, json async def event_gen(): # token_queue 由另一个协程用 llm.astream 填充 while True: try: chunk await asyncio.wait_for(token_queue.get(), timeout15) yield fdata: {json.dumps({type: token, content: chunk}, ensure_asciiFalse)}\n\n except asyncio.TimeoutError: yield : heartbeat\n\n另外SSE 的event:字段和id:字段也值得利用。event: tool_call可以和data: {json}组合让前端通过addEventListener(tool_call, ...)单独监听。不过我在 Vue 项目里更习惯全用data: 内部type字段因为 fetch 手写解析时处理起来更统一少一套分支逻辑。3.3 Python 与 Vue 前端的流式消费后端把流推出来了前端怎么接这里有个最常见的坑EventSource 只支持 GET 请求。你如果想让后端通过 POST 接收消息体带长 prompt、带会话 ID浏览器原生 EventSource 就无能为力了必须用 fetch ReadableStream 手写消费逻辑。我封装了一个 Vue 侧的流式消费工具核心代码长这样async function consumeSSE(url: string, body: object, handlers: { onToken: (text: string) void onToolCall?: (data: any) void onDone?: () void onError?: (err: Error) void }) { const response await fetch(url, { method: POST, headers: { Content-Type: application/json }, body: JSON.stringify(body), }) if (!response.ok || !response.body) { throw new Error(HTTP ${response.status}) } const reader response.body.getReader() const decoder new TextDecoder(utf-8) let buffer while (true) { const { done, value } await reader.read() if (done) break buffer decoder.decode(value, { stream: true }) // SSE 帧以空行分隔每次从缓冲里切出完整帧 const frames buffer.split(\n\n) buffer frames.pop() ?? for (const frame of frames) { const line frame.trim() if (!line.startsWith(data:)) continue const dataStr line.slice(5).trim() // 心跳注释行被过滤data 为空则跳过 if (!dataStr) continue try { const payload JSON.parse(dataStr) switch (payload.type) { case token: handlers.onToken(payload.content) break case tool_call: handlers.onToolCall?.(payload) break case done: handlers.onDone?.() break } } catch (e) { // 半截帧解析失败的场景跳过等待下一次 read console.warn(parse frame failed, dataStr) } } } }这里两个细节我觉得比代码本身更重要。第一个是缓冲拆分。SSE 帧之间由空行\n\n分隔网络读取的 chunk 边界不会恰好落在帧边界上所以必须维护一个buffer把不完整的尾部留到下次。这个逻辑写错就会出现“最后一段数据永远丢”或“JSON 被截断”的诡异 bug。第二个是解码器参数。decoder.decode(value, { stream: true })里的{ stream: true }必须带上否则多字节 UTF-8 字符比如中文“你”被 TCP 分包拆成两半时字节流会在第一半就尝试解码结果出来个乱码。这个参数告诉解码器“后面还有数据攒一攒再解码”。我光这个问题就被同事拉着排查过一下午。4. ToolCall 方案让模型真正“下地干活”4.1 bind_tools 与函数调用链路前面讲 OutputParser 解决的是“文本 → 数据”ToolCall 解决的是“意图 → 动作”。在 AI Agent 场景里模型不能只回嘴皮子它得会查订单、写工单、调 API。LangChain 给我们提供了一套统一的函数调用抽象先定义工具再绑定到模型上模型在推理时如果要调用工具就会在输出里带出结构化的tool_calls。from langchain_core.tools import tool from langchain_core.pydantic_v1 import BaseModel, Field from langchain_openai import ChatOpenAI class OrderQueryInput(BaseModel): order_id: str Field(description订单号形如 ORD-2024-0815) need_detail: bool Field(defaultFalse, description是否需要物流明细) tool(args_schemaOrderQueryInput) def query_order(order_id: str, need_detail: bool False) - str: 根据订单号查询订单状态和物流信息当用户询问订单进度、物流状态时使用。 # 这里替换成你的真实业务查询逻辑 return f订单 {order_id} 状态: 已发货; 物流: 顺丰 SF123456 llm ChatOpenAI(modelgpt-4o, temperature0) llm_with_tools llm.bind_tools([query_order]) result llm_with_tools.invoke(我的订单 ORD-2024-0815 什么时候到) print(result.tool_calls) # [{name: query_order, args: {order_id: ORD-2024-0815, need_detail: True}, id: call_abc123, type: tool_call}]有几个要点我要特别强调。第一tool装饰器下面函数的 docstring 就是给模型看的“工具说明书”非常重要。模型不会读你代码注释它只读 docstring Pydantic 字段描述来决定什么时候调用、传什么参数。实践里我见过太多人 docstring 写一句“查询订单”就完事结果模型把用户消息里所有字段乱塞进参数。我的经验是 docstring 里至少写清楚工具什么时候用、什么时候不要用、每个参数的含义、返回什么格式。第二args_schema比直接在函数签名里写类型注解更稳。Pydantic 模型可以加Field(description...)这个 description 会映射到模型 API 的parametersschema 里直接影响模型生成参数的质量。不要图省事儿只写order_id: str。第三temperature记得调低。工具调用的参数生成是“确定性任务”temperature0 能明显减少参数幻觉。这个设置单独为工具绑定场景调不影响你普通对话链路的参数。4.2 ToolCall 结果的解析与路由拿到result.tool_calls之后它只是一个声明——模型“想”调用这个工具但工具到底执行没有、执行结果是什么需要你自己完成“执行 → 回填 → 再交给模型”的循环。这个循环如果用 LangGraph可以直接用ToolNode自动包一层from langgraph.prebuilt import ToolNode, tools_condition from langgraph.graph import StateGraph, START, END from langgraph.graph.message import add_messages from typing import Annotated, TypedDict class AgentState(TypedDict): messages: Annotated[list, add_messages] def build_graph(): graph StateGraph(AgentState) graph.add_node(llm, llm_with_tools) graph.add_node(tools, ToolNode([query_order])) graph.add_edge(START, llm) graph.add_conditional_edges(llm, tools_condition, {tools: tools, END: END}) graph.add_edge(tools, llm) return graph.compile()tools_condition会自动检查上一步的 AIMessage 里有没有tool_calls有就进工具节点执行完把结果以 ToolMessage 形式追加回消息列表再让模型根据结果生成最终回复。这就是所谓的“Agent 循环”。如果你不想引入 LangGraph手写循环也不复杂核心逻辑messages [{role: user, content: 我的订单什么时候到}] for _ in range(5): # 最多轮询 5 轮防止死循环 response llm_with_tools.invoke(messages) messages.append(response) if not response.tool_calls: break # 模型没有工具调用欲望输出最终回复 for tc in response.tool_calls: tool_result execute_tool(tc[name], tc[args]) messages.append({ role: tool, tool_call_id: tc[id], content: str(tool_result), })这里一个隐藏 bug 是openai 兼容接口要求 tool message 必须携带tool_call_id这个 id 必须和你收到的 tool_calls 里那个 id 一一对应。很多人自己手写循环时忘了带导致第二次请求直接 400。用 LangGraph 的 ToolNode 就没这个问题所以我后来基本都直接上 LangGraph。另外如果 Agent 流程需要人工审批human-in-the-loopLangGraph 的interrupt机制配合 Agent Inbox 模式可以在 ToolNode 执行前挂起把待确认的 tool_call 发给用户确认后再继续。这类场景在工单创建、付款确认类 Agent 里特别常见值得单独玩玩。4.3 with_structured_output结构化输出与工具调用的统一入口前面我在解析器那节埋了个伏笔现在展开。LangChain 0.2 里有一个非常实用的方法with_structured_output()。它底层做的事情其实是“把你的 Pydantic 模型变成工具定义强制模型走 tool calling / function calling 通道来输出”。class TicketInfo(BaseModel): title: str Field(description工单标题) category: Literal[shipping, refund, quality, other] Field(description分类) urgency: int Field(description1-5, ge1, le5) action: str Field(description处理动作) structured_llm llm.with_structured_output(TicketInfo) result structured_llm.invoke(用户说订单三天没到要求马上退款) # 直接得到 TicketInfo 实例类型已校验这比 “Prompt 拼接 format_instructions 解析器” 稳得多因为模型厂商对 function calling 有专门的微调结构化输出的可靠性普遍比“从自由文本里解析 JSON”高一个量级。我的经验里同一个模型用with_structured_output做分类任务的字段正确率比 PydanticOutputParser 高约 5~8 个百分点虽然不算巨大但在生产环境已经很值了。那什么时候用 OutputParser、什么时候用 with_structured_output我的判断标准是用的是 OpenAI / Anthropic / 通义等有原生 function calling 支持的模型就用with_structured_output用的是本地 vLLM 部署的模型且没有开 function calling或者输出格式要求特别“软”比如只要一个 JSON、schema 每天都在变就用 OutputParser两者也可以组合ToolCall 本身就用bind_tools最终回答需要结构化给前端时再用with_structured_output生成一个附带的结构化摘要。5. 常见问题与排查实录5.1 stream disconnected before completion: idle timeout这个报错我在生产环境至少见过三种形态Nginx 层的proxy_read_timeout、云厂商负载均衡的连接空闲限制、以及浏览器对 fetch 流的内部超时。表象都是一个流没推完连接被服务端或中间层掐断。排查路径我建议按这个顺序走先确认是不是你的后端程序主动崩了日志里有没有异常再确认是不是 Nginx/网关层超时看 access log 的 upstream_status出现 499 或 502 基本就是这层问题最后确认是不是前端断开的浏览器 Network 面板看响应是否中断。解决方案分三层后端在生成器里加心跳注释帧: keep-alive\n\n15 秒一条能顶住绝大多数 idle 超时Nginx显式配置proxy_read_timeout 300s; proxy_buffering off;同时加上X-Accel-Buffering: no头前端对 fetch 流做异常重连捕获 TypeError网络中断会抛这个后根据断点位置决定是重发整请求还是提示用户重试。其中续传这个事目前没有完美方案我一般策略是断路器 用户手动重试并明确记录已经拼接的文本。流式应用对于断线要有一个共识宁可让用户点一下重试也不能自动重发导致重复内容。5.2 流式 JSON 解析失败与半截帧做流式聊天时最大痛点你在流式输出过程中想提前把 JSON 里的title字段抽出来做实时展示但模型还只输出到一半JSON 就是个残缺品。这里我的方案是“分两路推进”第一路展示文本流直接用 token 拼接不解析 第二路用一个partial_json_parser持续尝试把已收到的文本解析成“尽可能完整的 JSON”把能提取的字段实时更新到前端状态里。一旦最终流结束再走一次完整 Pydantic 校验覆盖掉中间值。LangChain 的 JsonOutputParser 本质上就是在做这件事它内部有parse_partial_json容错逻辑所以如果你用astream时每一帧都对 parse 前的文本调用parser.parse_partial_json(text)就能拿到“当前最接近完整”的 dict。注意要 catch 异常因为极端情况下解析器也救不回来。另外一个特别 dirty 的场景模型输出的 JSON 里带了多余的前缀或后缀比如“好的以下是结果{...}”。好在 LangChain 的extract_json会做智能抽取能容忍这种情况。但如果你用原生json.loads去解析八成会踩坑。所以我建议解析一律走 LangChain 的 parser不要自己图省事直接 load。5.3 SSE 中文字符与编码问题这个坑来得毫无防备。FastAPI 的 StreamingResponse 默认会帮你做 UTF-8 编码但如果你用sse-starlette之类的库或者自己拼data:字符串时忘了ensure_asciiFalse中文就会以\uXXXX形式推给前端。前端拿到合法 JSON 却显示成“\u5f00\u59cb”字符串而不是“开始”。解决办法就一条服务端生成 SSE 帧时统一json.dumps(data, ensure_asciiFalse)前端解码统一TextDecoder(utf-8)JSON.parse两边都保持“JSON 内不转义、传输层不二次转义”。还有一个隐藏点如果服务端用yield fdata: {json_str}\n\n而json_str本身含换行会破坏 SSE 帧格式。因为 SSE 帧的 data 不能包含裸换行解决办法是json_str.replace(\n, \\n)。这个我见过有人踩坑前端疯狂解析错误。5.4 ToolCall 参数幻觉与兜底最后聊聊 ToolCall 最常见的问题参数幻觉。模型“觉得”用户想查订单就编了一个order_id: ORD-2024-0815但用户根本没提供订单号。这类问题在小模型上尤其严重。我的兜底方案Pydantic 模型里给参数加min_length之类的约束或者用自定义校验检查订单号前缀工具执行前做二次校验参数不合法直接返回工具错误信息让模型知道“参数错了请向用户要正确的订单号”而不是崩溃工具执行结果统一返回结构化文本不要返回 Python 对象避免模型在下一轮推理时被非文本对象搞晕。第二个方案我特别推荐把“工具参数错误”当成一次正常的工具结果返回给模型模型会很自然地生成“抱歉我需要确认您的订单号”这类回复。这比让程序抛异常、把整个 Agent 循环打挂体验好 10 倍。最后再分享一个我的实战习惯。上面这套链路我每次接到新项目都不会急着写业务代码而是先把“一条消息从进来到出结果”的完整链路用最小 demo 跑通然后把五个关键节点打上日志入参、LLM 原始输出、解析器结果、ToolCall 参数、最终结构化输出。这五个节点的日志打全了后面上线排查基本不用动脑子直接看日志点就能定位是哪一环出了幺蛾子。工期紧的时候这个习惯帮我把联调时间压缩了至少一半你也试试。