LangChain结构化输出(Structured output)详解

在自然语言处理任务中,模型返回的非结构化文本往往需要额外解析才能被应用程序直接使用,而 LangChain 的结构化输出功能则解决了这一痛点。它允许智能体(Agent)以特定、可预测的格式返回数据,无需手动解析自然语言,直接生成 JSON 对象、Pydantic 模型或数据类(dataclasses)等结构化数据,极大提升了开发效率与数据可用性。

LangChain 的 create_agent 函数会自动处理结构化输出流程:用户只需定义期望的结构化输出 schema(格式模板),模型生成结构化数据后,会被自动捕获、验证,并最终存储在智能体状态的 structured_response 键中,供开发者直接调用。其核心定义如下:

1
2
3
4
5
6
7
def create_agent(
...
response_format: Union[
ToolStrategy[StructuredResponseT],
ProviderStrategy[StructuredResponseT],
type[StructuredResponseT],
]

一、响应格式(Response Format)

response_format 参数是控制智能体返回结构化数据的核心,它支持四种取值,分别对应不同的结构化输出策略,开发者可根据模型能力与需求选择:

取值类型 说明 适用场景
ToolStrategy[StructuredResponseT] 基于工具调用(Tool Calling)实现结构化输出 不支持原生结构化输出的模型(大部分现代模型均支持工具调用,因此适用性广)
ProviderStrategy[StructuredResponseT] 基于模型提供商原生能力实现结构化输出 支持原生结构化输出的模型(目前仅 OpenAI、Grok)
type[StructuredResponseT] 直接传入 schema 类型(如 Pydantic 模型) 希望 LangChain 自动选择最优策略的场景(优先原生能力,无则 fallback 到工具调用)
None 不启用结构化输出 仅需自然语言响应的普通任务

当直接传入 schema 类型时,LangChain 会自动判断模型能力:

  • 若模型支持原生结构化输出(如 OpenAI GPT-4/5、Grok),则自动使用 ProviderStrategy
  • 若模型不支持原生能力,则自动切换为 ToolStrategy

最终的结构化结果统一存储在智能体最终状态的 structured_response 键中,调用方式如下:

1
2
result = agent.invoke(...)
structured_data = result["structured_response"] # 直接获取结构化数据

二、ProviderStrategy

部分模型提供商(目前仅 OpenAI、Grok)通过 API 原生支持结构化输出,这是可靠性最高的方式——由提供商直接强制 schema 约束,避免额外解析误差。

1. 核心定义

ProviderStrategy 是基于原生能力的策略类,仅需传入 schema 即可初始化:

1
2
class ProviderStrategy(Generic[SchemaT]):
schema: type[SchemaT] # 必选参数,定义结构化输出格式

2. 支持的 Schema 类型

schema 参数支持四种格式,覆盖主流结构化需求:

  • Pydantic 模型:继承 BaseModel 的子类,支持字段校验(如数值范围、必填项);
  • Python 数据类(dataclasses):带类型注解的 @dataclass 装饰类;
  • TypedDict:带类型约束的字典类;
  • JSON Schema:符合 JSON 规范的字典(如 {"type": "object", "properties": {...}})。

3. 使用示例(Pydantic 模型)

以提取联系人信息为例,通过 Pydantic 定义 schema 后,直接传入 create_agentresponse_format,LangChain 会自动使用 ProviderStrategy

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
from pydantic import BaseModel, Field
from langchain.agents import create_agent

# 1. 定义 Pydantic 模型(schema)
class ContactInfo(BaseModel):
"""联系人信息结构化模型"""
name: str = Field(description="联系人姓名")
email: str = Field(description="联系人邮箱地址")
phone: str = Field(description="联系人电话号码")

# 2. 创建智能体(自动使用 ProviderStrategy)
agent = create_agent(
model="openai:gpt-5", # 支持原生结构化输出的模型
tools=tools, # 按需传入工具(无工具可传空列表)
response_format=ContactInfo # 直接传入 schema 类型
)

# 3. 调用智能体并获取结构化结果
result = agent.invoke({
"messages": [{"role": "user", "content": "提取信息:John Doe,邮箱 john@example.com,电话 (555) 123-4567"}]
})

# 4. 输出结构化数据
print(result["structured_response"])
# 输出:ContactInfo(name='John Doe', email='john@example.com', phone='(555) 123-4567')

4. 注意事项

若模型支持原生结构化输出,以下两种写法功能完全等价:

1
2
3
4
5
# 写法1:直接传入 schema(推荐,简洁)
response_format=ContactInfo

# 写法2:显式指定 ProviderStrategy
response_format=ProviderStrategy(schema=ContactInfo)

若模型不支持原生能力,即使显式指定 ProviderStrategy,LangChain 也会自动 fallback 到 ToolStrategy

三、工具调用策略(ToolStrategy)

对于不支持原生结构化输出的模型,LangChain 通过工具调用实现等效功能——将结构化输出封装为一个“工具”,模型通过调用该工具返回符合 schema 的数据。该策略适用于所有支持工具调用的现代模型(如 Claude、Gemini 等)。

1. 核心定义

ToolStrategy 是基于工具调用的策略类,除了必选的 schema,还支持自定义工具消息与错误处理:

1
2
3
4
5
6
7
8
9
10
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],
]

2. 关键参数详解

(1)schema

ProviderStrategyschema 完全一致,支持 Pydantic 模型、dataclasses、TypedDict、JSON Schema 四种类型,且额外支持联合类型(Union)——允许模型根据上下文从多个 schema 中选择最合适的一种。

(2)tool_message_content

自定义工具调用成功后的反馈消息,会显示在对话历史中。若不设置,默认消息为“Returning structured response: {结构化数据}”。

(3)handle_errors

控制结构化输出验证失败时的处理逻辑,默认值为 True(捕获所有错误并使用默认模板重试),支持五种取值:

  • True:捕获所有错误,使用默认错误模板提示模型重试;
  • str:捕获所有错误,使用自定义字符串提示模型重试;
  • type[Exception]:仅捕获指定类型的错误(如 ValueError),其他错误直接抛出;
  • tuple[type[Exception], ...]:捕获多个指定类型的错误(如 (ValueError, TypeError));
  • Callable[[Exception], str]:自定义错误处理函数,根据错误类型返回个性化提示;
  • False:不处理错误,所有验证失败直接抛出异常。

3. 使用示例(Pydantic 模型)

以分析产品评论为例,通过 ToolStrategy 定义结构化输出,并指定错误处理逻辑:

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
from pydantic import BaseModel, Field
from typing import Literal
from langchain.agents import create_agent
from langchain.agents.structured_output import ToolStrategy

# 1. 定义 Pydantic 模型(带字段校验)
class ProductReview(BaseModel):
"""产品评论结构化分析模型"""
rating: int | None = Field(description="产品评分", ge=1, le=5) # 评分范围 1-5
sentiment: Literal["positive", "negative"] = Field(description="评论情感倾向") # 仅支持两种值
key_points: list[str] = Field(description="评论关键点,小写,每个 1-3 个词")

# 2. 创建智能体(显式指定 ToolStrategy)
agent = create_agent(
model="xxx:xxx", # 不支持原生结构化输出的模型
tools=tools,
response_format=ToolStrategy(
schema=ProductReview,
handle_errors=True # 捕获所有错误并默认重试
)
)

# 3. 调用智能体
result = agent.invoke({
"messages": [{"role": "user", "content": "分析评论:'很棒的产品,5星!物流快,但价格贵'"}]
})

# 4. 获取结果
print(result["structured_response"])
# 输出:ProductReview(rating=5, sentiment='positive', key_points=['fast shipping', 'expensive'])

4. 自定义工具消息示例

通过 tool_message_content 自定义工具调用后的反馈,提升对话体验:

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
from pydantic import BaseModel, Field
from typing import Literal

# 1. 定义“会议任务”schema
class MeetingAction(BaseModel):
"""会议记录中的任务项"""
task: str = Field(description="具体任务内容")
assignee: str = Field(description="任务负责人")
priority: Literal["low", "medium", "high"] = Field(description="任务优先级")

# 2. 创建智能体(自定义工具消息)
agent = create_agent(
model="openai:gpt-5",
tools=[], # 无额外工具
response_format=ToolStrategy(
schema=MeetingAction,
tool_message_content="任务项已捕获并添加到会议纪要!" # 自定义反馈
)
)

# 3. 调用后,对话历史中的 Tool Message 会显示自定义内容
agent.invoke({
"messages": [{"role": "user", "content": "会议记录:Sarah 需要尽快更新项目时间线"}]
})

此时对话历史中的工具消息为:

================================ Tool Message =================================
Name: MeetingAction
任务项已捕获并添加到会议纪要!

若不设置 tool_message_content,默认消息为:

================================ Tool Message =================================
Name: MeetingAction
Returning structured response: {'task': '更新项目时间线', 'assignee': 'Sarah', 'priority': 'high'}

四、错误处理(Error Handling)

模型生成结构化数据时可能出现错误(如格式不匹配、多输出冲突),LangChain 提供智能重试机制,通过 handle_errors 参数灵活控制错误处理逻辑。以下是常见错误类型及处理方式。

1. 常见错误类型

(1)多结构化输出错误(Multiple Structured Outputs Error)

当模型错误地调用多个结构化输出工具(而实际仅需一个)时,智能体会生成错误提示并要求模型重试。

示例:定义“联系人”与“事件”两个 schema 的联合类型,但模型同时返回两种数据:

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
from pydantic import BaseModel, Field
from typing import Union

# 定义两个 schema
class ContactInfo(BaseModel):
name: str = Field(description="姓名")
email: str = Field(description="邮箱")

class EventDetails(BaseModel):
event_name: str = Field(description="事件名称")
date: str = Field(description="事件日期")

# 创建智能体(支持二选一 schema)
agent = create_agent(
model="openai:gpt-5",
tools=[],
response_format=ToolStrategy(Union[ContactInfo, EventDetails]) # 二选一
)

# 调用:模型错误返回两种数据
result = agent.invoke({
"messages": [{"role": "user", "content": "提取信息:John Doe(john@email.com)组织 3月15日的 Tech Conference"}]
})

此时智能体返回错误提示:

================================ Tool Message =================================
Name: ContactInfo
Error: Model incorrectly returned multiple structured responses (ContactInfo, EventDetails) when only one is expected. Please fix your mistakes.

模型会根据提示重试,最终返回单一正确的结构化数据。

(2)Schema 验证错误(Schema Validation Error)

当结构化输出不符合 schema 约束(如数值超出范围、字段缺失)时,智能体会返回具体错误信息(如“评分需在 1-5 之间”),引导模型修正。

示例:评分字段要求 1-5,但模型返回 10:

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
from pydantic import BaseModel, Field

# 定义带校验的 schema
class ProductRating(BaseModel):
rating: int | None = Field(description="评分(1-5)", ge=1, le=5)
comment: str = Field(description="评论内容")

# 创建智能体
agent = create_agent(
model="openai:gpt-5",
tools=[],
response_format=ToolStrategy(ProductRating),
system_prompt="解析产品评论,不虚构字段或值"
)

# 调用:评论提到“10/10”,模型错误返回 rating=10
result = agent.invoke({
"messages": [{"role": "user", "content": "解析评论:'超棒的产品,10/10!'"}]
})

智能体返回验证错误:

================================ Tool Message =================================
Name: ProductRating
Error: Failed to parse structured output for tool 'ProductRating': 1 validation error for ProductRating.rating
  Input should be less than or equal to 5 [type=less_than_equal, input_value=10, input_type=int]. Please fix your mistakes.

模型重试后会将 rating 修正为 5,符合 schema 约束。

2. 自定义错误处理策略

通过 handle_errors 参数可灵活配置错误处理逻辑,以下是常见场景示例:

(1)自定义错误提示

直接传入字符串,所有错误均使用该提示引导重试:

1
2
3
4
ToolStrategy(
schema=ProductRating,
handle_errors="请提供 1-5 之间的有效评分,并包含评论内容。"
)

错误发生时,工具消息为:

================================ Tool Message =================================
Name: ProductRating
请提供 1-5 之间的有效评分,并包含评论内容。

(2)仅处理特定错误

传入单个或多个异常类型,仅捕获指定错误,其他错误直接抛出:

1
2
3
4
5
6
7
8
9
10
11
# 仅处理 ValueError
ToolStrategy(
schema=ProductRating,
handle_errors=ValueError
)

# 处理 ValueError 和 TypeError
ToolStrategy(
schema=ProductRating,
handle_errors=(ValueError, TypeError)
)

(3)自定义错误处理函数

传入函数,根据错误类型返回个性化提示:

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
from langchain.agents.structured_output import StructuredOutputValidationError, MultipleStructuredOutputsError

# 定义错误处理函数
def custom_error_handler(error: Exception) -> str:
if isinstance(error, StructuredOutputValidationError):
return "格式错误,请按要求重新生成。"
elif isinstance(error, MultipleStructuredOutputsError):
return "仅需返回一个结构化结果,请选择最相关的类型。"
else:
return f"未知错误:{str(error)}"

# 应用到 ToolStrategy
ToolStrategy(
schema=Union[ContactInfo, EventDetails],
handle_errors=custom_error_handler
)

(4)禁用错误处理

设置 handle_errors=False,所有错误直接抛出,不重试:

1
2
3
4
ToolStrategy(
schema=ProductRating,
handle_errors=False
)

五、核心组件总结

LangChain 结构化输出的核心是“schema 定义 + 策略选择 + 错误处理”,三者协同实现高效、可靠的结构化数据生成:

  1. Schema:定义结构化输出的“模板”,支持 Pydantic、dataclasses 等多种格式,带校验能力;
  2. 策略(Strategy):根据模型能力选择原生(Provider)或工具调用(Tool)方式,平衡可靠性与兼容性;
  3. 错误处理:通过智能重试与自定义提示,降低模型生成错误的影响,提升结构化数据准确性。

通过上述组件,开发者可轻松将模型输出转换为应用程序可直接使用的结构化数据,无需手动解析,大幅简化自然语言任务的工程落地流程。