LangChain 结构化输出终于讲透了:ProviderStrategy、ToolStrategy、动态 Schema 一篇全会
LangChain 结构化输出:从入门到踩坑
摘要:结构化输出让 Agent 返回可预测的 JSON/Pydantic 模型,而不是自然语言。本文深入解析 Provider Strategy 和 Tool Strategy 两种策略,附完整代码和踩坑经验。
一、为什么会写这篇
最近在项目里做 Agent 开发,遇到一个头疼的问题:大模型返回的内容格式飘忽不定,有时候是 JSON,有时候是纯文本,解析起来特别痛苦。
后来发现 LangChain 提供了**结构化输出(Structured Output)**机制,能让 Agent 按照我们定义的格式返回数据。听起来简单,实际踩了不少坑。
门主这篇文章把结构化输出的两种策略讲清楚,顺便把踩过的坑分享出来,帮你少走弯路。
二、什么是结构化输出
结构化输出允许Agent以特定的、可预测的格式返回数据。这样你无需解析自然语言响应,就能获得JSON 对象、Pydantic 模型或数据类(dataclasses)形式的结构化数据,供应用程序直接使用。
简单说就是:你定义好格式,模型按格式返回。
from pydantic import BaseModel, Fieldfrom langchain.agents import create_agentclass Answer(BaseModel): summary: str confidence: floatagent = create_agent(model="openai:gpt-5.5", response_format=Answer)result = agent.invoke({"messages": [{"role": "user", "content": "总结 AI 趋势"}]})# 直接拿到结构化对象print(result["structured_response"]) # Answer(summary=..., confidence=...)三、响应格式类型
LangChain 的create_agent通过response_format参数控制结构化输出方式:
| 类型 | 说明 | 适用场景 |
|---|---|---|
ToolStrategy | 通过工具调用实现结构化输出 | 所有支持工具调用的模型 |
ProviderStrategy | 使用提供商原生结构化输出 | OpenAI、Anthropic、xAI 等 |
type[Schema] | 自动选择最佳策略 | 推荐写法 |
None | 不请求结构化输出 | 默认 |
自动选择逻辑:
四、提供商策略(Provider Strategy)
4.1 原理
部分模型提供商通过 API 原生支持结构化输出(如 OpenAI、xAI、Gemini、Anthropic)。这是最可靠的方式。
当模型支持原生结构化输出时,直接传 Schema 类型即可自动启用:
from pydantic import BaseModel, Fieldfrom langchain.agents import create_agentclass ContactInfo(BaseModel): """联系人信息""" name: str = Field(description="姓名") email: str = Field(description="邮箱") phone: str = Field(description="电话")# 自动选择 ProviderStrategyagent = create_agent( model="openai:gpt-5.5", response_format=ContactInfo)result = agent.invoke({ "messages": [{"role": "user", "content": "提取联系人:张三, zhangsan@example.com, 13800138000"}]})print(result["structured_response"])# ContactInfo(name='张三', email='zhangsan@example.com', phone='13800138000')4.2 支持的 Schema 类型
| Schema 类型 | 返回类型 | 特点 |
|---|---|---|
| Pydantic Model | Pydantic 实例 | 支持字段验证,推荐 |
| Dataclass | dict | 简单轻量 |
| TypedDict | dict | 类型提示友好 |
| JSON Schema | dict | 灵活但无代码提示 |
4.3 严格模式
ProviderStrategy支持strict参数启用严格模式(需要langchain>=1.2):
from langchain.agents.structured_output import ProviderStrategyagent = create_agent( model="openai:gpt-5.5", response_format=ProviderStrategy(schema=ContactInfo, strict=True))门主提醒:严格模式要求模型完全遵守 Schema,部分国产模型可能不支持,建议先测试再上线。
五、工具调用策略(Tool Strategy)
5.1 原理
对于不支持原生结构化输出的模型,LangChain 通过工具调用实现结构化输出。模型会"假装"调用一个工具,工具的参数就是结构化数据。
这是兼容性最好的方式,几乎所有支持工具调用的模型都能用。
5.2 基本用法
from pydantic import BaseModel, Fieldfrom langchain.agents import create_agentfrom langchain.agents.structured_output import ToolStrategyfrom langchain.tools import tool# 定义输出格式class WeatherStrategy(BaseModel): city: str = Field(description="城市名称") weather: str = Field(description="天气描述") temperature: str = Field(description="温度") activity: str = Field(description="建议活动")# 定义工具@tool(description="查询城市天气的工具")def get_weather(city: str): return f"今天{city}的天气晴,温度为30度,适合户外活动"# 创建 Agentagent = create_agent( model=chanAI, tools=[get_weather], response_format=ToolStrategy(WeatherStrategy),)result = agent.invoke({ "messages": [{"role": "user", "content": "长沙今天是什么天气"}]})print(result["structured_response"])# WeatherStrategy(city='长沙', weather='晴', temperature='30度', activity='适合户外活动')5.3 执行流程
六、自定义工具消息内容
tool_message_content参数允许自定义生成结构化输出时,对话历史中显示的消息:
from langchain.agents.structured_output import ToolStrategyagent = create_agent( model=chanAI, tools=[get_weather], response_format=ToolStrategy( schema=WeatherStrategy, tool_message_content="天气查询已完成" ),)对比效果:
| 设置 | ToolMessage 内容 |
|---|---|
| 不设置 | Returning structured response: {'city': '长沙', ...} |
| 设置后 | 天气查询已完成 |
门主建议:生产环境建议自定义,方便日志排查和调试。
七、错误处理
模型在通过工具调用生成结构化输出时可能会出错。LangChain 提供了智能的重试机制。
7.1 handle_errors 参数
| 值 | 行为 |
|---|---|
True | 捕获所有错误,使用默认错误模板(默认值) |
str | 捕获所有错误,使用自定义消息 |
type[Exception] | 只捕获指定异常类型 |
Callable[[Exception], str] | 自定义错误处理函数 |
False | 不重试,直接抛出异常 |
7.2 完整示例
from typing import Literalfrom pydantic import BaseModel, Field, field_validatorfrom langchain.agents import create_agentfrom langchain.agents.structured_output import ToolStrategyfrom langchain.tools import toolclass WeatherStrategy(BaseModel): city: str = Field(description="城市名称") weather: str = Field(description="天气") temperature: str = Field(description="温度") activity: str = Field(description="建议活动") @field_validator("city") @classmethod def check_city(cls, value): # 模拟 Schema 校验失败 raise ValueError("故意触发 Schema 校验失败")@tool(description="查询天气")def get_weather(city: str): return f"城市:{city}\n天气:晴\n温度:30℃\n建议:适合出去玩"agent = create_agent( model=chanAI, tools=[get_weather], response_format=ToolStrategy( schema=WeatherStrategy, tool_message_content="天气查询完成", handle_errors=True # 开启错误重试 ),)result = agent.invoke({ "messages": [{"role": "user", "content": "长沙今天什么天气"}]})7.3 常见错误类型
| 错误类型 | 原因 | 解决方案 |
|---|---|---|
| Schema 校验失败 | 模型返回数据不符合 Schema | 检查 Schema 定义,开启重试 |
| 多次调用结构化输出工具 | 模型一次返回多个结构化数据 | LangChain 自动处理 |
| 工具调用格式错误 | 模型生成的 JSON 格式不正确 | 使用更强大的模型 |
八、多格式动态选择:不同问答不同 Schema
实际项目中,经常会遇到这种情况:
- 用户问天气 → 返回天气格式
- 用户问联系人 → 返回联系人格式
- 用户问产品信息 → 返回产品格式
总不能写死一个 Schema 吧?LangChain 提供了几种方式解决这个问题。
8.1 Union Types:多 Schema 自动匹配
ToolStrategy支持传入Union类型,模型会根据上下文自动选择最合适的 Schema:
from pydantic import BaseModel, Fieldfrom typing import Literal, Unionfrom langchain.agents import create_agentfrom langchain.agents.structured_output import ToolStrategyclass ProductReview(BaseModel): """产品评价分析""" rating: int | None = Field(description="产品评分 1-5", ge=1, le=5) sentiment: Literal["positive", "negative"] = Field(description="情感倾向") key_points: list[str] = Field(description="关键要点")class CustomerComplaint(BaseModel): """客户投诉""" issue_type: Literal["product", "service", "shipping", "billing"] = Field(description="问题类型") severity: Literal["low", "medium", "high"] = Field(description="严重程度") description: str = Field(description="问题描述")class WeatherInfo(BaseModel): """天气信息""" city: str = Field(description="城市") weather: str = Field(description="天气状况") temperature: str = Field(description="温度")# 多个 Schema 联合,模型自动选择agent = create_agent( model=chanAI, tools=tools, response_format=ToolStrategy(Union[ProductReview, CustomerComplaint, WeatherInfo]))# 模型会根据用户问题自动匹配合适的 Schemaresult1 = agent.invoke({"messages": [{"role": "user", "content": "分析这个评价:质量很好,5星推荐"}]})# → ProductReview(rating=5, sentiment='positive', key_points=['质量很好'])result2 = agent.invoke({"messages": [{"role": "user", "content": "长沙今天天气怎么样"}]})# → WeatherInfo(city='长沙', weather='晴', temperature='25°C')执行流程:
8.2 运行时动态切换 Schema
如果需要更灵活的控制,可以在不同场景下创建不同的 Agent 实例:
from langchain.agents import create_agentfrom langchain.agents.structured_output import ToolStrategy# 定义多个 Schemaclass WeatherSchema(BaseModel): city: str = Field(description="城市") temperature: str = Field(description="温度") weather: str = Field(description="天气")class ContactSchema(BaseModel): name: str = Field(description="姓名") phone: str = Field(description="电话") email: str = Field(description="邮箱")class OrderSchema(BaseModel): order_id: str = Field(description="订单号") status: str = Field(description="订单状态") amount: float = Field(description="金额")# 工厂函数:根据场景创建不同 Agentdef create_agent_by_scene(scene: str): scene_config = { "weather": { "schema": WeatherSchema, "system_prompt": "你是天气查询助手", "tool_message_content": "天气查询完成" }, "contact": { "schema": ContactSchema, "system_prompt": "你是联系人管理助手", "tool_message_content": "联系人信息提取完成" }, "order": { "schema": OrderSchema, "system_prompt": "你是订单查询助手", "tool_message_content": "订单查询完成" } } config = scene_config.get(scene, scene_config["weather"]) return create_agent( model=chanAI, response_format=ToolStrategy( schema=config["schema"], tool_message_content=config["tool_message_content"] ), system_prompt=config["system_prompt"] )# 使用示例weather_agent = create_agent_by_scene("weather")contact_agent = create_agent_by_scene("contact")8.3 根据用户意图动态路由
更智能的做法是先识别用户意图,再路由到对应的 Agent:
from pydantic import BaseModel, Fieldfrom typing import Literalclass IntentSchema(BaseModel): """用户意图识别""" intent: Literal["weather", "contact", "order", "other"] = Field(description="用户意图") extracted_info: str = Field(description="提取的关键信息")def smart_route(user_input: str): # 先用轻量模型识别意图 intent_agent = create_agent( model=chanAI, response_format=IntentSchema ) intent_result = intent_agent.invoke( {"messages": [{"role": "user", "content": user_input}]} ) intent = intent_result["structured_response"].intent # 根据意图路由到对应 Agent agent = create_agent_by_scene(intent) return agent.invoke( {"messages": [{"role": "user", "content": user_input}]} )# 使用result = smart_route("帮我查一下北京的天气")8.4 对比总结
| 方式 | 优点 | 缺点 | 适用场景 |
|---|---|---|---|
| Union Types | 简单,一行代码 | 模型可能选错 Schema | Schema 数量少,差异明显 |
| 工厂模式 | 清晰可控 | 需要预设场景 | 场景固定,数量有限 |
| 意图路由 | 最灵活,可扩展 | 多一次模型调用 | 复杂场景,Schema 数量多 |
门主建议:如果 Schema 不超过 5 个,直接用 Union 就行。场景很多的话,上意图路由更稳。
九、两种策略对比
| 维度 | Provider Strategy | Tool Strategy |
|---|---|---|
| 可靠性 | ⭐⭐⭐⭐⭐ | ⭐⭐⭐⭐ |
| 兼容性 | 仅支持原生输出的模型 | 所有支持工具调用的模型 |
| 性能 | 更快,一次调用 | 可能需要多次调用 |
| Schema 复杂度 | 支持复杂 Schema | 同样支持 |
| 错误处理 | 提供商处理 | LangChain 自动重试 |
| 推荐场景 | OpenAI/Anthropic 等 | 国产模型/开源模型 |
十、实战代码:完整示例
10.1 基础示例:古诗生成
from langchain_openai import ChatOpenAIfrom langchain.agents import create_agentfrom pydantic import BaseModel, Fieldfrom langchain.agents.structured_output import ToolStrategyimport dotenvdotenv.load_dotenv()chanAI = ChatOpenAI( model="qwen3.7-plus", temperature=0.7, extra_body={"enable_thinking": False})class PoemStrategy(BaseModel): name: str = Field(description="古诗名称") content: str = Field(description="古诗内容")agentChat = create_agent( model=chanAI, response_format=PoemStrategy,)result = agentChat.invoke({ "messages": [ {"role": "system", "content": "你是一个古诗创作助手"}, {"role": "user", "content": "今天长沙的天气如何"} ]})print(result["structured_response"])# PoemStrategy(name='长沙今日即景', content='湘水悠悠绕古城...')10.2 进阶示例:带工具的天气查询
from langchain.tools import toolfrom langchain.agents.structured_output import ToolStrategyclass WeatherStrategy(BaseModel): city: str = Field(description="城市名称") weather: str = Field(description="天气描述") temperature: str = Field(description="温度") activity: str = Field(description="建议活动")@tool(description="查询城市天气的工具")def get_weather(city: str): return f"今天{city}的天气晴,温度为30度,适合户外活动"agentChat = create_agent( model=chanAI, tools=[get_weather], response_format=ToolStrategy( schema=WeatherStrategy, tool_message_content="天气查询已完成", handle_errors=True ),)result = agentChat.invoke({ "messages": [{"role": "user", "content": "长沙今天是什么天气"}]})print(result["structured_response"])# WeatherStrategy(city='长沙', weather='晴', temperature='30度', activity='适合户外活动')学AI大模型的正确顺序,千万不要搞错了
🤔2026年AI风口已来!各行各业的AI渗透肉眼可见,超多公司要么转型做AI相关产品,要么高薪挖AI技术人才,机遇直接摆在眼前!
有往AI方向发展,或者本身有后端编程基础的朋友,直接冲AI大模型应用开发转岗超合适!
就算暂时不打算转岗,了解大模型、RAG、Prompt、Agent这些热门概念,能上手做简单项目,也绝对是求职加分王🔋
📝给大家整理了超全最新的AI大模型应用开发学习清单和资料,手把手帮你快速入门!👇👇
学习路线:
✅大模型基础认知—大模型核心原理、发展历程、主流模型(GPT、文心一言等)特点解析
✅核心技术模块—RAG检索增强生成、Prompt工程实战、Agent智能体开发逻辑
✅开发基础能力—Python进阶、API接口调用、大模型开发框架(LangChain等)实操
✅应用场景开发—智能问答系统、企业知识库、AIGC内容生成工具、行业定制化大模型应用
✅项目落地流程—需求拆解、技术选型、模型调优、测试上线、运维迭代
✅面试求职冲刺—岗位JD解析、简历AI项目包装、高频面试题汇总、模拟面经
以上6大模块,看似清晰好上手,实则每个部分都有扎实的核心内容需要吃透!
我把大模型的学习全流程已经整理📚好了!抓住AI时代风口,轻松解锁职业新可能,希望大家都能把握机遇,实现薪资/职业跃迁~