Skip to content

结构化输出

本节解决什么问题

在实际应用中,我们通常不只是需要 LLM 返回一段自然语言文本--我们需要结构化数据:提取的实体、分类标签、评分数值等。LangChain 的结构化输出(Structured Output)功能允许你定义一个 Schema,让模型的输出自动解析为符合该 Schema 的 Python 对象,而不是原始字符串。

必须深刻理解,不能跳过:结构化输出是"生成时约束",不是"后处理校验"

前端的 Zod / Joi 是后处理校验:先拿到数据,再验证是否符合 Schema,不符合则报错。LangChain 的结构化输出是生成时约束:在模型推理阶段就引导输出格式,模型"知道"必须按 Schema 生成。这意味着输出几乎总是符合 Schema,但仍可能在极端情况下出错(模型幻觉、格式错误),因此需要错误处理机制(handle_errorsinclude_raw)。

前端类比

结构化输出类似于前端中的 TypeScript 接口 + Zod 验证。你定义一个 interface 描述数据结构,Zod 在运行时校验数据是否合规。LangChain 的结构化输出做的是同一件事:用 Pydantic Model 定义"接口",框架自动确保 LLM 输出符合该"接口"。

不过要注意,LLM 的结构化输出是生成时约束,而非后处理校验--它在模型推理阶段就引导输出格式,这比前端的"先拿到数据再校验"更高效。

原生语义

Zod 校验的是已经存在的数据对象,通过则放行、不通过则抛错。LangChain 结构化输出约束的是模型的生成过程:ProviderStrategy 通过模型的 JSON Schema / JSON mode 原生能力在生成时约束;ToolStrategy 通过"虚拟工具调用"机制让模型按工具参数 Schema 生成。两者都不是"生成完再校验",而是"引导生成符合 Schema"。

🔗 LangChain Structured Output 官方文档

为什么需要结构化输出?

  1. 类型安全 - 直接获得 Python 对象(Pydantic Model / TypedDict),无需手动解析 JSON
  2. 减少幻觉 - Schema 约束模型输出范围,降低生成无关内容的概率
  3. 管线友好 - 输出可直接传入下游函数或写入数据库,无需 JSON.parse
  4. 降低成本 - 不需要额外的 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.5

strict 模式:普通模式下模型可能返回 Schema 中未定义的额外字段;strict=True 禁止额外字段,确保输出严格匹配 Schema 定义。适用于对数据格式有严格要求的场景(如写入数据库、API 交互)。

代码示例 3: ToolStrategy + 错误处理

当模型不支持原生结构化输出,或你想确保使用工具调用方式时,显式使用 ToolStrategyhandle_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_errorstool_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_errorstool_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 和模型参数
  • 进阶 工具系统 - 结合工具调用 + 结构化输出构建完整应用

参考资源

学习文档整合站点