Structured Output
结构化输出(Structured Output)是 LangChain Agent 的核心能力之一。它确保模型的返回结果遵循预定义的数据结构,从而方便程序化处理和类型安全。LangChain 提供了两种策略来实现结构化输出。
两种策略
ProviderStrategy(提供商原生策略)
利用模型提供商原生支持的结构化输出功能(如 OpenAI 的 JSON Mode、Gemini 的 response_schema、Claude 的 structured output)。
python
from langchain.chat_models import init_chat_model
from langchain.structured_output import ProviderStrategy
from pydantic import BaseModel
class Weather(BaseModel):
city: str
temperature: float
# 默认使用 ProviderStrategy
model = init_chat_model("openai/gpt-4o")
# 直接使用模型的结构化输出
structured_model = model.with_structured_output(Weather)
result = structured_model.invoke("北京的天气?")
print(result.city) # "北京"工作原理:ProviderStrategy 将 JSON Schema 通过 API 参数传递给模型提供商,由模型端直接生成结构化输出。这是最快、最可靠的方式。
适用模型:
- OpenAI:JSON Mode + Structured Outputs
- Gemini:response_schema 参数
- Claude:扩展到 JSON 格式
ToolStrategy(工具调用策略)
通过模拟工具调用来实现结构化输出。模型将结构化输出当作一个"工具调用来执行"。
python
from langchain.structured_output import ToolStrategy
from pydantic import BaseModel
class Analysis(BaseModel):
topic: str
key_points: list[str]
conclusion: str
# 指定使用 ToolStrategy
model = init_chat_model("openai/gpt-4o")
structured_model = model.with_structured_output(
Analysis,
method="tool_calling", # 使用工具调用策略
)
result = structured_model.invoke("分析人工智能的发展趋势")适用场景:
- 模型原生不支持结构化输出时
- 输出 Schema 非常复杂时
- 需要更灵活的解析控制时
策略对比
| 特性 | ProviderStrategy | ToolStrategy |
|---|---|---|
| 性能 | 更快 | 略有额外开销 |
| 可靠性 | 更高 | 取决于工具调用能力 |
| 兼容性 | 需提供商支持 | 所有支持工具调用的模型 |
| 复杂 Schema | 有限制 | 更灵活 |
| 嵌套结构 | 支持 | 支持 |
定义输出结构
Pydantic BaseModel(推荐)
Pydantic 提供了最丰富的类型支持和验证能力:
python
from pydantic import BaseModel, Field
from typing import Optional
from datetime import datetime
class Address(BaseModel):
"""地址信息"""
street: str = Field(description="街道地址")
city: str = Field(description="城市")
state: Optional[str] = Field(description="州/省")
country: str = Field(description="国家", default="中国")
zip_code: Optional[str] = Field(description="邮政编码")
class Person(BaseModel):
"""个人信息"""
name: str = Field(description="姓名")
age: int = Field(description="年龄", ge=0, le=150)
email: Optional[str] = Field(description="电子邮箱")
address: Address = Field(description="地址")
created_at: datetime = Field(description="创建时间")
# 嵌套模型自动处理
agent = create_agent(
model=init_chat_model("openai/gpt-4o"),
tools=[],
system_prompt="提取个人信息。",
response_format=Person,
)
result = agent.invoke({
"messages": [{"role": "user", "content": "我叫张三,28岁,住在北京市朝阳区"}]
})
print(result.name) # "张三"
print(result.address.city) # "北京"dataclass
Python 标准库的 dataclass 也是轻量级选择:
python
from dataclasses import dataclass, field
from typing import Optional
@dataclass
class Product:
name: str
price: float
category: str
tags: list[str] = field(default_factory=list)
description: Optional[str] = None
agent = create_agent(
model=init_chat_model("openai/gpt-4o"),
tools=[],
system_prompt="提取商品信息。",
response_format=Product,
)TypedDict
适用于不需要验证逻辑的纯数据结构:
python
from typing import TypedDict, Optional
class BookInfo(TypedDict):
title: str
author: str
isbn: Optional[str]
year: int
genres: list[str]
agent = create_agent(
model=init_chat_model("openai/gpt-4o"),
tools=[],
system_prompt="提取书籍信息。",
response_format=BookInfo,
)
result = agent.invoke({
"messages": [{"role": "user", "content": "《三体》刘慈欣,2008年出版"}]
})
print(result["title"]) # "三体"
print(result["author"]) # "刘慈欣"JSON Schema
直接使用原生 JSON Schema,最大灵活性:
python
json_schema = {
"type": "object",
"properties": {
"name": {"type": "string"},
"metrics": {
"type": "object",
"properties": {
"accuracy": {"type": "number"},
"latency_ms": {"type": "integer"}
},
"required": ["accuracy", "latency_ms"]
}
},
"required": ["name", "metrics"]
}
agent = create_agent(
model=init_chat_model("openai/gpt-4o"),
tools=[],
system_prompt="提取评估指标。",
response_format=json_schema,
)在 Agent 中使用
在 create_agent 中使用 response_format 是最推荐的方式:
python
from langchain import create_agent
from langchain.chat_models import init_chat_model
from pydantic import BaseModel, Field
class ResearchReport(BaseModel):
title: str = Field(description="报告标题")
summary: str = Field(description="摘要,100-200字")
key_findings: list[str] = Field(description="主要发现")
recommendations: list[str] = Field(description="建议")
confidence_score: float = Field(description="可信度评分 0-1", ge=0, le=1)
agent = create_agent(
model=init_chat_model("openai/gpt-4o"),
tools=[search_tool, database_tool],
system_prompt="进行深入研究和分析。",
response_format=ResearchReport,
)
report = agent.invoke({
"messages": [{"role": "user", "content": "研究2025年新能源汽车市场趋势"}]
})
print(f"标题:{report.title}")
print(f"发现:{report.key_findings}")流式结构化输出
LangChain 支持逐步输出结构化数据:
python
from langchain.chat_models import init_chat_model
from pydantic import BaseModel
class StreamingResult(BaseModel):
content: str
tokens: int
model = init_chat_model("openai/gpt-4o")
structured = model.with_structured_output(StreamingResult)
# 流式获取部分结果
for chunk in structured.stream("讲一个故事"):
print(chunk) # 逐步返回结构化的部分结果处理输出失败
python
from langchain.structured_output import StructuredOutputError
try:
result = agent.invoke({
"messages": [{"role": "user", "content": "你好"}]
})
except StructuredOutputError as e:
print(f"结构化输出失败:{e}")
# 回退到无结构输出
fallback_result = agent.invoke(
{"messages": [{"role": "user", "content": "你好"}]},
response_format=None,
)最佳实践
- 优先使用 ProviderStrategy,仅在必要时回退到 ToolStrategy
- Pydantic BaseModel 是首选,平衡了类型安全和灵活性
- 为每个字段添加描述,显著提升生成准确性
- 避免过深的嵌套结构(建议不超过 3 层)
- 字段名使用 snake_case,保持 Python 风格一致
- 对枚举值使用
Literal类型,约束输出范围 - 测试不同的 Schema 设计,找到最适合你场景的结构