LangChain 源码剖析-结构化输出详解(Structured output)

结构化输出允许代理以特定的、可预测的格式返回数据。您可以获得JSON对象、Pydantic模型或应用程序可以直接使用的数据类形式的结构化数据，而不是解析自然语言响应。
LangChain的create_agent自动处理结构化输出。用户设置他们想要的结构化输出模式，当模型生成结构化数据时，它会被捕获、验证，并在代理状态的"structured_response"键中返回。

def create_agent(    ...    response_format: Union[        ToolStrategy[StructuredResponseT],        ProviderStrategy[StructuredResponseT],        type[StructuredResponseT],    ]

响应格式(Response Format)

控制代理返回结构化数据的方式
当直接提供模式类型时，LangChain会自动选择

供应商策略(ProviderStrategy[StructuredResponseT])

使用提供程序本机结构化输出
一些模型提供者通过其API原生支持结构化输出（例如OpenAI、Grok、Gemini）。这是最可靠的方法。

class ProviderStrategy(Generic[SchemaT]):    schema: type[SchemaT]

ProviderStrategy: schema参数

Pydantic模型：带字段验证的BaseModel子类

pip install pydanic

from pydantic import BaseModel, Fieldfrom langchain.agents import create_agentclass ContactInfo(BaseModel):    """Contact information for a person."""    name: str = Field(description="The name of the person")    email: str = Field(description="The email address of the person")    phone: str = Field(description="The phone number of the person")agent = create_agent(    model="gpt-5",    response_format=ContactInfo  # Auto-selects ProviderStrategy)result = agent.invoke({    "messages": [{"role": "user", "content": "Extract contact info from: John Doe, john@example.com, (555) 123-4567"}]})print(result["structured_response"])# ContactInfo(name='John Doe', email='john@example.com', phone='(555) 123-4567')

Dataclasses：带有类型注释的Python数据类

from dataclasses import dataclassfrom langchain.agents import create_agent@dataclassclass ContactInfo:    """Contact information for a person."""    name: str # The name of the person    email: str # The email address of the person    phone: str # The phone number of the personagent = create_agent(    model="gpt-5",    tools=tools,    response_format=ContactInfo  # Auto-selects ProviderStrategy)result = agent.invoke({    "messages": [{"role": "user", "content": "Extract contact info from: John Doe, john@example.com, (555) 123-4567"}]})result["structured_response"]# ContactInfo(name='John Doe', email='john@example.com', phone='(555) 123-4567')

TypedDict:类型化字典类

from typing_extensions import TypedDictfrom langchain.agents import create_agentclass ContactInfo(TypedDict):    """Contact information for a person."""    name: str # The name of the person    email: str # The email address of the person    phone: str # The phone number of the personagent = create_agent(    model="gpt-5",    tools=tools,    response_format=ContactInfo  # Auto-selects ProviderStrategy)result = agent.invoke({    "messages": [{"role": "user", "content": "Extract contact info from: John Doe, john@example.com, (555) 123-4567"}]})result["structured_response"]# {'name': 'John Doe', 'email': 'john@example.com', 'phone': '(555) 123-4567'}

JSON Schema：具有JSON模式规范的字典

from langchain.agents import create_agentcontact_info_schema = {    "type": "object",    "description": "Contact information for a person.",    "properties": {        "name": {"type": "string", "description": "The name of the person"},        "email": {"type": "string", "description": "The email address of the person"},        "phone": {"type": "string", "description": "The phone number of the person"}    },    "required": ["name", "email", "phone"]}agent = create_agent(    model="gpt-5",    tools=tools,    response_format=ProviderStrategy(contact_info_schema))result = agent.invoke({    "messages": [{"role": "user", "content": "Extract contact info from: John Doe, john@example.com, (555) 123-4567"}]})result["structured_response"]# {'name': 'John Doe', 'email': 'john@example.com', 'phone': '(555) 123-4567'}

工具策略(ToolStrategy[StructuredResponseT])

使用工具调用进行结构化输出
对于不支持本机结构化输出的模型，LangChain使用工具调用来实现相同的结果。这适用于支持工具调用的所有模型，这是最现代的模型。

class ToolStrategy(Generic[SchemaT]):    schema: type[SchemaT]    tool_message_content: str | None    handle_errors: Union[        bool,        str,        type[Exception],        tuple[type[Exception], ...],        Callable[[Exception], str],    ]

ToolStrategy: schema参数

Pydantic模型：带字段验证的BaseModel子类

from pydantic import BaseModel, Fieldfrom typing import Literalfrom langchain.agents import create_agentfrom langchain.agents.structured_output import ToolStrategyclass ProductReview(BaseModel):    """Analysis of a product review."""    rating: int | None = Field(description="The rating of the product", ge=1, le=5)    sentiment: Literal["positive", "negative"] = Field(description="The sentiment of the review")    key_points: list[str] = Field(description="The key points of the review. Lowercase, 1-3 words each.")agent = create_agent(    model="gpt-5",    tools=tools,    response_format=ToolStrategy(ProductReview))result = agent.invoke({    "messages": [{"role": "user", "content": "Analyze this review: 'Great product: 5 out of 5 stars. Fast shipping, but expensive'"}]})result["structured_response"]# ProductReview(rating=5, sentiment='positive', key_points=['fast shipping', 'expensive'])

Dataclasses：带有类型注释的Python数据类

from dataclasses import dataclassfrom typing import Literalfrom langchain.agents import create_agentfrom langchain.agents.structured_output import ToolStrategy@dataclassclass ProductReview:    """Analysis of a product review."""    rating: int | None  # The rating of the product (1-5)    sentiment: Literal["positive", "negative"]  # The sentiment of the review    key_points: list[str]  # The key points of the reviewagent = create_agent(    model="gpt-5",    tools=tools,    response_format=ToolStrategy(ProductReview))result = agent.invoke({    "messages": [{"role": "user", "content": "Analyze this review: 'Great product: 5 out of 5 stars. Fast shipping, but expensive'"}]})result["structured_response"]# ProductReview(rating=5, sentiment='positive', key_points=['fast shipping', 'expensive'])

TypedDict:类型化字典类

from typing import Literalfrom typing_extensions import TypedDictfrom langchain.agents import create_agentfrom langchain.agents.structured_output import ToolStrategyclass ProductReview(TypedDict):    """Analysis of a product review."""    rating: int | None  # The rating of the product (1-5)    sentiment: Literal["positive", "negative"]  # The sentiment of the review    key_points: list[str]  # The key points of the reviewagent = create_agent(    model="gpt-5",    tools=tools,    response_format=ToolStrategy(ProductReview))result = agent.invoke({    "messages": [{"role": "user", "content": "Analyze this review: 'Great product: 5 out of 5 stars. Fast shipping, but expensive'"}]})result["structured_response"]# {'rating': 5, 'sentiment': 'positive', 'key_points': ['fast shipping', 'expensive']}

JSON Schema：具有JSON模式规范的字典

from langchain.agents import create_agentfrom langchain.agents.structured_output import ToolStrategyproduct_review_schema = {    "type": "object",    "description": "Analysis of a product review.",    "properties": {        "rating": {            "type": ["integer", "null"],            "description": "The rating of the product (1-5)",            "minimum": 1,            "maximum": 5        },        "sentiment": {            "type": "string",            "enum": ["positive", "negative"],            "description": "The sentiment of the review"        },        "key_points": {            "type": "array",            "items": {"type": "string"},            "description": "The key points of the review"        }    },    "required": ["sentiment", "key_points"]}agent = create_agent(    model="gpt-5",    tools=tools,    response_format=ToolStrategy(product_review_schema))result = agent.invoke({    "messages": [{"role": "user", "content": "Analyze this review: 'Great product: 5 out of 5 stars. Fast shipping, but expensive'"}]})result["structured_response"]# {'rating': 5, 'sentiment': 'positive', 'key_points': ['fast shipping', 'expensive']}

联合类型：多种模式选项。模型将根据上下文选择最合适的模式。

from pydantic import BaseModel, Fieldfrom typing import Literal, Unionfrom langchain.agents import create_agentfrom langchain.agents.structured_output import ToolStrategyclass ProductReview(BaseModel):    """Analysis of a product review."""    rating: int | None = Field(description="The rating of the product", ge=1, le=5)    sentiment: Literal["positive", "negative"] = Field(description="The sentiment of the review")    key_points: list[str] = Field(description="The key points of the review. Lowercase, 1-3 words each.")class CustomerComplaint(BaseModel):    """A customer complaint about a product or service."""    issue_type: Literal["product", "service", "shipping", "billing"] = Field(description="The type of issue")    severity: Literal["low", "medium", "high"] = Field(description="The severity of the complaint")    description: str = Field(description="Brief description of the complaint")agent = create_agent(    model="gpt-5",    tools=tools,    response_format=ToolStrategy(Union[ProductReview, CustomerComplaint]))result = agent.invoke({    "messages": [{"role": "user", "content": "Analyze this review: 'Great product: 5 out of 5 stars. Fast shipping, but expensive'"}]})result["structured_response"]# ProductReview(rating=5, sentiment='positive', key_points=['fast shipping', 'expensive'])

自定义工具消息内容(Custom tool message content)

tool_message_content参数允许您自定义生成结构化输出时出现在对话历史记录中的消息：

from pydantic import BaseModel, Fieldfrom typing import Literalfrom langchain.agents import create_agentfrom langchain.agents.structured_output import ToolStrategyclass MeetingAction(BaseModel):    """Action items extracted from a meeting transcript."""    task: str = Field(description="The specific task to be completed")    assignee: str = Field(description="Person responsible for the task")    priority: Literal["low", "medium", "high"] = Field(description="Priority level")agent = create_agent(    model="gpt-5",    tools=[],    response_format=ToolStrategy(        schema=MeetingAction,        tool_message_content="Action item captured and added to meeting notes!"    ))agent.invoke({    "messages": [{"role": "user", "content": "From our meeting: Sarah needs to update the project timeline as soon as possible"}]})

响应信息

================================ Human Message =================================From our meeting: Sarah needs to update the project timeline as soon as possible================================== Ai Message ==================================Tool Calls:  MeetingAction (call_1) Call ID: call_1  Args:    task: Update the project timeline    assignee: Sarah    priority: high================================= Tool Message =================================Name: MeetingActionAction item captured and added to meeting notes!

如果没有tool_message_content，我们的最终ToolMessage将是：

================================= Tool Message =================================Name: MeetingActionReturning structured response: {'task': 'update the project timeline', 'assignee': 'Sarah', 'priority': 'high'}

异常处理(Error handling)

模型在通过工具调用生成结构化输出时可能会出错。LangChain提供了智能重试机制来自动处理这些错误。

多重结构化输出错误(Multiple structured outputs error)

当模型错误地调用多个结构化输出工具时，代理会在ToolMessage中提供错误反馈，并提示模型重试：

from pydantic import BaseModel, Fieldfrom typing import Unionfrom langchain.agents import create_agentfrom langchain.agents.structured_output import ToolStrategyclass ContactInfo(BaseModel):    name: str = Field(description="Person's name")    email: str = Field(description="Email address")class EventDetails(BaseModel):    event_name: str = Field(description="Name of the event")    date: str = Field(description="Event date")agent = create_agent(    model="gpt-5",    tools=[],    response_format=ToolStrategy(Union[ContactInfo, EventDetails])  # Default: handle_errors=True)agent.invoke({    "messages": [{"role": "user", "content": "Extract info: John Doe (john@email.com) is organizing Tech Conference on March 15th"}]})

响应信息

================================ Human Message =================================Extract info: John Doe (john@email.com) is organizing Tech Conference on March 15thNone================================== Ai Message ==================================Tool Calls:  ContactInfo (call_1) Call ID: call_1  Args:    name: John Doe    email: john@email.com  EventDetails (call_2) Call ID: call_2  Args:    event_name: Tech Conference    date: March 15th================================= Tool Message =================================Name: ContactInfoError: Model incorrectly returned multiple structured responses (ContactInfo, EventDetails) when only one is expected. Please fix your mistakes.================================= Tool Message =================================Name: EventDetailsError: Model incorrectly returned multiple structured responses (ContactInfo, EventDetails) when only one is expected. Please fix your mistakes.================================== Ai Message ==================================Tool Calls:  ContactInfo (call_3) Call ID: call_3  Args:    name: John Doe    email: john@email.com================================= Tool Message =================================Name: ContactInfoReturning structured response: {'name': 'John Doe', 'email': 'john@email.com'}

架构验证错误(Schema validation error)

当结构化输出与预期模式不匹配时，代理会提供特定的错误反馈：

from pydantic import BaseModel, Fieldfrom langchain.agents import create_agentfrom langchain.agents.structured_output import ToolStrategyclass ProductRating(BaseModel):    rating: int | None = Field(description="Rating from 1-5", ge=1, le=5)    comment: str = Field(description="Review comment")agent = create_agent(    model="gpt-5",    tools=[],    response_format=ToolStrategy(ProductRating),  # Default: handle_errors=True    system_prompt="You are a helpful assistant that parses product reviews. Do not make any field or value up.")agent.invoke({    "messages": [{"role": "user", "content": "Parse this: Amazing product, 10/10!"}]})

响应信息

================================ Human Message =================================Parse this: Amazing product, 10/10!================================== Ai Message ==================================Tool Calls:  ProductRating (call_1) Call ID: call_1  Args:    rating: 10    comment: Amazing product================================= Tool Message =================================Name: ProductRatingError: Failed to parse structured output for tool 'ProductRating': 1 validation error for ProductRating.rating  Input should be less than or equal to 5 [type=less_than_equal, input_value=10, input_type=int]. Please fix your mistakes.================================== Ai Message ==================================Tool Calls:  ProductRating (call_2) Call ID: call_2  Args:    rating: 5    comment: Amazing product================================= Tool Message =================================Name: ProductRatingReturning structured response: {'rating': 5, 'comment': 'Amazing product'}

错误处理策略

您可以使用handle_errors参数自定义错误的处理方式：
自定义错误消息：

ToolStrategy(    schema=ProductRating,    handle_errors="Please provide a valid rating between 1-5 and include a comment.")

如果handle_errors是一个字符串，代理程序将始终提示模型使用固定的工具消息重试：

================================= Tool Message =================================Name: ProductRatingPlease provide a valid rating between 1-5 and include a comment.

仅处理特定异常：

ToolStrategy(    schema=ProductRating,    handle_errors=ValueError  # Only retry on ValueError, raise others)

如果handle_errors是异常类型，则代理将仅在引发的异常是指定类型时重试（使用默认错误消息）。在所有其他情况下，将提出例外情况。
处理多种异常类型：

ToolStrategy(    schema=ProductRating,    handle_errors=(ValueError, TypeError)  # Retry on ValueError and TypeError)

如果handle_errors是一个异常元组，则代理将仅在引发的异常是指定类型之一时重试（使用默认错误消息）。在所有其他情况下，将提出例外情况。
自定义错误处理函数：

from langchain.agents.structured_output import StructuredOutputValidationErrorfrom langchain.agents.structured_output import MultipleStructuredOutputsErrordef custom_error_handler(error: Exception) -> str:    if isinstance(error, StructuredOutputValidationError):        return "There was an issue with the format. Try again.    elif isinstance(error, MultipleStructuredOutputsError):        return "Multiple structured outputs were returned. Pick the most relevant one."    else:        return f"Error: {str(error)}"agent = create_agent(    model="gpt-5",    tools=[],    response_format=ToolStrategy(                        schema=Union[ContactInfo, EventDetails],                        handle_errors=custom_error_handler                    )  # Default: handle_errors=True)result = agent.invoke({    "messages": [{"role": "user", "content": "Extract info: John Doe (john@email.com) is organizing Tech Conference on March 15th"}]})for msg in result['messages']:    # If message is actually a ToolMessage object (not a dict), check its class name    if type(msg).__name__ == "ToolMessage":        print(msg.content)    # If message is a dictionary or you want a fallback    elif isinstance(msg, dict) and msg.get('tool_call_id'):        print(msg['content'])

在 StructuredOutputValidationError 响应:

================================= Tool Message =================================Name: ToolStrategyThere was an issue with the format. Try again.

在 MultipleStructuredOutputsError 响应:

================================= Tool Message =================================Name: ToolStrategyMultiple structured outputs were returned. Pick the most relevant one.

在 other errors 响应:

================================= Tool Message =================================Name: ToolStrategyError: <errormessage>

无错误处理:

response_format = ToolStrategy(    schema=ProductRating,    handle_errors=False  # All errors raised)