LangChain 结构化输出 Structured Output
大多数时候,我们需要的并不是一段自由文本,而是程序能够直接处理的数据,例如 JSON 对象、用户信息、商品数据或课程推荐结果。
LangChain 提供 Structured Output(结构化输出) 功能,可以让模型按照指定的数据结构返回结果,减少手动解析文本的工作。
在 LangChain 1.x 中,可以通过 create_agent(response_format=...) 配置 Agent 的结构化输出,也可以直接使用模型的 with_structured_output() 方法。
本章节介绍 Pydantic、TypedDict、ProviderStrategy、ToolStrategy、嵌套结构以及错误处理。
一、为什么需要结构化输出?
假设用户提供了一段描述:
张三今年 28 岁,是一名 Python 工程师。
普通模型可能返回:
张三今年 28 岁,职业是 Python 工程师。
如果需要将这些信息写入数据库,还需要从自然语言中提取姓名、年龄和职业。
而结构化输出可以直接返回:
{
"name": "张三",
"age": 28,
"job": "Python 工程师"
}
| 方式 | 返回内容 | 后续处理 |
|---|---|---|
| 普通文本 | 自然语言回答 | 需要解析和提取字段 |
| 结构化输出 | Pydantic 对象或字典 | 可以按字段访问和校验 |
结构化输出的优势在于:
- 明确字段名称和数据类型。
- 减少对正则表达式和文本解析的依赖。
- 支持字段范围、枚举及其他约束。
- 便于与数据库、API 和前端组件集成。
注意:结构化输出可以检查数据格式,但不能保证数据内容一定真实。例如,模型可以生成符合 Schema 的错误日期或不存在的课程名称。
二、安装与环境准备
安装 LangChain 及 DeepSeek 集成:
pip install -U langchain langchain-deepseek python-dotenv pydantic
项目目录创建 .env 文件:
DEEPSEEK_API_KEY=your_deepseek_api_key
使用以下代码加载环境变量:
from dotenv import load_dotenv load_dotenv()
本教程使用 deepseek:deepseek-flash 作为示例模型。不同模型对原生结构化输出、Tool Calling 和 JSON Schema 的支持可能不同,需要以实际集成能力为准。
三、最简单的用法:Pydantic 模型
使用 Pydantic 定义需要的输出结构,然后通过 response_format 传给 create_agent()。
例如,从一段用户描述中提取课程名称、难度、学习时长和是否免费。
实例:提取课程信息
实例
from dotenv import load_dotenv
from pydantic import BaseModel, Field
from langchain.agents import create_agent
load_dotenv()
class CourseInfo(BaseModel):
"""从文本中提取课程信息。"""
course_name: str = Field(description="课程名称")
difficulty: Literal["入门", "进阶", "高级"]
estimated_hours: int = Field(ge=1, description="预计学习小时数")
is_free: bool = Field(description="是否免费")
agent = create_agent(
model="deepseek:deepseek-flash",
tools=[],
response_format=CourseInfo,
system_prompt="只提取用户明确提供的课程信息,不要编造。"
)
result = agent.invoke({
"messages": [{"role": "user", "content": "我在学 Python3 基础教程,属于入门课程,预计学习20小时,而且是免费的。"}]
})
course = result["structured_response"]
print("课程名:", course.course_name)
print("难度:", course.difficulty)
print("时长:", course.estimated_hours, "小时")
print("免费:", course.is_free)
print("对象类型:", type(course).__name__)
如果模型成功提取,输出类似:
课程名: Python3 基础教程 难度: 入门 时长: 20 小时 免费: True 对象类型: CourseInfo
其中:
BaseModel:定义数据结构。Field:配置字段说明及约束。Literal:限制允许的字段值。response_format:指定 Agent 输出结构。structured_response:获取最终结构化结果。
使用 Pydantic Schema 且解析成功后,structured_response 是 Pydantic 模型实例,可以通过属性直接访问。
转换为 JSON
如果需要发送给前端,可以使用:
print(course.model_dump()) print(course.model_dump_json(indent=2))
model_dump() 返回 Python 字典,model_dump_json() 返回 JSON 字符串。
注意:如果原文没有提供某个字段,不应该让模型随意猜测。可以使用可选字段或要求补充信息。
四、ProviderStrategy 与 ToolStrategy
LangChain 1.x 提供两种主要的 Agent 结构化输出策略:
| 策略 | 实现方式 | 适用场景 |
|---|---|---|
ProviderStrategy | 使用模型服务商原生结构化输出能力 | 提供商支持原生 Schema 约束 |
ToolStrategy | 通过 Tool Calling 生成结构化数据 | 支持工具调用的模型 |
| 直接传 Schema | 由 LangChain 根据模型能力选择策略 | 希望自动选择的常规场景 |
1、自动选择策略
最简单的写法:
agent = create_agent(
model="deepseek:deepseek-flash",
response_format=CourseInfo
)
LangChain 会根据当前模型的能力信息,选择支持的结构化输出方式。
如果模型支持原生结构化输出,可能使用 ProviderStrategy;否则可采用 ToolStrategy。
自动选择依赖模型能力信息,并不保证所有模型都能完成结构化输出。
2、显式使用 ToolStrategy
对于使用 Tool Calling 完成结构化输出的模型,可以显式配置:
实例
from langchain.agents.structured_output import ToolStrategy
agent = create_agent(
model="deepseek:deepseek-flash",
tools=[],
response_format=ToolStrategy(CourseInfo),
system_prompt="从用户描述中提取课程信息。"
)
result = agent.invoke({
"messages": [{"role": "user", "content": "Python3 基础教程是免费入门课程,预计学习20小时。"}]
})
print(result["structured_response"].model_dump())
ToolStrategy 会通过模型的工具调用机制传递结构化数据,再由 LangChain 进行解析和校验。
这里的结构化输出工具主要用于传递数据 Schema,不等于执行了一个真实的课程查询 Python 函数。
3、显式使用 ProviderStrategy
如果模型及服务商明确支持原生结构化输出,可以使用:
from langchain.agents.structured_output import ProviderStrategy
agent = create_agent(
model=model,
response_format=ProviderStrategy(CourseInfo)
)
这种方式要求实际使用的模型及提供商支持相应结构化输出能力。
注意:不要仅因为模型支持 JSON 文本输出,就认定它完全支持原生 JSON Schema 约束。具体能力应参考对应模型集成文档。
五、与工具共存的 Structured Output
结构化输出和普通工具可以组合使用。
例如,课程推荐 Agent 可以先搜索课程,再将最终结果整理成指定的结构。
需要注意,模型必须支持当前结构化输出策略及普通工具调用的组合。
实例:搜索课程并生成推荐
实例
from langchain.tools import tool
from langchain.agents import create_agent
from langchain.agents.structured_output import ToolStrategy
COURSES = {
"python": "Python3 基础教程,入门,模拟学习时长20小时",
"java": "Java 基础教程,入门,模拟学习时长25小时"
}
@tool
def search_course(keyword: str) -> str:
"""根据编程语言查询模拟课程。"""
return COURSES.get(keyword.lower(), "未找到相关模拟课程")
class CourseRecommendation(BaseModel):
"""课程推荐结果。"""
course_name: str = Field(description="推荐课程")
reason: str = Field(description="推荐理由")
difficulty: str = Field(description="课程难度")
agent = create_agent(
model="deepseek:deepseek-flash",
tools=[search_course],
response_format=ToolStrategy(CourseRecommendation),
system_prompt="你是编程学习顾问。查询课程时使用工具,仅推荐工具实际返回的课程。"
)
result = agent.invoke({
"messages": [{"role": "user", "content": "我零基础想学 Python,请查询课程并推荐。"}]
})
rec = result["structured_response"]
print("课程:", rec.course_name)
print("理由:", rec.reason)
print("难度:", rec.difficulty)
for msg in result["messages"]:
if msg.type == "tool":
print("工具消息:", msg.name, msg.content)
如果模型正确调用查询工具,可能得到:
课程: Python3 基础教程 理由: 适合没有编程基础的学习者 难度: 入门
工具消息中可能包含课程搜索结果。
但需要注意,result["messages"] 中可能还有结构化输出策略产生的 ToolMessage,因此不应将每条 ToolMessage 都当作实际业务数据库查询。
注意:模型不一定严格按照预期调用普通工具。如果必须查询真实数据库并保证数据来源,应在业务工作流中强制执行查询。
六、复杂嵌套结构
Pydantic 支持列表、嵌套模型、枚举和字段限制,适合用于学习计划、订单信息等复杂数据结构。
实例:生成 Python 学习计划
实例
from pydantic import BaseModel, Field, model_validator
from langchain.agents import create_agent
from langchain.agents.structured_output import ToolStrategy
class Topic(BaseModel):
"""单个学习主题。"""
name: str
order: int = Field(ge=1)
minutes: int = Field(gt=0)
class LearningPlan(BaseModel):
"""学习计划。"""
goal: str
level: Literal["入门", "进阶", "高级"]
total_hours: float = Field(gt=0, le=10)
topics: list[Topic] = Field(min_length=1)
@model_validator(mode="after")
def check_duration(self):
total = sum(topic.minutes for topic in self.topics)
if abs(total / 60 - self.total_hours) > 0.01:
raise ValueError("total_hours 必须等于全部主题分钟数之和")
return self
agent = create_agent(
model="deepseek:deepseek-flash",
response_format=ToolStrategy(LearningPlan),
system_prompt="制定合理的学习计划。所有主题分钟数之和必须等于 total_hours × 60。"
)
result = agent.invoke({
"messages": [{"role": "user", "content": "制定一个总时长不超过10小时的 Python 入门学习计划。"}]
})
plan = result["structured_response"]
print("目标:", plan.goal)
print("难度:", plan.level)
print("总时长:", plan.total_hours, "小时")
for topic in plan.topics:
print(f"{topic.order}. {topic.name}({topic.minutes}分钟)")
成功执行后,可能得到:
目标:掌握 Python 基础语法 难度:入门 总时长:6.0 小时 1. Python 环境与变量(60分钟) 2. 数据类型(60分钟) 3. 条件判断(60分钟) 4. 循环(60分钟) 5. 函数(60分钟) 6. 综合练习(60分钟)
这里不仅验证 total_hours 不超过 10 小时,还通过 model_validator 校验总时长与各主题分钟数是否一致。
如果模型返回不符合约束的数据,Pydantic 会产生验证错误。使用 ToolStrategy 时,LangChain 可以根据错误处理策略要求模型重新生成。
但不能保证模型一定能够在有限次数内生成有效结果。
七、使用 TypedDict 返回字典
除了 Pydantic,LangChain 也支持使用 TypedDict 定义结构化输出。
与 Pydantic 模型不同,TypedDict 方案通常返回普通 Python 字典。
实例:提取人员信息
实例
from langchain.agents import create_agent
from langchain.agents.structured_output import ToolStrategy
class PersonInfo(TypedDict):
"""人员信息。"""
name: str
age: int
job: str
agent = create_agent(
model="deepseek:deepseek-flash",
response_format=ToolStrategy(PersonInfo),
system_prompt="提取用户明确提供的人员信息,不要猜测。"
)
result = agent.invoke({
"messages": [{"role": "user", "content": "张三今年28岁,是一名Python工程师。"}]
})
person = result["structured_response"]
print(person["name"])
print(person["age"])
print(person["job"])
print(type(person).__name__)
可能输出:
张三 28 Python工程师 dict
两种方式的区别:
| 方式 | 返回类型 | 主要特点 |
|---|---|---|
| Pydantic | BaseModel 实例 | 支持丰富的字段约束与自定义验证 |
| TypedDict | dict | 声明简单,适合普通结构化数据 |
如果涉及复杂业务校验,通常优先选择 Pydantic。
八、可选字段与缺失数据
实际文本中经常存在信息不完整的情况。
例如:
我正在学习 Python 基础教程,属于入门级别。
用户没有提供预计学习时长,也没有说明是否收费。
此时不应该让模型自行编造这两个字段。
可以将字段定义为可选:
实例
from pydantic import BaseModel, Field
class CourseInfo(BaseModel):
"""允许缺失字段的课程信息。"""
course_name: str
difficulty: Literal["入门", "进阶", "高级"] | None = None
estimated_hours: int | None = Field(default=None, ge=1)
is_free: bool | None = None
未提供的数据可以使用 None 表示。
例如:
{
"course_name": "Python 基础教程",
"difficulty": "入门",
"estimated_hours": null,
"is_free": null
}
注意:声明可选字段只是允许缺失值,并不能彻底防止模型猜测。因此仍需在 System Prompt 中说明只提取明确提供的信息。
九、直接使用 with_structured_output()
如果任务只是从文本中提取信息,不需要 Agent 调用工具,可以直接使用聊天模型的 with_structured_output()。
这通常比创建 Agent 更简单。
实例:情感分析
实例
from pydantic import BaseModel, Field
from langchain.chat_models import init_chat_model
class SentimentResult(BaseModel):
"""情感分析结果。"""
sentiment: Literal["积极", "消极", "中性"]
score: float = Field(ge=0, le=1, description="情感判断强度")
keywords: list[str]
model = init_chat_model("deepseek:deepseek-flash")
structured_model = model.with_structured_output(SentimentResult)
texts = [
"菜鸟教程真的很好用,强烈推荐!",
"这个教程内容比较少,有些失望。",
"我今天学习了 Python 变量。"
]
for text in texts:
result = structured_model.invoke(text)
print(text)
print(result.model_dump())
可能得到:
积极,关键词:好用、推荐 消极,关键词:内容少、失望 中性,关键词:Python、变量
其中的情感分类和评分由模型判断,不具有绝对客观性。
尤其要注意,score 并不自动代表经过统计校准的概率或置信度。
如果需要用于真实业务决策,应结合测试数据评估模型表现。
与 Agent 的区别
| 方式 | 返回值 | 适用场景 |
|---|---|---|
create_agent(response_format=...) | Agent 状态,包含 structured_response | 需要工具调用、多步骤处理 |
model.with_structured_output(...) | 解析后的结构化数据 | 文本提取、分类、简单解析 |
注意:with_structured_output() 的内部实现由模型集成决定,可能使用工具调用或提供商原生输出能力。
十、include_raw:查看原始模型响应
使用 with_structured_output() 时,有时需要同时查看模型原始消息及解析结果。
可以设置:
include_raw=True
实例
实例
from langchain.chat_models import init_chat_model
class Person(BaseModel):
name: str
age: int = Field(ge=0)
model = init_chat_model("deepseek:deepseek-flash")
structured_model = model.with_structured_output(Person, include_raw=True)
result = structured_model.invoke("张三今年28岁。")
print("解析结果:", result["parsed"])
print("原始消息:", result["raw"])
print("解析错误:", result["parsing_error"])
返回结构通常包含:
| 字段 | 说明 |
|---|---|
raw | 模型生成的原始 AIMessage |
parsed | 解析成功的结构化对象,失败时可能为 None |
parsing_error | 解析异常信息,成功时通常为 None |
include_raw=True 适合调试模型输出格式。
注意:它并不意味着所有错误都会转化为 parsing_error。例如,网络连接失败、API 鉴权错误和请求超时仍可能直接抛出异常。
十一、结构化输出的错误处理
模型生成结构化数据时,可能出现以下问题:
- 缺少必填字段。
- 字段类型不正确。
- 枚举值不在允许范围内。
- 数值超出限制。
- 一次返回多个互相冲突的结构化结果。
对于 ToolStrategy,LangChain 提供 handle_errors 参数,可以控制 Schema 验证失败后的处理方式。
| 配置 | 行为 |
|---|---|
True | 自动处理受支持的结构化输出错误,并尝试让模型修正 |
False | 不进行该策略的自动错误恢复,异常继续抛出 |
| 字符串 | 使用指定的错误提示反馈给模型 |
| 异常类型 | 只处理指定类型的异常 |
| Callable | 通过自定义函数生成错误提示 |
实例:自定义错误提示
实例
from langchain.agents import create_agent
from langchain.agents.structured_output import ToolStrategy
class ProductRating(BaseModel):
"""商品评分。"""
rating: int = Field(ge=1, le=5)
comment: str
agent = create_agent(
model="deepseek:deepseek-flash",
response_format=ToolStrategy(
ProductRating,
handle_errors="评分必须为1到5的整数,并且需要提供评论。"
)
)
result = agent.invoke({
"messages": [{"role": "user", "content": "这个商品不错,我给它4分。"}]
})
print(result["structured_response"])
如果模型返回非法评分,LangChain 可以将错误提示交给模型,让其尝试重新生成。
但重试不保证最终一定成功,也可能增加模型调用次数和 Token 消耗。
注意:这里的 ToolStrategy(handle_errors=...) 与普通工具的 BaseTool.handle_tool_error 是不同机制,不应混淆。
十二、完整实例:从用户描述中提取课程信息
下面整合 Pydantic、可选字段、字段约束、ToolStrategy 和 JSON 输出,创建一个相对完整的课程信息提取程序。
文件名:structured_output_demo.py
完整实例
from dotenv import load_dotenv
from pydantic import BaseModel, Field
from langchain.agents import create_agent
from langchain.agents.structured_output import ToolStrategy
load_dotenv()
class CourseInfo(BaseModel):
"""从用户文本提取课程资料。"""
course_name: str = Field(min_length=1, description="课程名称")
difficulty: Literal["入门", "进阶", "高级"] | None = None
estimated_hours: int | None = Field(default=None, ge=1, le=1000)
is_free: bool | None = Field(default=None, description="是否免费")
agent = create_agent(
model="deepseek:deepseek-flash",
response_format=ToolStrategy(CourseInfo),
system_prompt=(
"从用户描述中提取课程信息。"
"只使用用户明确提供的数据,缺失字段使用 null。"
"不要猜测课程价格、会员权限或学习时长。"
)
)
texts = [
"Python3 基础教程是入门课程,预计20小时,免费。",
"Java 进阶课程需要大约35小时。",
"我最近正在学习 HTML 基础教程。"
]
for text in texts:
print("\n原文:", text)
try:
result = agent.invoke({"messages": [{"role": "user", "content": text}]})
course = result["structured_response"]
print(course.model_dump_json(indent=2))
except Exception as error:
print("提取失败:", type(error).__name__)
其中第一段文本如果提取成功,结构化结果应接近:
{
"course_name": "Python3 基础教程",
"difficulty": "入门",
"estimated_hours": 20,
"is_free": true
}
第三段文本没有说明难度、时长和收费信息,因此期望返回:
{
"course_name": "HTML 基础教程",
"difficulty": null,
"estimated_hours": null,
"is_free": null
}
具体结果取决于模型输出。即使 Schema 允许空值,模型仍有可能错误猜测,因此需要根据业务需求进行数据验证。
十三、常见问题
1、structured_response 是字符串还是对象?
取决于使用的 Schema。Pydantic 通常返回对应模型实例;TypedDict、dataclass 和 JSON Schema 策略通常返回字典。
2、为什么 structured_response 不存在?
可能是没有配置 response_format,或者结构化输出流程未成功完成。
3、Pydantic 和 TypedDict 应该选哪个?
如果需要复杂字段校验、嵌套模型或自定义验证器,推荐 Pydantic;如果只需要简单字典结构,可以考虑 TypedDict。
4、ProviderStrategy 和 ToolStrategy 有什么区别?
ProviderStrategy 使用模型服务商原生结构化输出机制;ToolStrategy 通过模型工具调用完成结构化数据生成。
5、所有模型都支持结构化输出吗?
不是。不同模型支持的功能存在差异,尤其是原生 JSON Schema、Tool Calling 和复杂嵌套结构。
6、response_format 可以和 tools 同时使用吗?
可以,但需要模型和所使用的结构化输出方式支持相应组合。
7、如何防止模型编造缺失字段?
可以将字段定义为可选,在系统提示词中要求缺失信息返回 null,并在应用层进行验证。
8、结构化输出能保证数据正确吗?
不能。Schema 主要约束数据格式及部分数值关系,不负责验证所有业务事实。
9、为什么模型一直重试结构化输出?
可能因为模型多次生成不符合 Schema 的内容。可以简化 Schema、改进字段描述,或调整 ToolStrategy 的错误处理配置。
10、什么时候不需要 create_agent()?
如果任务只是文本分类、信息提取或简单结构化转换,可以直接使用模型的 with_structured_output()。
总结
LangChain 结构化输出可以让模型按照指定 Schema 返回数据,方便程序直接使用。
本章节介绍了:
response_format:配置 Agent 结构化输出。Pydantic:定义数据结构和字段校验。TypedDict:定义字典格式的输出。ProviderStrategy:使用模型原生结构化输出。ToolStrategy:通过工具调用实现结构化输出。structured_response:获取 Agent 结构化结果。with_structured_output():直接让聊天模型输出结构化数据。include_raw=True:检查原始响应与解析异常。handle_errors:处理 ToolStrategy 的结构化输出错误。
最常见的写法:
agent = create_agent(
model="deepseek:deepseek-flash",
response_format=CourseInfo
)
result = agent.invoke({
"messages": [{"role": "user", "content": "提取课程信息"}]
})
data = result["structured_response"]
如果不需要 Agent 工具调用,也可以直接使用:
structured_model = model.with_structured_output(CourseInfo)
data = structured_model.invoke("需要提取的文本")
核心原则:使用 Schema 约束数据格式,使用 Pydantic 检查字段和业务规则,并通过可信数据源验证重要事实。结构化输出解决的是数据组织问题,而不是模型内容准确性的全部问题。
