结构化输出允许智能体以特定、可预测的格式返回数据。无需解析自然语言响应,您将直接获得结构化数据(如 JSON 对象、Pydantic 模型或数据类),供应用程序直接使用。
LangChain 预构建的 ReAct 智能体 create_agent() 自动处理结构化输出。用户设置所需的结构化输出模式后,当模型生成结构化数据时,系统将捕获、验证并将其返回在智能体状态的 'structured_response' 键中。
def create_agent(
...
response_format: Union[
ToolStrategy[StructuredResponseT],
ProviderStrategy[StructuredResponseT],
type[StructuredResponseT],
]
控制智能体如何返回结构化数据:
ToolStrategy[StructuredResponseT]:使用工具调用实现结构化输出
ProviderStrategy[StructuredResponseT]:使用提供商原生结构化输出功能
type[StructuredResponseT]:模式类型——根据模型能力自动选择最佳策略
None:无结构化输出
当直接提供模式类型时,LangChain 会自动选择:
- 对支持原生结构化输出的模型(如 OpenAI、Grok)使用
ProviderStrategy
- 对其他所有模型使用
ToolStrategy
结构化响应将返回在智能体最终状态的 structured_response 键中。
提供商策略
某些模型提供商通过其 API 原生支持结构化输出(目前仅限 OpenAI 和 Grok)。当可用时,这是最可靠的方法。
要使用此策略,请配置 ProviderStrategy:
class ProviderStrategy(Generic[SchemaT]):
schema: type[SchemaT]
定义结构化输出格式的模式。支持:
- Pydantic 模型:带有字段验证的
BaseModel 子类
- 数据类:带有类型注解的 Python 数据类
- TypedDict:类型化字典类
- JSON 模式:包含 JSON 模式规范的字典
当您直接向 create_agent.response_format 传递模式类型且模型支持原生结构化输出时,LangChain 会自动使用 ProviderStrategy:
from pydantic import BaseModel
from langchain.agents import create_agent
class ContactInfo(BaseModel):
"""人员的联系信息。"""
name: str = Field(description="人员姓名")
email: str = Field(description="人员邮箱地址")
phone: str = Field(description="人员电话号码")
agent = create_agent(
model="openai:gpt-5",
tools=tools,
response_format=ContactInfo # 自动选择 ProviderStrategy
)
result = agent.invoke({
"messages": [{"role": "user", "content": "从以下内容提取联系信息:John Doe, [email protected], (555) 123-4567"}]
})
result["structured_response"]
# ContactInfo(name='John Doe', email='[email protected]', phone='(555) 123-4567')
提供商原生结构化输出提供高可靠性和严格验证,因为模型提供商强制执行模式。在可用时请优先使用。
如果提供商对您选择的模型原生支持结构化输出,则写 response_format=ProductReview 与 response_format=ToolStrategy(ProductReview) 功能等效。无论哪种情况,如果结构化输出不被支持,智能体将回退到工具调用策略。
工具调用策略
对于不支持原生结构化输出的模型,LangChain 使用工具调用实现相同效果。此方法适用于所有支持工具调用的模型(即大多数现代模型)。
要使用此策略,请配置 ToolStrategy:
class ToolStrategy(Generic[SchemaT]):
schema: type[SchemaT]
tool_message_content: str | None
handle_errors: Union[
bool,
str,
type[Exception],
tuple[type[Exception], ...],
Callable[[Exception], str],
]
定义结构化输出格式的模式。支持:
- Pydantic 模型:带有字段验证的
BaseModel 子类
- 数据类:带有类型注解的 Python 数据类
- TypedDict:类型化字典类
- JSON 模式:包含 JSON 模式规范的字典
- 联合类型:多个模式选项。模型将根据上下文选择最合适的模式。
生成结构化输出时返回的工具消息的自定义内容。
若未提供,默认显示结构化响应数据的消息。
结构化输出验证失败的错误处理策略。默认为 True。
True:捕获所有错误并使用默认错误模板
str:捕获所有错误并使用此自定义消息
type[Exception]:仅捕获此异常类型并使用默认消息
tuple[type[Exception], ...]:仅捕获这些异常类型并使用默认消息
Callable[[Exception], str]:自定义函数返回错误消息
False:不重试,让异常传播
from pydantic import BaseModel, Field
from typing import Literal, Optional
from langchain.agents import create_agent
from langchain.agents.structured_output import ToolStrategy
class ProductReview(BaseModel):
"""产品评论分析。"""
rating: Optional[int] = Field(description="产品评分", ge=1, le=5)
sentiment: Literal["positive", "negative"] = Field(description="评论情感倾向")
key_points: list[str] = Field(description="评论要点。小写,每项 1-3 个词。")
agent = create_agent(
model="openai:gpt-5",
tools=tools,
response_format=ToolStrategy(ProductReview)
)
result = agent.invoke({
"messages": [{"role": "user", "content": "分析此评论:‘优秀产品:5 星满分。发货快,但价格贵’"}]
})
result["structured_response"]
# ProductReview(rating=5, sentiment='positive', key_points=['发货快', '价格贵'])
自定义工具消息内容
tool_message_content 参数允许您自定义生成结构化输出时出现在对话历史中的消息:
from pydantic import BaseModel, Field
from typing import Literal
from langchain.agents import create_agent
from langchain.agents.structured_output import ToolStrategy
class MeetingAction(BaseModel):
"""从会议记录中提取的行动项。"""
task: str = Field(description="需完成的具体任务")
assignee: str = Field(description="任务负责人")
priority: Literal["low", "medium", "high"] = Field(description="优先级")
agent = create_agent(
model="openai:gpt-5",
tools=[],
response_format=ToolStrategy(
schema=MeetingAction,
tool_message_content="行动项已捕获并添加至会议记录!"
)
)
agent.invoke({
"messages": [{"role": "user", "content": "根据我们的会议:Sarah 需尽快更新项目时间表"}]
})
================================ 人类消息 =================================
根据我们的会议:Sarah 需尽快更新项目时间表
================================== AI 消息 ==================================
工具调用:
MeetingAction (call_1)
调用 ID: call_1
参数:
task: 更新项目时间表
assignee: Sarah
priority: high
================================= 工具消息 =================================
名称: MeetingAction
行动项已捕获并添加至会议记录!
若无 tool_message_content,最终的 ToolMessage 将为:
================================= 工具消息 =================================
名称: MeetingAction
返回结构化响应:{'task': '更新项目时间表', 'assignee': 'Sarah', 'priority': 'high'}
错误处理
模型通过工具调用生成结构化输出时可能会出错。LangChain 提供智能重试机制自动处理这些错误。
多重结构化输出错误
当模型错误地调用多个结构化输出工具时,智能体会在 ToolMessage 中提供错误反馈并提示模型重试:
from pydantic import BaseModel, Field
from typing import Union
from langchain.agents import create_agent
from langchain.agents.structured_output import ToolStrategy
class ContactInfo(BaseModel):
name: str = Field(description="人员姓名")
email: str = Field(description="邮箱地址")
class EventDetails(BaseModel):
event_name: str = Field(description="活动名称")
date: str = Field(description="活动日期")
agent = create_agent(
model="openai:gpt-5",
tools=[],
response_format=ToolStrategy(Union[ContactInfo, EventDetails]) # 默认:handle_errors=True
)
agent.invoke({
"messages": [{"role": "user", "content": "提取信息:John Doe ([email protected]) 正在组织 3 月 15 日的技术会议"}]
})
================================ 人类消息 =================================
提取信息:John Doe ([email protected]) 正在组织 3 月 15 日的技术会议
None
================================== AI 消息 ==================================
工具调用:
ContactInfo (call_1)
调用 ID: call_1
参数:
name: John Doe
email: [email protected]
EventDetails (call_2)
调用 ID: call_2
参数:
event_name: 技术会议
date: March 15th
================================= 工具消息 =================================
名称: ContactInfo
错误:模型错误地返回了多个结构化响应(ContactInfo, EventDetails),但仅期望一个。
请修正错误。
================================= 工具消息 =================================
名称: EventDetails
错误:模型错误地返回了多个结构化响应(ContactInfo, EventDetails),但仅期望一个。
请修正错误。
================================== AI 消息 ==================================
工具调用:
ContactInfo (call_3)
调用 ID: call_3
参数:
name: John Doe
email: [email protected]
================================= 工具消息 =================================
名称: ContactInfo
返回结构化响应:{'name': 'John Doe', 'email': '[email protected]'}
模式验证错误
当结构化输出与预期模式不匹配时,智能体会提供具体错误反馈:
from pydantic import BaseModel, Field
from typing import Optional
from langchain.agents import create_agent
from langchain.agents.structured_output import ToolStrategy
class ProductRating(BaseModel):
rating: Optional[int] = Field(description="评分 1-5", ge=1, le=5)
comment: str = Field(description="评论内容")
agent = create_agent(
model="openai:gpt-5",
tools=[],
response_format=ToolStrategy(ProductRating), # 默认:handle_errors=True
prompt="您是一个解析产品评论的助手。请勿虚构任何字段或值。"
)
agent.invoke({
"messages": [{"role": "user", "content": "解析此内容:优秀产品,10/10!"}]
})
================================ 人类消息 =================================
解析此内容:优秀产品,10/10!
================================== AI 消息 ==================================
工具调用:
ProductRating (call_1)
调用 ID: call_1
参数:
rating: 10
comment: 优秀产品
================================= 工具消息 =================================
名称: ProductRating
错误:解析工具 'ProductRating' 的结构化输出失败:1 个验证错误:ProductRating.rating
输入值应小于或等于 5 [类型=less_than_equal, 输入值=10, 输入类型=int]。
请修正错误。
================================== AI 消息 ==================================
工具调用:
ProductRating (call_2)
调用 ID: call_2
参数:
rating: 5
comment: 优秀产品
================================= 工具消息 =================================
名称: ProductRating
返回结构化响应:{'rating': 5, 'comment': '优秀产品'}
错误处理策略
您可以使用 handle_errors 参数自定义错误处理方式:
自定义错误消息:
ToolStrategy(
schema=ProductRating,
handle_errors="请提供 1-5 之间的有效评分并包含评论内容。"
)
若 handle_errors 为字符串,智能体将始终提示模型使用固定工具消息重试:
================================= 工具消息 =================================
名称: ProductRating
请提供 1-5 之间的有效评分并包含评论内容。
仅处理特定异常:
ToolStrategy(
schema=ProductRating,
handle_errors=ValueError # 仅在 ValueError 时重试,其他异常抛出
)
若 handle_errors 为异常类型,智能体仅在抛出指定类型的异常时重试(使用默认错误消息)。其他情况下,异常将被抛出。
处理多个异常类型:
ToolStrategy(
schema=ProductRating,
handle_errors=(ValueError, TypeError) # 在 ValueError 和 TypeError 时重试
)
若 handle_errors 为异常元组,智能体仅在抛出指定类型之一的异常时重试(使用默认错误消息)。其他情况下,异常将被抛出。
自定义错误处理函数:
def custom_error_handler(error: Exception) -> str:
if isinstance(error, StructuredOutputValidationError):
return "格式存在问题。请重试。"
elif isinstance(error, MultipleStructuredOutputsError):
return "返回了多个结构化输出。请选择最相关的一个。"
else:
return f"错误:{str(error)}"
ToolStrategy(
schema=ToolStrategy(Union[ContactInfo, EventDetails]),
handle_errors=custom_error_handler
)
在 StructuredOutputValidationError 时:
================================= 工具消息 =================================
名称: ToolStrategy
格式存在问题。请重试。
在 MultipleStructuredOutputsError 时:
================================= 工具消息 =================================
名称: ToolStrategy
返回了多个结构化输出。请选择最相关的一个。
在其他错误时:
================================= 工具消息 =================================
名称: ToolStrategy
错误:<错误消息>
无错误处理:
response_format = ToolStrategy(
schema=ProductRating,
handle_errors=False # 所有错误均抛出
)