结果
结果是从 运行代理 返回的最终值。 结果值被包装在 AgentRunResult 和 StreamedRunResult 中，因此您可以访问其他数据，如 运行的使用情况 和 消息历史
Both RunResult and StreamedRunResult 是通用的，包装它们的数据，因此保留了代理返回的数据的类型信息。
olympics.py
from pydantic import BaseModel
from pydantic_ai import Agent
class CityLocation(BaseModel):
city: str
country: str
agent = Agent('google-gla:gemini-1.5-flash', result_type=CityLocation)
result = agent.run_sync('Where were the olympics held in 2012?')
print(result.data)
#> city='London' country='United Kingdom'
print(result.usage())
"""
Usage(requests=1, request_tokens=57, response_tokens=8, total_tokens=65, details=None)
"""
(这个例子是完整的，可以“原样”运行)
当收到纯文本响应或模型调用与其中一种结构化结果类型相关联的工具时，运行结束。我们将添加限制以确保运行不会无限进行，请参见 #70。
结果数据
当结果类型为 str，或包含 str 的联合类型时，模型启用了纯文本响应，模型的原始文本响应被用作响应数据。
如果结果类型是一个包含多个成员的联合类型（在从成员中移除str后），每个成员都作为一个单独的工具与模型注册，以减少工具模式的复杂性并最大化模型正确响应的机会。
如果结果类型模式不是 "object" 类型，结果类型将被包装在一个单一元素对象中，因此与模型注册的所有工具的模式都是对象模式。
结构化结果（如工具）使用Pydantic构建用于该工具的JSON模式，并验证模型返回的数据。
带上PEP-747
在PEP-747 "注解类型形式" 发布之前，联合类型在Python中不是有效的type。
创建代理时，我们需要# type: ignore result_type 参数，并添加类型提示以告知类型检查器有关代理的类型。
这是返回文本或结构化值的一个例子
box_or_error.py
from typing import Union
from pydantic import BaseModel
from pydantic_ai import Agent
class Box(BaseModel):
width: int
height: int
depth: int
units: str
agent: Agent[None, Union[Box, str]] = Agent(
'openai:gpt-4o-mini',
result_type=Union[Box, str], # type: ignore
system_prompt=(
"Extract me the dimensions of a box, "
"if you can't extract all data, ask the user to try again."
),
)
result = agent.run_sync('The box is 10x20x30')
print(result.data)
#> Please provide the units for the dimensions (e.g., cm, in, m).
result = agent.run_sync('The box is 10x20x30 cm')
print(result.data)
#> width=10 height=20 depth=30 units='cm'
(这个例子是完整的，可以“原样”运行)
这是一个使用联合返回类型的示例，注册了多个工具，并将非对象模式包装在一个对象中：
colors_or_sizes.py
from typing import Union
from pydantic_ai import Agent
agent: Agent[None, Union[list[str], list[int]]] = Agent(
'openai:gpt-4o-mini',
result_type=Union[list[str], list[int]], # type: ignore
system_prompt='Extract either colors or sizes from the shapes provided.',
)
result = agent.run_sync('red square, blue circle, green triangle')
print(result.data)
#> ['red', 'blue', 'green']
result = agent.run_sync('square size 10, circle size 20, triangle size 30')
print(result.data)
#> [10, 20, 30]
(这个例子是完整的，可以“原样”运行)
结果验证器函数
在Pydantic验证器中，有些验证是不方便或不可能进行的，特别是当验证需要输入输出并且是异步的。 PydanticAI提供了一种通过agent.result_validator装饰器添加验证函数的方法。
这是SQL生成示例的简化版本：
sql_gen.py
from typing import Union
from fake_database import DatabaseConn, QueryError
from pydantic import BaseModel
from pydantic_ai import Agent, RunContext, ModelRetry
class Success(BaseModel):
sql_query: str
class InvalidRequest(BaseModel):
error_message: str
Response = Union[Success, InvalidRequest]
agent: Agent[DatabaseConn, Response] = Agent(
'google-gla:gemini-1.5-flash',
result_type=Response, # type: ignore
deps_type=DatabaseConn,
system_prompt='Generate PostgreSQL flavored SQL queries based on user input.',
)
@agent.result_validator
async def validate_result(ctx: RunContext[DatabaseConn], result: Response) -> Response:
if isinstance(result, InvalidRequest):
return result
try:
await ctx.deps.execute(f'EXPLAIN {result.sql_query}')
except QueryError as e:
raise ModelRetry(f'Invalid query: {e}') from e
else:
return result
result = agent.run_sync(
'get me users who were last active yesterday.', deps=DatabaseConn()
)
print(result.data)
#> sql_query='SELECT * FROM users WHERE last_active::date = today() - interval 1 day'
(这个例子是完整的，可以“原样”运行)
流结果
流式结果有两个主要挑战：
- 在结构化响应完成之前进行验证，这通过最近添加到Pydantic中的“部分验证”实现，详见 pydantic/pydantic#10748。
- 在接收到响应时，我们无法知道这是否是最终响应，除非开始流式传输并查看内容。 PydanticAI 仅流式传输足够的响应以判断它是工具调用还是结果，然后流式传输整个内容并调用工具，或将流作为
StreamedRunResult返回。
流式文本
流式文本结果示例：
streamed_hello_world.py
from pydantic_ai import Agent
agent = Agent('google-gla:gemini-1.5-flash') # (1)!
async def main():
async with agent.run_stream('Where does "hello world" come from?') as result: # (2)!
async for message in result.stream_text(): # (3)!
print(message)
#> The first known
#> The first known use of "hello,
#> The first known use of "hello, world" was in
#> The first known use of "hello, world" was in a 1974 textbook
#> The first known use of "hello, world" was in a 1974 textbook about the C
#> The first known use of "hello, world" was in a 1974 textbook about the C programming language.
- 流式处理与标准
Agent类一起工作，无需任何特殊设置，只需一个支持流式处理的模型（目前所有模型都支持流式处理）。 - 该
Agent.run_stream()方法用于启动一个流式运行，该方法返回一个上下文管理器，以便在流完成时可以关闭连接。 - 由
StreamedRunResult.stream_text()产生的每个项目都是完整的文本响应，随着新数据的接收而扩展。
(这个例子是完整的，可以“直接运行”——你需要添加 asyncio.run(main()) 来运行 main)
我们还可以将文本作为增量流而不是每个项中的整个文本：
streamed_delta_hello_world.py
from pydantic_ai import Agent
agent = Agent('google-gla:gemini-1.5-flash')
async def main():
async with agent.run_stream('Where does "hello world" come from?') as result:
async for message in result.stream_text(delta=True): # (1)!
print(message)
#> The first known
#> use of "hello,
#> world" was in
#> a 1974 textbook
#> about the C
#> programming language.
stream_text如果响应不是文本，将会出错
(这个例子是完整的，可以“直接运行”——你需要添加 asyncio.run(main()) 来运行 main)
结果消息未包含在 messages 中
如果您使用 .stream_text(delta=True)，最终结果消息将 不会 添加到结果消息中，详见 消息和聊天记录 以获取更多信息。
流式结构化响应
并非所有类型在Pydantic中都支持部分验证，请参阅 pydantic/pydantic#10748，一般而言，对于类似模型的结构，目前最好使用 TypeDict。
以下是构建使用档案的流式传输示例：
streamed_user_profile.py
from datetime import date
from typing_extensions import TypedDict
from pydantic_ai import Agent
class UserProfile(TypedDict, total=False):
name: str
dob: date
bio: str
agent = Agent(
'openai:gpt-4o',
result_type=UserProfile,
system_prompt='Extract a user profile from the input',
)
async def main():
user_input = 'My name is Ben, I was born on January 28th 1990, I like the chain the dog and the pyramid.'
async with agent.run_stream(user_input) as result:
async for profile in result.stream():
print(profile)
#> {'name': 'Ben'}
#> {'name': 'Ben'}
#> {'name': 'Ben', 'dob': date(1990, 1, 28), 'bio': 'Likes'}
#> {'name': 'Ben', 'dob': date(1990, 1, 28), 'bio': 'Likes the chain the '}
#> {'name': 'Ben', 'dob': date(1990, 1, 28), 'bio': 'Likes the chain the dog and the pyr'}
#> {'name': 'Ben', 'dob': date(1990, 1, 28), 'bio': 'Likes the chain the dog and the pyramid'}
#> {'name': 'Ben', 'dob': date(1990, 1, 28), 'bio': 'Likes the chain the dog and the pyramid'}
(这个例子是完整的，可以“直接运行”——你需要添加 asyncio.run(main()) 来运行 main)
如果您想要对验证进行细致的控制，特别是捕获验证错误，您可以使用以下模式：
streamed_user_profile.py
from datetime import date
from pydantic import ValidationError
from typing_extensions import TypedDict
from pydantic_ai import Agent
class UserProfile(TypedDict, total=False):
name: str
dob: date
bio: str
agent = Agent('openai:gpt-4o', result_type=UserProfile)
async def main():
user_input = 'My name is Ben, I was born on January 28th 1990, I like the chain the dog and the pyramid.'
async with agent.run_stream(user_input) as result:
async for message, last in result.stream_structured(debounce_by=0.01): # (1)!
try:
profile = await result.validate_structured_result( # (2)!
message,
allow_partial=not last,
)
except ValidationError:
continue
print(profile)
#> {'name': 'Ben'}
#> {'name': 'Ben'}
#> {'name': 'Ben', 'dob': date(1990, 1, 28), 'bio': 'Likes'}
#> {'name': 'Ben', 'dob': date(1990, 1, 28), 'bio': 'Likes the chain the '}
#> {'name': 'Ben', 'dob': date(1990, 1, 28), 'bio': 'Likes the chain the dog and the pyr'}
#> {'name': 'Ben', 'dob': date(1990, 1, 28), 'bio': 'Likes the chain the dog and the pyramid'}
#> {'name': 'Ben', 'dob': date(1990, 1, 28), 'bio': 'Likes the chain the dog and the pyramid'}
stream_structured将数据作为ModelResponse对象流式传输，因此迭代不会因ValidationError而失败。validate_structured_result验证数据，allow_partial=True启用 pydantic 的experimental_allow_partial标志在TypeAdapter上。
(这个例子是完整的，可以“直接运行”——你需要添加 asyncio.run(main()) 来运行 main)
示例
以下示例演示了如何在PydanticAI中使用流式响应：