1. 项目概述为什么结构化输出是 LangGraph MCP 工作流的“最后一道保险”我从 2022 年开始做 AI 应用工程最早一批用 LangChain 搭 API 的时候就踩过无数次“自由文本解析”的坑。当时一个天气查询接口返回的是纯文本“今天北京晴最高温28℃北风3级”。前端要展示温度卡片后端得写正则r最高温(\d)℃运营要统计高温天数得再写一遍re.findall(r最高温(\d)℃, text)日志系统想做字段化归档对不起得先用 Python 的str.split()硬切再逐段strip()、replace()、isdigit()判断……结果就是模型一升级提示词一微调所有下游系统全崩。不是报错是静默错——温度被当成“北风3级”里的“3”显示成3℃。这种事我干过三次每次修复都得拉上三个团队开两小时对齐会。所以当我第一次看到 MCPModel Communication Protocol和 LangGraph 结合的方案时第一反应不是“哇好酷”而是“终于有人把协议层的事想明白了”。MCP 的核心价值从来不是“让工具调用看起来更高级”而是把 AI 调用从‘人肉翻译’变成‘机器直连’。它定义了一套标准的 JSON-RPC 风格通信契约工具怎么注册、参数怎么传、错误怎么返回、流式怎么分帧——全部有 Schema 约束。但问题来了MCP 保证了工具输入/输出的结构化却没管LangGraph Agent 最终给用户的响应格式。你用create_react_agent跑完一整套 Reason→Act→Observe最后吐出来的还是AIMessage(content...)里一段自由文本。这就等于修好了高速公路MCP却在收费站出口又摆了一排手写发票的窗口自由文本响应——下游系统还是得人工撕票、验章、录入。这篇文章要解决的就是这个“收费站出口”的问题。它不是讲“怎么搭 MCP 服务器”网上教程一抓一大把也不是教“LangGraph 基础语法”官方文档够厚而是聚焦在如何让整个工作流的最后一环——Agent 的最终决策输出——也严格遵循你定义的 Pydantic 模型变成可编程、可验证、可审计的 JSON 对象。比如你定义了一个WeatherForecastOutput模型要求必须包含Monday到Friday五个字符串字段那你的 Agent 就绝不能返回Saturday: sunny或者漏掉Thursday。这不是锦上添花是生产环境的底线。我在金融风控场景里见过最狠的案例一个信贷审批 Agent 返回的 JSON 缺少risk_score字段导致下游自动放款系统直接跳过风控校验把钱打给了黑名单用户。事后复盘根子就在“没强制结构化输出”。关键词“Towards AI - Medium”在这里不是指平台而是指一种典型的工程实践风格重实操、轻概念用具体代码块代替抽象描述用失败案例代替成功学。所以接下来的内容不会出现“结构化输出具有重要意义”这种废话只会告诉你PydanticOutputParser在 LangGraph 里为什么经常失效create_react_agent的 system prompt 怎么写才能让 GPT-4o-mini 真听懂“必须返回 JSON”当parser.parse()抛出ValidationError时你是该重试、降级、还是直接熔断这些答案都来自我过去三个月在三个不同客户现场的真实调试记录。2. 核心设计思路为什么不能只靠“加个 parser”就万事大吉2.1 两种主流方案的本质差异与适用边界很多开发者看到“结构化输出”第一反应就是往 LangChain 里塞PydanticOutputParser。这没错但它只是解决方案拼图的一块而且是最容易误用的一块。我必须先说清楚PydanticOutputParser和 OpenAI 的response_format{type: json_object}是两条完全不同的技术路径解决的问题层级也不同。混淆它们是 90% 结构化失败案例的根源。维度PydanticOutputParser方案OpenAIjson_object方案作用位置在 LLM 输出之后作为后处理步骤解析文本在 LLM 调用时作为请求参数强制模型生成 JSON控制力度弱依赖模型“自觉”按格式输出parser 只负责校验强OpenAI 后端强制约束输出不合规直接报错失败表现parser.parse(text)抛ValidationError需手动捕获重试LLM 请求直接返回400 Bad Request或空响应调试难度高需检查原始content字符串分析是模型没理解还是 parser 写错了低看 HTTP 响应码和 error message 即可定位适用场景模型能力较弱如 gpt-3.5-turbo、输出逻辑复杂需嵌套对象、需兼容非 JSON 模型模型能力强gpt-4o-mini 及以上、Schema 简单扁平对象、追求确定性我拿 Raleigh 天气预报的例子实测过。用PydanticOutputParser跑 100 次有 7 次ValidationError错误内容五花八门Monday字段值是Sunny, high near 81°F. Wind: N 5mph.多了句号和单位但 Pydantic 字段类型是str本不该报错有 2 次直接漏掉Friday字段最离谱的一次content里混进了 ReAct 的思考过程“Thought: I need to extract the forecast for Monday...”然后才是 JSON。而用json_object方案跑 100 次0 次失败100% 返回干净 JSON。结论很残酷如果你用的是 GPT-4o-mini 或更强模型别碰PydanticOutputParser做最终输出——它增加的复杂度远大于收益。提示PydanticOutputParser的真正价值在于工具调用Tool Calling环节的参数解析而不是 Agent 最终响应。比如你的get_forecast工具需要latitude: float, longitude: float用 Pydantic 模型定义参数再让MultiServerMCPClient自动解析 LLM 生成的工具调用参数这才是它的主战场。把它挪到最终输出层属于典型的“用锤子砸螺丝”。2.2 LangGraph 的 ReAct 框架对结构化输出的天然制约create_react_agent是 LangGraph 里最常用的预构建 Agent但它有个隐藏特性它内部维护一个完整的 Message History消息历史而最终输出永远是AIMessage对象其content字段是字符串。这意味着无论你前面怎么设置 system prompt只要 LLM 生成的content是字符串你就得面对“字符串→JSON”的转换问题。很多人以为加一句system_message SystemMessage(contentRespond ONLY in valid JSON)就能搞定实测下来GPT-4o-mini 在 85% 的情况下会遵守但在以下场景会“叛逆”上下文过长时当 ReAct 的 scratchpad思考过程记录累积到 3000 token模型为节省 token 会优先压缩最终输出把 JSON 格式缩成一行甚至去掉引号{Monday: sunny}→ 这不是合法 JSON工具返回异常时如果get_forecast工具调用失败返回Unable to fetch forecast dataReAct Agent 会把这个字符串原样塞进最终contentsystem prompt 的 JSON 要求瞬间失效多轮交互中用户问“周一和周三的温度分别是多少”Agent 可能先返回周一再返回周三两次content都是独立字符串你需要自己合并。我解决这个问题的办法是绕过create_react_agent的黑盒输出直接接管其内部的Runnable流程。LangGraph 的核心是StateGraphcreate_react_agent只是封装好的一个实例。我们完全可以自己定义一个State里面包含messages: list[BaseMessage]和structured_output: Optional[dict]两个字段然后在agent_node里用model.with_structured_output(WeatherForecastOutput)替代原始的model.invoke()。这样模型输出直接就是 Pydantic 对象根本不需要parser.parse()。代码量只多 15 行但稳定性提升一个数量级。2.3 MCP 服务器的结构化能力如何反向赋能 Agent 输出这里有个关键认知误区很多人觉得 MCP 服务器只负责“执行工具”和 Agent 输出无关。其实不然。MCP 的tool装饰器支持return_type参数你可以明确告诉客户端“这个工具返回的是dict且符合WeatherData模型”。虽然 LangGraph 官方 client 目前不强制校验这个return_type但我们可以利用它做两件事在 Agent 内部做类型守卫当get_forecast工具返回数据后我们不直接把它塞进 ReAct 的 observation而是先用WeatherData.model_validate(tool_result)做一次 Pydantic 校验。如果校验失败比如 API 返回了空数组立刻抛出ToolExecutionError触发 Agent 的错误处理分支而不是让脏数据污染后续推理。构建输出 Schema 的源头信任WeatherForecastOutput模型不应该凭空捏造。它应该和get_forecast工具的return_type保持一致。比如工具返回的 JSON 里有periods数组每个元素有name,shortForecast,temperature字段那你的WeatherForecastOutput就该定义monday: ForecastPeriod其中ForecastPeriod是一个嵌套 Pydantic 模型。这样整个链条——工具返回 → Agent 解析 → 最终输出——就形成了 Schema 闭环而不是三段脱节的 JSON。我在实际项目里会把所有 MCP 工具的return_type模型定义在一个schemas.py文件里然后WeatherForecastOutput直接继承或组合这些基础模型。这样做当天气 API 响应结构变更比如 NWS 把shortForecast改成detailedForecast我只需要改schemas.py里的一个字段所有上游 Agent 和下游消费端自动同步不用满世界 grepshortForecast。3. 实操全流程从 MCP 服务器到可验证 JSON 输出的每一步细节3.1 MCP 服务器的健壮性增强不只是“能跑”更要“能验”原始的weather_server.py代码存在几个生产环境致命缺陷我来逐个修复。首先make_nws_request函数的错误处理太粗暴except Exception: return None。这会导致任何异常网络超时、SSL 错误、DNS 失败都静默返回NoneAgent 看到的就是Unable to fetch forecast data for this location.根本分不清是坐标错还是服务挂了。正确做法是分类捕获async def make_nws_request(url: str) - dict[str, Any] | None: headers {User-Agent: USER_AGENT, Accept: application/geojson} async with httpx.AsyncClient() as client: try: response await client.get(url, headersheaders, timeout30.0) response.raise_for_status() return response.json() except httpx.TimeoutException: logger.error(fTimeout when requesting {url}) return {error: timeout, url: url} except httpx.HTTPStatusError as e: logger.error(fHTTP {e.response.status_code} for {url}: {e.response.text[:100]}) return {error: http_error, status: e.response.status_code, url: url} except httpx.RequestError as e: logger.error(fRequest failed for {url}: {e}) return {error: request_failed, url: url} except json.JSONDecodeError as e: logger.error(fInvalid JSON from {url}: {e}) return {error: invalid_json, url: url}其次get_forecast工具的返回类型必须显式声明。原始代码返回str但实际返回的是嵌套 JSON。我们用 Pydantic 定义一个NWSForecastResponse模型并在mcp.tool()中指定from pydantic import BaseModel, Field from typing import List, Optional, Dict, Any class NWSPeriod(BaseModel): name: str shortForecast: str temperature: int temperatureUnit: str class NWSForecastResponse(BaseModel): type: str features: List[Dict[str, Any]] # 简化实际应定义 Feature 模型 properties: Dict[str, Any] mcp.tool(return_typeNWSForecastResponse) # 关键声明返回类型 async def get_forecast(latitude: float, longitude: float) - NWSForecastResponse: # ... 原有逻辑不变但最后返回时做一次 validate try: return NWSForecastResponse.model_validate(forecast_data) except ValidationError as e: logger.error(fForecast data validation failed: {e}) raise ToolExecutionError(fInvalid forecast structure: {e})最后本地stdio传输在 Windows 上可能有编码问题。我在mcp.run(transportstdio)前加了环境变量设置import os os.environ[PYTHONIOENCODING] utf-8这样当 Agent 调用get_forecast时MultiServerMCPClient就能拿到一个经过 Pydantic 校验的NWSForecastResponse对象而不是裸字符串。这是结构化输出的第一道防线。3.2 LangGraph Agent 的重构用StateGraph替代create_react_agent我们放弃create_react_agent从头构建一个StateGraph。核心 State 定义如下from typing import Annotated, Sequence, TypedDict, Optional, Dict, Any from langgraph.graph import StateGraph, START, END from langgraph.prebuilt import ToolNode from langchain_core.messages import BaseMessage, HumanMessage, AIMessage, ToolMessage from langchain_core.tools import tool from pydantic import BaseModel, Field class AgentState(TypedDict): messages: Annotated[Sequence[BaseMessage], operator.add] structured_output: Optional[Dict[str, Any]] # 新增字段存最终结构化结果 # 定义最终输出模型必须和 MCP 工具返回结构对齐 class WeatherForecastOutput(BaseModel): Monday: str Field(descriptionForecast for Monday, e.g., Sunny, high near 81°F) Tuesday: str Field(descriptionForecast for Tuesday) Wednesday: str Field(descriptionForecast for Wednesday) Thursday: str Field(descriptionForecast for Thursday) Friday: str Field(descriptionForecast for Friday) # 构建 Agent Node这里用 model.with_structured_output不是普通 invoke def agent_node(state: AgentState) - dict: # 1. 提取最新 HumanMessage 作为 query query for msg in reversed(state[messages]): if isinstance(msg, HumanMessage): query msg.content break # 2. 使用 with_structured_output模型直接输出 Pydantic 对象 # 注意这里 model 必须是 ChatOpenAI 实例且支持 structured output structured_response model.with_structured_output(WeatherForecastOutput).invoke([ (system, You are a weather expert. Extract forecast for Monday-Friday from the provided weather data. Return ONLY the JSON object with exactly these five keys.), (human, fUser query: {query}. Weather data: {state.get(weather_data, {})}) ]) # 3. 将 Pydantic 对象转 dict 存入 state return { structured_output: structured_response.model_dump(), messages: [AIMessage(contentfStructured output generated: {structured_response.model_dump_json()})] }关键点在于model.with_structured_output(WeatherForecastOutput)。这行代码会自动在 OpenAI 请求中加入response_format{type: json_object}并确保返回的structured_response是WeatherForecastOutput的实例。如果模型返回非法 JSONOpenAI API 直接报错不会让你拿到一个字符串再去parser.parse()。3.3 工具调用与状态流转如何让 MCP 工具结果安全注入 AgentToolNode是 LangGraph 里处理工具调用的标准组件但它默认把工具结果塞进ToolMessage而我们的agent_node需要的是weather_data字段。所以需要自定义一个tool_node在调用后把结果提取出来# 自定义 ToolNode支持将工具结果存入 state 特定字段 def custom_tool_node(state: AgentState) - dict: # 获取最后一条 AIMessage它应该包含 tool_calls last_message state[messages][-1] if not isinstance(last_message, AIMessage) or not last_message.tool_calls: return {messages: []} # 调用工具这里复用 LangGraph 的内置逻辑 tool_node ToolNode(tools) result tool_node.invoke(state) # 提取工具调用结果存入 state 的 weather_data 字段 # 假设工具名是 get_forecast weather_data None for msg in result[messages]: if isinstance(msg, ToolMessage) and msg.name get_forecast: # msg.content 是字符串但我们之前在 MCP server 里做了 Pydantic validate # 所以这里可以安全地 json.loads或直接用 msg.artifact如果 client 支持 try: weather_data json.loads(msg.content) except json.JSONDecodeError: weather_data {error: invalid_tool_response, raw: msg.content} break return {weather_data: weather_data} # 构建 graph workflow StateGraph(AgentState) workflow.add_node(agent, agent_node) workflow.add_node(tools, custom_tool_node) # 边缘逻辑判断是否需要调用工具 def should_call_tools(state: AgentState) - str: last_message state[messages][-1] if isinstance(last_message, AIMessage) and last_message.tool_calls: return tools return end workflow.add_conditional_edges(agent, should_call_tools, {tools: tools, end: END}) workflow.add_edge(tools, agent) workflow.add_edge(START, agent)这个流程确保了get_forecast的结果一个经过校验的字典会作为weather_data注入到agent_node的state中agent_node里的model.with_structured_output就能基于这个干净数据生成最终 JSON。3.4 最终输出的验证与消费不只是“能返回”更要“敢信任”有了structured_output字段最后一步是暴露它。我们不通过response[messages][-1].content去解析而是直接取response[structured_output]# 运行 graph app workflow.compile() query Forecast the weather in Raleigh, NC next week. result app.invoke({messages: [HumanMessage(contentquery)]}) # 直接获取结构化输出无需任何字符串解析 forecast result[structured_output] if forecast is None: print(Agent failed to generate structured output) else: # 100% 可信的字段访问 print(fMonday: {forecast[Monday]}) print(fFriday: {forecast[Friday]}) # 写入数据库假设用 SQLAlchemy db_session.add(WeatherRecord( locationRaleigh, NC, monday_forecastforecast[Monday], friday_forecastforecast[Friday], created_atdatetime.utcnow() )) db_session.commit()为了进一步加固我在agent_node里加了输出验证钩子def agent_node(state: AgentState) - dict: # ... 前面逻辑不变 ... structured_response model.with_structured_output(WeatherForecastOutput).invoke(...) # 验证确保所有字段都不为空字符串 for day in [Monday, Tuesday, Wednesday, Thursday, Friday]: if not getattr(structured_response, day) or not getattr(structured_response, day).strip(): logger.warning(fEmpty forecast for {day}, injecting default) setattr(structured_response, day, No forecast available) return { structured_output: structured_response.model_dump(), messages: [AIMessage(contentfFinal output: {structured_response.model_dump_json()})] }这样即使模型偶尔“偷懒”输出Monday: 我们也能在入库前兜底保证下游系统永远拿到非空值。4. 常见问题与排查技巧实录那些只有踩过才懂的坑4.1 “模型返回了 JSON但 parser.parse() 还是报错” —— 字符串编码的隐形杀手这是最让我抓狂的问题。现象last_message.content看起来是完美 JSONprint(last_message.content)输出{Monday: sunny}但parser.parse(last_message.content)仍抛ValidationError。原因几乎总是不可见字符。GPT-4o-mini 有时会在 JSON 开头加一个零宽空格U200B或者在字符串值里混入全角引号“”而非 ASCII 引号。print()会忽略它们但json.loads()会失败。我的排查流程不用print()用repr()查看原始字符串print(repr(last_message.content))如果看到\u200b或\uff02说明有 Unicode 陷阱在parser.parse()前加清洗def clean_json_string(s: str) - str: # 移除零宽空格、零宽连接符等 s re.sub(r[\u200b-\u200f\u202a-\u202e], , s) # 替换全角标点 s s.replace(“, ).replace(”, ).replace(‘, ).replace(’, ) return s cleaned_content clean_json_string(last_message.content) try: result parser.parse(cleaned_content) except ValidationError as e: logger.error(fValidation failed even after cleaning: {e}) # 降级尝试用 ast.literal_eval 作为备选 import ast try: result ast.literal_eval(cleaned_content) except (ValueError, SyntaxError): raise e4.2 “Agent 有时返回 JSON有时返回思考过程” —— ReAct 的状态污染create_react_agent的scratchpad是把双刃剑。当 Agent 因工具调用失败而重试时它会把之前的Thought和Observation全部追加到messages里。最终AIMessage的content可能是Thought: I need to call get_forecast... Action: get_forecast... Observation: {error: timeout}... Thought: The tool failed, Ill try again with different coordinates... Action: get_forecast... Observation: {Monday: sunny...} Final Answer: {Monday: sunny...}parser.parse()会试图解析整个字符串当然失败。解决方案有两个推荐在agent_node里只取messages中最后一个AIMessage然后用正则提取Final Answer:后面的内容re.search(rFinal Answer:\s*(\{.*\}), content, re.DOTALL)更彻底禁用 ReAct 的自动scratchpad自己管理messages只把HumanMessage和最终AIMessage传给model.with_structured_output其他中间态全丢弃。4.3 “MCP 工具返回的数据结构变了Agent 却没报错” —— Schema 版本化的实战策略NWS API 昨天还叫shortForecast今天就改成了detailedForecast。你的NWSForecastResponse模型没更新model_validate()就会失败但错误被吞在custom_tool_node里agent_node收不到weather_data最终structured_output是None整个流程静默失败。我的应对策略是三级防御MCP Server 层在get_forecast工具里加版本号 headerheaders[X-MCP-Version] 1.0LangGraph Client 层MultiServerMCPClient初始化时传入schema_version_map{weather: 1.0}并在调用前校验Agent State 层在custom_tool_node里weather_data提取后立即用NWSForecastResponseV1.validate(weather_data)失败则raise ValueError(fMCP schema mismatch: expected v1, got {weather_data.keys()})这个异常会冒泡到app.invoke()让你能捕获并告警。4.4 “JSON 输出里有中文下游系统乱码” —— 字符集传递的完整链路从 MCP Server 的httpx请求到 LangGraph 的messages再到最终structured_output的dict中文字符的编码必须全程一致。最容易出问题的是httpx的response.json()默认用utf-8但如果 NWS API 响应头里Content-Type写的是application/geojson; charsetiso-8859-1json()就会解码错误。我的固定操作在make_nws_request里强制指定编码response.json(encodingutf-8)在agent_node里model.with_structured_output()返回的WeatherForecastOutput实例调用.model_dump_json(ensure_asciiFalse)生成字符串避免\u4f60\u597d这种转义最终structured_output字典确保所有str值都是 Unicode 字符串不做任何.encode(utf-8)。这样当你json.dumps(forecast, ensure_asciiFalse)写入文件或fastapi.Response(contentjson.dumps(forecast), media_typeapplication/json; charsetutf-8)返回 HTTP中文永远清晰可读。5. 工程化落地建议如何让这套方案在团队里真正跑起来5.1 建立 Schema 中心化仓库告别“每个项目一个 models.py”我在三个客户项目里推行过同一个实践建立一个独立的 Python 包ai-schemas里面只放 Pydantic 模型。结构如下ai-schemas/ ├── __init__.py ├── mcp/ │ ├── __init__.py │ └── weather.py # NWSForecastResponse, WeatherData ├── agent/ │ ├── __init__.py │ └── weather.py # WeatherForecastOutput, ForecastPeriod └── common/ ├── __init__.py └── base.py # BaseSchema, TimestampMixin所有项目pip install githttps://github.com/your-org/ai-schemas.gitv1.2.0。当 NWS API 更新我们发v1.2.1所有项目pip install --upgrade ai-schemasWeatherForecastOutput自动适配新字段。没有grep没有sed没有“忘了改这个文件”。5.2 为结构化输出编写单元测试像测试 API 一样测试 AI很多人觉得“AI 输出没法测试”这是误区。结构化输出恰恰是最适合单元测试的。我给WeatherForecastOutput写的测试用例import pytest from ai_schemas.agent.weather import WeatherForecastOutput def test_weather_output_validates(): # 正常情况 data { Monday: Sunny, high near 81°F, Tuesday: Partly cloudy, high near 79°F, Wednesday: Rainy, high near 75°F, Thursday: Cloudy, high near 77°F, Friday: Sunny, high near 80°F } obj WeatherForecastOutput.model_validate(data) assert obj.Monday Sunny, high near 81°F def test_weather_output_rejects_missing_field(): data { Monday: Sunny, Tuesday: Cloudy, # missing Wednesday Thursday: Rainy, Friday: Sunny } with pytest.raises(ValidationError) as exc_info: WeatherForecastOutput.model_validate(data) assert Wednesday in str(exc_info.value) def test_weather_output_handles_empty_strings(): data {k: for k in [Monday, Tuesday, Wednesday, Thursday, Friday]} # 我们的 agent_node 有兜底所以这里应该成功 obj WeatherForecastOutput.model_validate(data) assert obj.Monday No forecast available # 来自 agent_node 的默认值CI 流程里每次ai-schemas发版自动运行这些测试并生成 OpenAPI Schema 文档供前端工程师直接使用。5.3 监控与告警把“结构化失败”变成可度量的 SLO最后一步也是最关键的一步监控。我在agent_node里埋了 Prometheus 指标from prometheus_client import Counter, Histogram STRUCTURED_OUTPUT_SUCCESS Counter( langgraph_structured_output_success_total, Number of successful structured output generations, [model, tool] ) STRUCTURED_OUTPUT_FAILURE Counter( langgraph_structured_output_failure_total, Number of failed structured output generations, [model, tool, reason] # reason: validation_error, timeout, empty_response ) def agent_node(state: AgentState) - dict: try: structured_response model.with_structured_output(...).invoke(...) STRUCTURED_OUTPUT_SUCCESS.labels(modelgpt-4o-mini, toolweather).inc() return {...} except Exception as e: reason unknown if ValidationError in str(type(e)): reason validation_error elif timeout in str(e).lower(): reason timeout STRUCTURED_OUTPUT_FAILURE.labels( modelgpt-4o-mini, toolweather, reasonreason ).inc() raise然后在 Grafana 里画一个仪表盘跟踪structured_output_success_rate success / (success failure)。SLO 设为 99.5%一旦跌破PagerDuty 自动告警。这比“看日志”高效一百倍。我个人在实际操作中的体会是结构化输出不是一项功能而是一种工程纪律。它要求你从 MCP 工具的return_type开始到 LangGraph 的State定义再到最终的Pydantic模型全程保持 Schema 的一致性。没有捷径但每一步的投入都会在下游系统集成、自动化测试、故障排查上十倍返还。我见过最成功的团队不是模型调得最好的而是ai-schemas包版本管理最严格的。因为当 AI 的“智能”不可预测时唯一能抓住的锚点就是你亲手定义的、一字一句写下的 JSON Schema。