Appearance
结构化输出
本节解决什么问题
在实际应用中,我们通常不只是需要 LLM 返回一段自然语言文本--我们需要结构化数据:提取的实体、分类标签、评分数值等。LangChain 的结构化输出(Structured Output)功能允许你定义一个 Schema,让模型的输出自动解析为符合该 Schema 的 Python 对象,而不是原始字符串。
必须深刻理解,不能跳过:结构化输出是"生成时约束",不是"后处理校验"
前端的 Zod / Joi 是后处理校验:先拿到数据,再验证是否符合 Schema,不符合则报错。LangChain 的结构化输出是生成时约束:在模型推理阶段就引导输出格式,模型"知道"必须按 Schema 生成。这意味着输出几乎总是符合 Schema,但仍可能在极端情况下出错(模型幻觉、格式错误),因此需要错误处理机制(
handle_errors、include_raw)。
前端类比
结构化输出类似于前端中的 TypeScript 接口 + Zod 验证。你定义一个 interface 描述数据结构,Zod 在运行时校验数据是否合规。LangChain 的结构化输出做的是同一件事:用 Pydantic Model 定义"接口",框架自动确保 LLM 输出符合该"接口"。
不过要注意,LLM 的结构化输出是生成时约束,而非后处理校验--它在模型推理阶段就引导输出格式,这比前端的"先拿到数据再校验"更高效。
原生语义
Zod 校验的是已经存在的数据对象,通过则放行、不通过则抛错。LangChain 结构化输出约束的是模型的生成过程:ProviderStrategy 通过模型的 JSON Schema / JSON mode 原生能力在生成时约束;ToolStrategy 通过"虚拟工具调用"机制让模型按工具参数 Schema 生成。两者都不是"生成完再校验",而是"引导生成符合 Schema"。
🔗 LangChain Structured Output 官方文档
为什么需要结构化输出?
- 类型安全 - 直接获得 Python 对象(Pydantic Model / TypedDict),无需手动解析 JSON
- 减少幻觉 - Schema 约束模型输出范围,降低生成无关内容的概率
- 管线友好 - 输出可直接传入下游函数或写入数据库,无需 JSON.parse
- 降低成本 - 不需要额外的 LLM 调用来"修正"输出格式
核心概念
两种策略
LangChain 提供两种结构化输出策略:
| 策略 | 原理 | 适用场景 |
|---|---|---|
| ProviderStrategy | 利用模型提供商的原生结构化输出能力(JSON Schema / JSON mode) | 模型支持原生结构化输出时,延迟更低 |
| ToolStrategy | 创建一个"虚拟工具",利用工具调用机制约束输出格式 | 任何支持工具调用的模型均可使用,兼容性更广 |
选择逻辑:当你直接将 Schema 类型传给 response_format 时,create_agent 会根据模型能力自动选择策略;如果你需要显式指定,可以用 ToolStrategy(Schema) 或 ProviderStrategy(Schema) 包装。
导入
python
from langchain.agents.structured_output import ToolStrategy, ProviderStrategy支持的 Schema 类型
LangChain 支持四种 Schema 定义方式,按推荐程度排列:
Pydantic BaseModel ▸ 最推荐,支持验证 + 字段描述 + 嵌套模型
Python dataclass ▸ 轻量替代,适合简单结构
TypedDict ▸ 最简洁,返回纯字典,无运行时验证
JSON Schema dict ▸ 最大灵活性,适合动态 Schema 场景结构化输出数据流
代码示例 1: Pydantic Model + ProviderStrategy(自动)
最推荐的使用方式。定义 Pydantic BaseModel,直接传给 response_format:
python
import os
from pydantic import BaseModel, Field
from langchain.agents import create_agent
from langchain.chat_models import init_chat_model
class ContactInfo(BaseModel):
"""联系人信息"""
name: str = Field(description="姓名")
email: str = Field(description="电子邮箱地址")
phone: str = Field(description="电话号码")
company: str | None = Field(default=None, description="所在公司(可选)")
model = init_chat_model(os.environ["LLM_MODEL"])
# 直接传 Pydantic Model -> 自动使用 ProviderStrategy
agent = create_agent(
model=model,
response_format=ContactInfo,
)
result = agent.invoke({
"messages": [{
"role": "user",
"content": "提取联系信息:张三,邮箱 zhangsan@example.com,手机 138-0000-1234,就职于字节跳动"
}]
})
contact = result["structured_response"]
print(contact) # ContactInfo(name='张三', email='zhangsan@example.com', ...)
print(contact.name) # 张三要点:Field(description=...) 提高提取准确率;str | None + default=None 表示可选字段;通过 result["structured_response"] 访问 Pydantic 实例。
代码示例 2: ProviderStrategy 显式指定 + strict 模式
需要显式控制策略时,用 ProviderStrategy 包装。strict=True(langchain>=1.2)强制模型严格遵循 Schema,不允许额外字段:
python
import os
from pydantic import BaseModel, Field
from langchain.agents import create_agent
from langchain.chat_models import init_chat_model
from langchain.agents.structured_output import ProviderStrategy
class SensorReading(BaseModel):
"""传感器读数"""
sensor_id: str = Field(description="传感器编号")
value: float = Field(description="读数值")
unit: str = Field(description="单位")
timestamp: str = Field(description="时间戳,ISO 8601 格式")
model = init_chat_model(os.environ["LLM_MODEL"])
# ProviderStrategy + strict=True:强制严格遵循 Schema(langchain>=1.2)
agent = create_agent(
model=model,
response_format=ProviderStrategy(SensorReading, strict=True),
)
result = agent.invoke({
"messages": [{"role": "user", "content": "传感器 A-12 当前读数 42.5 ppm,时间 2026-07-15T10:30:00Z"}]
})
reading = result["structured_response"]
print(reading.sensor_id) # A-12
print(reading.value) # 42.5strict 模式:普通模式下模型可能返回 Schema 中未定义的额外字段;strict=True 禁止额外字段,确保输出严格匹配 Schema 定义。适用于对数据格式有严格要求的场景(如写入数据库、API 交互)。
代码示例 3: ToolStrategy + 错误处理
当模型不支持原生结构化输出,或你想确保使用工具调用方式时,显式使用 ToolStrategy。handle_errors 参数控制解析失败时的行为:
python
import os
from pydantic import BaseModel, Field
from typing import Literal
from langchain.agents import create_agent
from langchain.chat_models import init_chat_model
from langchain.agents.structured_output import ToolStrategy
class ProductReview(BaseModel):
"""商品评价分析结果"""
rating: int = Field(description="评分(1-5)", ge=1, le=5)
sentiment: Literal["positive", "negative", "neutral"] = Field(description="情感倾向")
key_points: list[str] = Field(description="关键要点,每项 1-3 个词")
def search_reviews(product_name: str) -> str:
"""搜索商品评价"""
return f"找到 {product_name} 的评价:质量很好,物流很快,但价格偏贵。4.5星"
model = init_chat_model(os.environ["LLM_MODEL"])
# handle_errors=True:解析失败时返回错误信息而非抛异常
agent = create_agent(
model=model,
tools=[search_reviews],
response_format=ToolStrategy(ProductReview, handle_errors=True),
)
result = agent.invoke({
"messages": [{"role": "user", "content": "分析:'质量不错,5星好评!发货速度快,但包装简陋'"}]
})
review = result["structured_response"]
print(review.rating) # 5
print(review.sentiment) # positive
print(review.key_points) # ['质量不错', '发货快', '包装简陋']handle_errors 选项详解
handle_errors 控制 Schema 解析失败时的行为:
| 值 | 行为 |
|---|---|
True | 解析失败时返回错误信息字符串,不抛异常 |
False(默认) | 解析失败时抛出 ValidationError |
"exception" | 解析失败时抛出异常(与 False 类似,但更明确) |
callable | 传入自定义处理函数,接收错误信息并返回自定义结果 |
python
# 自定义错误处理函数
def on_parse_error(error: str) -> str:
"""解析失败时记录日志并返回默认值"""
print(f"[警告] 结构化输出解析失败: {error}")
return f"解析失败: {error}"
agent = create_agent(
model=model,
tools=[search_reviews],
response_format=ToolStrategy(ProductReview, handle_errors=on_parse_error),
)tool_message_content 选项
tool_message_content 控制虚拟工具调用返回给 Agent 的 ToolMessage 内容。默认情况下,工具结果就是解析后的结构化数据。你可以自定义此内容:
python
agent = create_agent(
model=model,
tools=[search_reviews],
response_format=ToolStrategy(
ProductReview,
handle_errors=True,
tool_message_content="评价分析已完成", # 自定义工具消息内容
),
)适用场景:当 Agent 在后续推理中需要引用结构化输出结果时,自定义 tool_message_content 可以提供更友好的上下文提示。
策略选择:ProviderStrategy 延迟更低(无额外工具调用开销),但需模型原生支持;ToolStrategy 兼容性更广,适合作为兜底方案。ToolStrategy 还支持 handle_errors 和 tool_message_content 等精细控制。
代码示例 4: Union 类型支持
ToolStrategy 支持 Union 类型,允许模型在多种 Schema 中选择最匹配的一种返回:
python
import os
from pydantic import BaseModel, Field
from typing import Union
from langchain.agents import create_agent
from langchain.chat_models import init_chat_model
from langchain.agents.structured_output import ToolStrategy
class TextSummary(BaseModel):
"""文本摘要结果"""
summary: str = Field(description="摘要内容")
word_count: int = Field(description="原文字数")
class DataExtraction(BaseModel):
"""数据提取结果"""
entities: list[str] = Field(description="提取的实体列表")
count: int = Field(description="实体数量")
model = init_chat_model(os.environ["LLM_MODEL"])
# Union 类型:模型根据输入内容自动选择返回 TextSummary 或 DataExtraction
agent = create_agent(
model=model,
response_format=ToolStrategy(Union[TextSummary, DataExtraction]),
)
# 摘要类输入 -> 返回 TextSummary
result1 = agent.invoke({
"messages": [{"role": "user", "content": "总结:Python 是一种广泛使用的编程语言..."}]
})
print(type(result1["structured_response"]).__name__) # TextSummary
# 提取类输入 -> 返回 DataExtraction
result2 = agent.invoke({
"messages": [{"role": "user", "content": "提取实体:张三在字节跳动担任前端工程师"}]
})
print(type(result2["structured_response"]).__name__) # DataExtraction适用场景:同一 Agent 需要根据输入内容返回不同结构的输出(如既能做摘要又能做实体提取)。
代码示例 5: 使用 TypedDict
不需要 Pydantic 运行时验证时,可以用更轻量的 TypedDict:
python
import os
from typing_extensions import TypedDict, Annotated
from langchain.agents import create_agent
from langchain.chat_models import init_chat_model
class AnalysisResult(TypedDict):
"""数据分析结果"""
total_count: Annotated[int, ..., "数据总条数"]
average_value: Annotated[float, ..., "平均值"]
trend: Annotated[str, ..., "趋势:上升/下降/平稳"]
model = init_chat_model(os.environ["LLM_MODEL"])
agent = create_agent(model=model, response_format=AnalysisResult)
result = agent.invoke({
"messages": [{"role": "user", "content": "分析销售:1月100万,2月120万,3月115万,4月135万,5月150万"}]
})
analysis = result["structured_response"]
# 返回的是普通字典,不是 Pydantic 实例
print(analysis["trend"]) # 上升TypedDict vs Pydantic:TypedDict 返回普通字典、无运行时验证、不支持字段约束;Pydantic 返回模型实例、有完整验证、支持 ge/le 等约束和嵌套模型。简单场景选 TypedDict,需要验证选 Pydantic。
代码示例 6: 直接在模型上使用
除了 create_agent,也可以直接在模型实例上调用 with_structured_output:
python
import os
from langchain.chat_models import init_chat_model
from pydantic import BaseModel, Field
class Movie(BaseModel):
"""电影信息"""
title: str = Field(description="电影标题")
year: int = Field(description="上映年份")
director: str = Field(description="导演")
rating: float = Field(description="评分(满分 10)")
model = init_chat_model(os.environ["LLM_MODEL"])
structured_model = model.with_structured_output(Movie)
movie = structured_model.invoke("介绍一下电影《盗梦空间》")
print(movie) # Movie(title='盗梦空间', year=2010, director='克里斯托弗·诺兰', rating=9.3)使用 JSON Schema
当 Schema 需要动态生成时,可以传入 JSON Schema 字典:
python
json_schema = {
"title": "Event",
"type": "object",
"properties": {
"title": {"type": "string", "description": "事件标题"},
"date": {"type": "string", "description": "日期,YYYY-MM-DD"},
"attendees": {"type": "array", "items": {"type": "string"}, "description": "参与人"},
},
"required": ["title", "date"],
}
structured_model = model.with_structured_output(json_schema, method="json_schema")
event = structured_model.invoke("明天下午3点在会议室A开项目评审会,参加人:张三、李四")
print(event)
# {'title': '项目评审会', 'date': '2026-02-22', 'attendees': ['张三', '李四']}代码示例 7: 复杂嵌套 Schema
Pydantic 天然支持模型嵌套,适合复杂业务场景:
python
import os
from pydantic import BaseModel, Field
from langchain.agents import create_agent
from langchain.chat_models import init_chat_model
class Address(BaseModel):
"""地址信息"""
province: str = Field(description="省份")
city: str = Field(description="城市")
street: str = Field(description="街道地址")
class Education(BaseModel):
"""教育经历"""
school: str = Field(description="学校名称")
degree: str = Field(description="学位")
major: str = Field(description="专业")
class PersonProfile(BaseModel):
"""个人档案"""
name: str = Field(description="姓名")
age: int = Field(description="年龄")
address: Address = Field(description="现居地址")
education: list[Education] = Field(description="教育经历列表")
skills: list[str] = Field(description="技能列表")
model = init_chat_model(os.environ["LLM_MODEL"])
agent = create_agent(model=model, response_format=PersonProfile)
result = agent.invoke({
"messages": [{
"role": "user",
"content": "提取简历:李明,28岁,现居北京市海淀区中关村大街1号。"
"清华大学硕士(计算机),北邮学士(软件工程)。擅长 Python、Go、K8s。"
}]
})
profile = result["structured_response"]
print(profile.name) # 李明
print(profile.address.city) # 北京市
print(profile.education[0].school) # 清华大学错误处理
Agent 级别的错误处理
当未使用 handle_errors 时,结构化输出解析失败会抛出异常。生产环境应捕获并处理:
python
from pydantic import ValidationError
try:
result = agent.invoke({
"messages": [{"role": "user", "content": "提取信息..."}]
})
data = result["structured_response"]
process_data(data)
except ValidationError as e:
print(f"数据验证失败: {e}") # Pydantic 字段约束不满足
except KeyError:
print("未获取到结构化输出") # 模型可能未返回结构化输出
except Exception as e:
print(f"调用失败: {e}") # 网络错误、API 限流等include_raw 获取原始输出
当你需要同时访问解析后的对象和原始输出(用于调试或日志记录)时,在模型上使用 include_raw=True:
python
structured_model = model.with_structured_output(Movie, include_raw=True)
result = structured_model.invoke("介绍电影《星际穿越》")
# result: {"raw": AIMessage(...), "parsed": Movie(...), "parsing_error": None}
if result["parsing_error"]:
print(f"解析失败: {result['parsing_error']}")
print(f"原始输出: {result['raw'].content}")
else:
print(result["parsed"].title)推荐的错误处理策略
python
# 生产推荐:ToolStrategy + handle_errors=True,解析失败不崩溃
agent = create_agent(
model=model,
response_format=ToolStrategy(MySchema, handle_errors=True),
)
result = agent.invoke({"messages": [...]})
structured = result["structured_response"]
if isinstance(structured, str):
# handle_errors=True 时,解析失败返回错误字符串
print(f"结构化输出解析失败: {structured}")
# 回退逻辑:使用默认值或重新请求
else:
# 正常处理结构化数据
process_data(structured)最佳实践
1. 为字段编写清晰的 description
python
# 好 - 描述具体、明确
class Order(BaseModel):
order_id: str = Field(description="订单编号,格式如 ORD-20260101-001")
amount: float = Field(description="订单金额,单位:人民币元")
# 差 - 描述模糊,模型容易误解
class Order(BaseModel):
order_id: str = Field(description="id")
amount: float = Field(description="金额")2. 使用 Literal 限制枚举值 + 合理设置可选字段
python
from typing import Literal
class Ticket(BaseModel):
priority: Literal["low", "medium", "high", "critical"] = Field(description="优先级")
title: str = Field(description="工单标题")
assignee: str | None = Field(default=None, description="负责人(可选)")
tags: list[str] = Field(default_factory=list, description="标签列表")3. 根据场景选择策略
python
from langchain.agents.structured_output import ToolStrategy, ProviderStrategy
# 模型支持原生结构化输出 -> 直接传 Schema(自动 ProviderStrategy)
agent = create_agent(model=model, response_format=MySchema)
# 需要严格模式 -> ProviderStrategy + strict=True(langchain>=1.2)
agent = create_agent(model=model, response_format=ProviderStrategy(MySchema, strict=True))
# 模型仅支持工具调用 / 需要错误处理 -> 显式 ToolStrategy
agent = create_agent(
model=model,
response_format=ToolStrategy(MySchema, handle_errors=True),
)
# 简单提取、无需 Agent -> 直接用模型
structured_model = model.with_structured_output(MySchema)常见问题
Q: ProviderStrategy 和 ToolStrategy 有什么区别?
A: ProviderStrategy 利用模型提供商的原生 JSON Schema 能力,延迟更低。ToolStrategy 通过"虚拟工具"约束输出格式,兼容性更广,且支持 handle_errors 和 tool_message_content。默认直接传 Schema 会自动选择 ProviderStrategy。
Q: strict=True 有什么作用?
A: strict=True(ProviderStrategy,langchain>=1.2)强制模型严格遵循 Schema,不允许额外字段。普通模式下模型可能返回 Schema 中未定义的字段;strict 模式禁止此行为,适用于对数据格式有严格要求的场景。
Q: 结构化输出和工具调用是什么关系?
A: 工具调用让模型调用外部函数获取结果;结构化输出让模型的最终回复符合指定格式。两者可同时使用--Agent 先调用工具收集信息,最后以结构化格式返回。ToolStrategy 本质上是利用工具调用机制来实现结构化输出。
Q: 输出不符合 Schema 怎么办?
A: 使用 ToolStrategy(Schema, handle_errors=True) 让解析失败返回错误信息而非抛异常;或在模型上使用 include_raw=True 捕获解析错误并获取原始输出调试。生产环境建议加重试逻辑或回退到 ToolStrategy。
Q: TypedDict 和 Pydantic 该用哪个?
A: 需要字段验证约束(ge/le/min_length)或嵌套模型用 Pydantic;只需简单键值结构用 TypedDict。
Q: 支持 Union 类型吗?
A: ToolStrategy 支持 Union[SchemaA, SchemaB],模型会根据输入内容自动选择最匹配的 Schema 返回。ProviderStrategy 对 Union 的支持取决于模型提供商。
下一步
现在你已经掌握了结构化输出的使用,接下来可以:
- 学习 智能体 Agent - 了解
create_agent的完整能力 - 探索 模型配置 - 深入了解
init_chat_model和模型参数 - 进阶 工具系统 - 结合工具调用 + 结构化输出构建完整应用