乐于分享
好东西不私藏

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

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

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 LiteralUnionfrom 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 =================================NameMeetingActionReturning 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.comis 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, EventDetailswhen only one is expected. Please fix your mistakes.================================= Tool Message =================================Name: EventDetailsError: Model incorrectly returned multiple structured responses (ContactInfo, EventDetailswhen 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(errorException) -> 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 =================================NameToolStrategyError<errormessage>
  • 无错误处理:
response_format = ToolStrategy(    schema=ProductRating,    handle_errors=False  # All errors raised)