结果
结果是运行代理返回的最终值。结果值被封装在 AgentRunResult 和 StreamedRunResult 中，因此您可以访问其他数据，例如运行的 使用情况 和 消息历史记录
RunResult 和 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 验证器中不方便或不可能完成，特别是当验证需要 IO 并且是异步的时。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#10748 中添加到 Pydantic。
- 当接收到响应时，我们不知道它是否是最终响应，除非开始流式传输它并查看内容。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 中使用流式响应