返回的形式:
JSON objects, Pydantic models,or dataclasses that your application can use directly
LangChain’s create_agent 会自动处理结构化输出
用户设置desired 结构化输出的schema
返回在 in the key strucured_responese key of the agent’s state
Field函数的作用:
结构化输出时,是创建元数据的重要途径,且其内部的description可以作为Prompt的一部分指导LLM的工作
1
2
3
4
5
6
7
8
9
| def create_agent(
...
response_format: Union[
ToolStrategy[StructuredResponseT],
ProviderStrategy[StructuredResponseT],
type[StructuredResponseT],
None,
]
)
|
使用response_format来控制agent返回结构化数据的方式
ToolStrategy[StructuredResponseT]: Uses tool calling for structured outputProviderStrategy[StructuredResponseT]: Uses provider-native structured outputtype[StructuredResponseT]: Schema type - automatically selects best strategy based on model capabilitiesNone: Structured output not explicitly requested
When a schema type is provided directly, LangChain automatically chooses:
Provider strategy#
一些provider支持原生的结构化输出
这是最可靠的方式
要使用这种方式,请配置 Providerstrategy
1
2
3
| class ProviderStrategy(Generic[SchemaT]):
schema: type[SchemaT]
strict: bool | None = None
|
参数
schema(required):
规定了结构化输出的format,取值有:
- Pydantic models:
BaseModel subclasses with field validation. Returns validated Pydantic instance. - Dataclasses: Python dataclasses with type annotations. Returns dict.
- TypedDict: Typed dictionary classes. Returns dict.
- JSON Schema: Dictionary with JSON schema specification. Must include top-level
title and description keys. Returns dict.
strict:
可选;默认None;控制是否严格遵守schema
当你直接把schema type传给create_agent.response_format 时且模型支持原生的结构化输出时,LangChain会自动选择ProviderStrategy
对于模型不支持原生的结构化输出,LangChain使用tool calling,这对于所有的支持tool calling的模型都适用
使用:配置 ToolStrategy
1
2
3
4
5
6
7
8
9
10
| 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],
]
|
schema: required
同上
tool_message_content:
生成结构化输出时的tool message的自定义内容
如果没有提供,为显示结构化响应的响应数据
handle_errors:
默认为True
当结构化输出的验证失败的时候的错误处理策略,参数如下
True: Catch all errors with default error templatestr: Catch all errors with this custom messagetype[Exception]: Only catch this exception type with default messagetuple[type[Exception], ...]: Only catch these exception types with default messageCallable[[Exception], str]: Custom function that returns error messageFalse: No retry, let exceptions propagate
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
| from pydantic import BaseModel, Field
from typing import Literal, Union
from langchain.agents import create_agent
from langchain.agents.structured_output import ToolStrategy
class 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.5",
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#
the tool_message_content 参数允许你自定义结构化输出产生时呈现在会话历史中的message
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
| from pydantic import BaseModel, Field
from typing import Literal, Union
from langchain.agents import create_agent
from langchain.agents.structured_output import ToolStrategy
class 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.5",
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'])
|
Error handing#
LangChain提供了智能重试机制
Multiple structured outputs error#
当模型错误地调用了多个结构化工具输出,会在ToolMessage中生成相应的反馈和提示词以便模型进行重试
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
| from pydantic import BaseModel, Field
from typing import Union
from langchain.agents import create_agent
from langchain.agents.structured_output import ToolStrategy
class 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.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"}]
})
|
Schema validation error#
当结构化输出和schema不匹配时,agent会返回特定的错误反馈
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
| from pydantic import BaseModel, Field
from langchain.agents import create_agent
from langchain.agents.structured_output import ToolStrategy
class 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.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!"}]
})
|
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
| ================================ 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: ProductRating
Error: 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: ProductRating
Returning structured response: {'rating': 5, 'comment': 'Amazing product'}
|
Error handling strategies#
你可以通过handle_errors参数自定义错误处理的方式
Custom error message:
1
2
3
4
| ToolStrategy(
schema=ProductRating,
handle_errors="Please provide a valid rating between 1-5 and include a comment."
)
|
If handle_errors is a string, the agent will always prompt the model to re-try with a fixed tool message:
1
2
3
4
| ================================= Tool Message =================================
Name: ProductRating
Please provide a valid rating between 1-5 and include a comment.
|
Handle specific exceptions only:
1
2
3
4
| ToolStrategy(
schema=ProductRating,
handle_errors=ValueError # Only retry on ValueError, raise others
)
|
If handle_errors is an exception type, the agent will only retry (using the default error message) if the exception raised is the specified type. In all other cases, the exception will be raised.Handle multiple exception types:
1
2
3
4
| ToolStrategy(
schema=ProductRating,
handle_errors=(ValueError, TypeError) # Retry on ValueError and TypeError
)
|
If handle_errors is a tuple of exceptions, the agent will only retry (using the default error message) if the exception raised is one of the specified types. In all other cases, the exception will be raised.Custom error handler function:
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
| from langchain.agents.structured_output import StructuredOutputValidationError
from langchain.agents.structured_output import MultipleStructuredOutputsError
def 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.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'])
|
On StructuredOutputValidationError:
1
2
3
4
| ================================= Tool Message =================================
Name: ToolStrategy
There was an issue with the format. Try again.
|
On MultipleStructuredOutputsError:
1
2
3
4
| ================================= Tool Message =================================
Name: ToolStrategy
Multiple structured outputs were returned. Pick the most relevant one.
|
On other errors:
1
2
3
4
| ================================= Tool Message =================================
Name: ToolStrategy
Error: <error message>
|
No error handling:
1
2
3
4
| response_format = ToolStrategy(
schema=ProductRating,
handle_errors=False # All errors raised
)
|