现在位置: 首页 > LangChain 教程 > 正文

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 typing import Literal
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 import create_agent
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 pydantic import BaseModel, Field
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 typing import Literal
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 typing_extensions import TypedDict
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

两种方式的区别:

方式返回类型主要特点
PydanticBaseModel 实例支持丰富的字段约束与自定义验证
TypedDictdict声明简单,适合普通结构化数据

如果涉及复杂业务校验,通常优先选择 Pydantic。


八、可选字段与缺失数据

实际文本中经常存在信息不完整的情况。

例如:

我正在学习 Python 基础教程,属于入门级别。

用户没有提供预计学习时长,也没有说明是否收费。

此时不应该让模型自行编造这两个字段。

可以将字段定义为可选:

实例

from typing import Literal
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 typing import Literal
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 pydantic import BaseModel, Field
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 pydantic import BaseModel, Field
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 typing import Literal
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 检查字段和业务规则,并通过可信数据源验证重要事实。结构化输出解决的是数据组织问题,而不是模型内容准确性的全部问题。