输出

“输出”指的是运行代理后返回的最终值。这可以是纯文本、结构化数据，或者是使用模型提供的参数调用的函数的结果。

输出被包装在 AgentRunResult 或 StreamedRunResult 中，以便您可以访问其他数据，例如运行的使用情况和消息历史。

AgentRunResult 和 StreamedRunResult 在它们包装的数据方面都是泛型的，因此代理返回的数据的类型信息得以保留。

当模型以一种结构化输出类型响应时，或者在未指定输出类型或 str 是允许的选项之一时，当接收到纯文本响应时，运行结束。如果超出使用限制，运行也可能被取消，请参见使用限制。

这是一个使用 Pydantic 模型作为 output_type 的示例，强制模型响应与我们的规范相匹配的数据。

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', output_type=CityLocation)
result = agent.run_sync('Where were the olympics held in 2012?')
print(result.output)
#> city='London' country='United Kingdom'
print(result.usage())
#> RunUsage(input_tokens=57, output_tokens=8, requests=1)

（这个例子是完整的，可以“按原样”运行）

输出数据

Agent 类的构造函数接受一个 output_type 参数，该参数接受一个或多个类型或输出函数。它支持简单的标量类型、列表和字典类型（包括 TypedDict 和StructuredDict）、数据类和 Pydantic 模型，以及类型联合——通常是 Pydantic 模型中作为类型提示支持的所有内容。您也可以传递一个包含多个选项的列表。

默认情况下，Pydantic AI 利用模型的工具调用能力使其返回结构化数据。当指定了多个输出类型时（在联合或列表中），每个成员都作为单独的输出工具注册到模型中，以降低 schema 的复杂性并最大化模型正确响应的机会。这已被证明在各种模型中都表现良好。如果您想更改输出工具的名称，使用模型的原生结构化输出功能，或者在其指令中将输出 schema 传递给模型，您可以使用输出模式标记类。

当未指定输出类型，或者 str 是输出类型之一时，来自模型的任何纯文本响应都将用作输出数据。如果 str 不在输出类型中，模型将被强制返回结构化数据或调用输出函数。

如果输出类型的 schema 类型不是 "object"（例如，它是 int 或 list[int]），则输出类型会被包装在一个单元素对象中，这样所有注册到模型中的工具的 schema 都是对象 schema。

结构化输出（如工具）使用 Pydantic 来构建用于工具的 JSON schema，并验证模型返回的数据。

类型检查注意事项

Agent 类在其输出类型上是泛型的，并且此类型会传递到 AgentRunResult.output 和 StreamedRunResult.output，以便您的 IDE 或静态类型检查器可以在您的代码未能正确考虑这些输出可能具有的所有值时向您发出警告。

像 pyright 和 mypy 这样的静态类型检查器会尽力从您指定的 output_type 推断代理的输出类型，但当您提供函数或在联合或列表中提供多种类型时，它们并不总能正确地做到这一点，尽管 Pydantic AI 会正常工作。当这种情况发生时，即使您确信自己传递了有效的 output_type，您的类型检查器也会报错，您需要通过在 Agent 构造函数上显式指定泛型参数来帮助类型检查器。这在下面的第二个示例和更下方的输出函数示例中有所展示。

具体来说，在以下三种有效使用 output_type 的情况下，您需要这样做：

1. 当使用类型的联合时，例如 output_type=Foo | Bar。在 PEP-747 "Annotating Type Forms" 于 Python 3.15 中落地之前，类型检查器不认为这是一个有效的 output_type 值。除了在 Agent 构造函数上使用泛型参数外，您还需要在将联合传递给 output_type 的那一行添加 # type: ignore。或者，您可以使用列表：output_type=[Foo, Bar]。
2. 对于 mypy：当使用列表时，作为联合的功能等效替代方案，或者因为您正在传入输出函数。Pyright 能够正确处理这种情况，我们已经向 mypy 提交了一个问题，希望能解决这个问题。
3. 对于 mypy：当使用异步输出函数时。Pyright 能够正确处理这种情况，我们已经向 mypy 提交了一个问题，希望能解决这个问题。

这是一个返回文本或结构化数据的示例：

box_or_error.py

from pydantic import BaseModel

from pydantic_ai import Agent


class Box(BaseModel):
    width: int
    height: int
    depth: int
    units: str


agent = Agent(
    'openai:gpt-4o-mini',
    output_type=[Box, str], # (1)!
    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.output)
#> Please provide the units for the dimensions (e.g., cm, in, m).

result = agent.run_sync('The box is 10x20x30 cm')
print(result.output)
#> width=10 height=20 depth=30 units='cm'

1. 这也可以是一个联合：output_type=Box | str。然而，如上文“类型检查注意事项”部分所述，为了能被正确地进行类型检查，这将需要显式指定 Agent 构造函数的泛型参数，并在此行添加 # type: ignore。

（这个例子是完整的，可以“按原样”运行）

这是一个使用联合返回类型的示例，它将注册多个输出工具并将非对象 schema 包装在对象中：

colors_or_sizes.py

from pydantic_ai import Agent

agent = Agent[None, list[str] | list[int]](
    'openai:gpt-4o-mini',
    output_type=list[str] | list[int],  # type: ignore # (1)!
    system_prompt='Extract either colors or sizes from the shapes provided.',
)

result = agent.run_sync('red square, blue circle, green triangle')
print(result.output)
#> ['red', 'blue', 'green']

result = agent.run_sync('square size 10, circle size 20, triangle size 30')
print(result.output)
#> [10, 20, 30]

1. 如上文“类型检查注意事项”部分所述，使用联合而不是列表，需要显式指定 Agent 构造函数的泛型参数，并在此行添加 # type: ignore 以便能被正确地进行类型检查。

（这个例子是完整的，可以“按原样”运行）

输出函数

您可能希望代理运行的输出不是纯文本或结构化数据，而是使用模型提供的参数调用的函数的结果，例如，进一步处理或验证通过参数提供的数据（并可以选择告知模型重试），或将其移交给另一个代理。

输出函数类似于函数工具，但模型被强制调用其中之一，该调用结束代理运行，并且结果不会传回模型。

与工具函数一样，模型提供的输出函数参数使用 Pydantic 进行验证，它们可以选择接受 RunContext 作为第一个参数，并且可以引发 ModelRetry 来请求模型使用修改后的参数（或不同的输出类型）重试。

要指定输出函数，您可以将代理的 output_type 设置为单个函数（或绑定的实例方法），或一个函数列表。该列表还可以包含其他输出类型，如简单标量或整个 Pydantic 模型。通常您不希望同时将输出函数注册为工具（使用 @agent.tool 装饰器或 tools 参数），因为这可能会使模型混淆应该调用哪一个。

以下是所有这些功能的一个示例：

output_functions.py

import re

from pydantic import BaseModel

from pydantic_ai import Agent, ModelRetry, RunContext, UnexpectedModelBehavior


class Row(BaseModel):
    name: str
    country: str


tables = {
    'capital_cities': [
        Row(name='Amsterdam', country='Netherlands'),
        Row(name='Mexico City', country='Mexico'),
    ]
}


class SQLFailure(BaseModel):
    """An unrecoverable failure. Only use this when you can't change the query to make it work."""

    explanation: str


def run_sql_query(query: str) -> list[Row]:
    """Run a SQL query on the database."""

    select_table = re.match(r'SELECT (.+) FROM (\w+)', query)
    if select_table:
        column_names = select_table.group(1)
        if column_names != '*':
            raise ModelRetry("Only 'SELECT *' is supported, you'll have to do column filtering manually.")

        table_name = select_table.group(2)
        if table_name not in tables:
            raise ModelRetry(
                f"Unknown table '{table_name}' in query '{query}'. Available tables: {', '.join(tables.keys())}."
            )

        return tables[table_name]

    raise ModelRetry(f"Unsupported query: '{query}'.")


sql_agent = Agent[None, list[Row] | SQLFailure](
    'openai:gpt-4o',
    output_type=[run_sql_query, SQLFailure],
    instructions='You are a SQL agent that can run SQL queries on a database.',
)


async def hand_off_to_sql_agent(ctx: RunContext, query: str) -> list[Row]:
    """I take natural language queries, turn them into SQL, and run them on a database."""

    # Drop the final message with the output tool call, as it shouldn't be passed on to the SQL agent
    messages = ctx.messages[:-1]
    try:
        result = await sql_agent.run(query, message_history=messages)
        output = result.output
        if isinstance(output, SQLFailure):
            raise ModelRetry(f'SQL agent failed: {output.explanation}')
        return output
    except UnexpectedModelBehavior as e:
        # Bubble up potentially retryable errors to the router agent
        if (cause := e.__cause__) and isinstance(cause, ModelRetry):
            raise ModelRetry(f'SQL agent failed: {cause.message}') from e
        else:
            raise


class RouterFailure(BaseModel):
    """Use me when no appropriate agent is found or the used agent failed."""

    explanation: str


router_agent = Agent[None, list[Row] | RouterFailure](
    'openai:gpt-4o',
    output_type=[hand_off_to_sql_agent, RouterFailure],
    instructions='You are a router to other agents. Never try to solve a problem yourself, just pass it on.',
)

result = router_agent.run_sync('Select the names and countries of all capitals')
print(result.output)
"""
[
    Row(name='Amsterdam', country='Netherlands'),
    Row(name='Mexico City', country='Mexico'),
]
"""

result = router_agent.run_sync('Select all pets')
print(repr(result.output))
"""
RouterFailure(explanation="The requested table 'pets' does not exist in the database. The only available table is 'capital_cities', which does not contain data about pets.")
"""

result = router_agent.run_sync('How do I fly from Amsterdam to Mexico City?')
print(repr(result.output))
"""
RouterFailure(explanation='I am not equipped to provide travel information, such as flights from Amsterdam to Mexico City.')
"""

文本输出

如果您提供一个接受字符串的输出函数，Pydantic AI 默认会像对待任何其他输出函数一样创建一个输出工具。如果您希望模型使用纯文本输出来提供该字符串，您可以将该函数包装在 TextOutput 标记类中。如果需要，此标记类可以与一个或多个 ToolOutput 标记类（或未标记的类型或函数）一起在提供给 output_type 的列表中使用。

text_output_function.py

from pydantic_ai import Agent, TextOutput


def split_into_words(text: str) -> list[str]:
    return text.split()


agent = Agent(
    'openai:gpt-4o',
    output_type=TextOutput(split_into_words),
)
result = agent.run_sync('Who was Albert Einstein?')
print(result.output)
#> ['Albert', 'Einstein', 'was', 'a', 'German-born', 'theoretical', 'physicist.']

（这个例子是完整的，可以“按原样”运行）

输出模式

Pydantic AI 实现了三种不同的方法来让模型输出结构化数据：

1. 工具输出，其中使用工具调用来生成输出。
2. 原生输出，其中要求模型生成符合所提供 JSON schema 的文本内容。
3. 提示输出，其中将包含所需 JSON schema 的提示注入到模型指令中，然后我们尝试将模型的纯文本响应解析为适当的格式。

工具输出

在默认的工具输出模式下，每个输出类型（或函数）的输出 JSON schema 会作为特殊输出工具的参数 schema 提供给模型。这是默认设置，因为它几乎被所有模型支持，并且已被证明效果很好。

如果您想更改输出工具的名称，传递自定义描述以辅助模型，或者开启或关闭严格模式，您可以将类型包装在 ToolOutput 标记类中并提供相应的参数。请注意，默认情况下，描述取自 Pydantic 模型或输出函数上指定的文档字符串，因此通常不需要使用标记类来指定它。

要在代理运行期间动态修改或过滤可用的输出工具，您可以定义一个代理范围的 prepare_output_tools 函数，该函数将在每次运行步骤之前被调用。此函数应为 ToolsPrepareFunc 类型，它接受 RunContext 和一个 ToolDefinition 列表，并返回一个新的工具定义列表（或返回 None 以禁用该步骤的所有工具）。这类似于用于非输出工具的 prepare_tools 函数。

tool_output.py

from pydantic import BaseModel

from pydantic_ai import Agent, ToolOutput


class Fruit(BaseModel):
    name: str
    color: str


class Vehicle(BaseModel):
    name: str
    wheels: int


agent = Agent(
    'openai:gpt-4o',
    output_type=[ # (1)!
        ToolOutput(Fruit, name='return_fruit'),
        ToolOutput(Vehicle, name='return_vehicle'),
    ],
)
result = agent.run_sync('What is a banana?')
print(repr(result.output))
#> Fruit(name='banana', color='yellow')

1. 如果我们只传递 Fruit 和 Vehicle 而不使用自定义工具名称，我们可以使用联合：output_type=Fruit | Vehicle。然而，由于 ToolOutput 是一个对象而不是一个类型，我们必须使用列表。

（这个例子是完整的，可以“按原样”运行）

原生输出

原生输出模式使用模型的原生“结构化输出”功能（也称为“JSON Schema 响应格式”），其中模型被强制只输出与所提供的 JSON schema 匹配的文本。请注意，并非所有模型都支持此功能，有时还附带限制。例如，Anthropic 完全不支持此功能，而 Gemini 不能同时使用工具和结构化输出，尝试这样做会导致错误。

要使用此模式，您可以将输出类型包装在 NativeOutput 标记类中，如果类型或函数的名称和文档字符串不足，该类还允许您指定 name 和 description。

native_output.py

from pydantic_ai import Agent, NativeOutput

from tool_output import Fruit, Vehicle

agent = Agent(
    'openai:gpt-4o',
    output_type=NativeOutput(
        [Fruit, Vehicle], # (1)!
        name='Fruit_or_vehicle',
        description='Return a fruit or vehicle.'
    ),
)
result = agent.run_sync('What is a Ford Explorer?')
print(repr(result.output))
#> Vehicle(name='Ford Explorer', wheels=4)

1. 这也可以是一个联合：output_type=Fruit | Vehicle。然而，如上文“类型检查注意事项”部分所述，为了能被正确地进行类型检查，这将需要显式指定 Agent 构造函数的泛型参数，并在此行添加 # type: ignore。

（这个例子是完整的，可以“按原样”运行）

提示输出

在这种模式下，模型通过其指令被提示输出与所提供的 JSON schema 匹配的文本，由模型来正确解释这些指令。这适用于所有模型，但通常是可靠性最低的方法，因为模型并非被强制匹配 schema。

虽然我们通常建议从工具输出或原生输出开始，但在某些情况下，此模式可能会产生更高质量的输出，并且对于没有原生工具调用或结构化输出支持的模型，这是生成结构化输出的唯一选择。

如果模型 API 支持“JSON 模式”功能（也称为“JSON 对象响应格式”）以强制模型输出有效的 JSON，该功能会被启用，但仍由模型来遵守 schema。Pydantic AI 将验证返回的结构化数据，并在验证失败时告知模型重试，但如果模型不够智能，这可能还不够。

要使用此模式，您可以将输出类型包装在 PromptedOutput 标记类中，如果类型或函数的名称和文档字符串不足，该类还允许您指定 name 和 description。此外，它支持一个 template 参数，让您可以指定一个自定义的指令模板来代替默认模板。

prompted_output.py

from pydantic import BaseModel

from pydantic_ai import Agent, PromptedOutput

from tool_output import Vehicle


class Device(BaseModel):
    name: str
    kind: str


agent = Agent(
    'openai:gpt-4o',
    output_type=PromptedOutput(
        [Vehicle, Device], # (1)!
        name='Vehicle or device',
        description='Return a vehicle or device.'
    ),
)
result = agent.run_sync('What is a MacBook?')
print(repr(result.output))
#> Device(name='MacBook', kind='laptop')

agent = Agent(
    'openai:gpt-4o',
    output_type=PromptedOutput(
        [Vehicle, Device],
        template='Gimme some JSON: {schema}'
    ),
)
result = agent.run_sync('What is a Ford Explorer?')
print(repr(result.output))
#> Vehicle(name='Ford Explorer', wheels=4)

1. 这也可以是一个联合：output_type=Vehicle | Device。然而，如上文“类型检查注意事项”部分所述，为了能被正确地进行类型检查，这将需要显式指定 Agent 构造函数的泛型参数，并在此行添加 # type: ignore。

（这个例子是完整的，可以“按原样”运行）

自定义 JSON schema

如果使用 Pydantic BaseModel、dataclass 或 TypedDict 来定义您期望的结构化输出对象不可行，例如当您从外部源获取 JSON schema 或动态生成它时，您可以使用 StructuredDict() 辅助函数来生成一个带有附加 JSON schema 的 dict[str, Any] 子类，Pydantic AI 会将该 schema 传递给模型。

请注意，Pydantic AI 不会对接收到的 JSON 对象执行任何验证，模型需要正确解释 schema 及其中的任何约束，如必需字段或整数值范围。

输出类型将是 dict[str, Any]，您的代码需要以防御性的方式从中读取数据，以防模型出错。您可以使用输出验证器将验证错误反馈给模型，让其重试。

除了 JSON schema，您还可以选择性地传递 name 和 description 参数，为模型提供额外的上下文。

from pydantic_ai import Agent, StructuredDict

HumanDict = StructuredDict(
    {
        'type': 'object',
        'properties': {
            'name': {'type': 'string'},
            'age': {'type': 'integer'}
        },
        'required': ['name', 'age']
    },
    name='Human',
    description='A human with a name and age',
)

agent = Agent('openai:gpt-4o', output_type=HumanDict)
result = agent.run_sync('Create a person')
#> {'name': 'John Doe', 'age': 30}

输出验证器

有些验证在 Pydantic 验证器中实现起来不方便或不可能，特别是当验证需要 IO 并且是异步的时候。Pydantic AI 提供了一种通过 agent.output_validator 装饰器添加验证函数的方法。

如果您想为不同的输出类型实现独立的验证逻辑，建议改用输出函数，以避免在输出验证器内部进行 isinstance 检查。如果您希望模型输出纯文本，进行您自己的处理或验证，然后让代理的最终输出成为您函数的结果，建议使用带有 TextOutput 标记类的输出函数。

这是一个SQL 生成示例的简化版本：

sql_gen.py

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


Output = Success | InvalidRequest
agent = Agent[DatabaseConn, Output](
    'google-gla:gemini-1.5-flash',
    output_type=Output,  # type: ignore
    deps_type=DatabaseConn,
    system_prompt='Generate PostgreSQL flavored SQL queries based on user input.',
)


@agent.output_validator
async def validate_sql(ctx: RunContext[DatabaseConn], output: Output) -> Output:
    if isinstance(output, InvalidRequest):
        return output
    try:
        await ctx.deps.execute(f'EXPLAIN {output.sql_query}')
    except QueryError as e:
        raise ModelRetry(f'Invalid query: {e}') from e
    else:
        return output


result = agent.run_sync(
    'get me users who were last active yesterday.', deps=DatabaseConn()
)
print(result.output)
#> sql_query='SELECT * FROM users WHERE last_active::date = today() - interval 1 day'

（这个例子是完整的，可以“按原样”运行）

流式结果

流式结果有两个主要挑战：

1. 在结构化响应完成之前对其进行验证，这通过“部分验证”实现，该功能最近已在 pydantic/pydantic#10748 中添加到 Pydantic。
2. 当接收到响应时，我们不知道它是否是最终响应，除非开始流式传输并窥探其内容。Pydantic AI 只流式传输足够多的响应来嗅探出它是一个工具调用还是一个输出，然后流式传输整个内容并调用工具，或者将流作为 StreamedRunResult 返回。

注意

由于 run_stream() 方法会将第一个与 output_type 匹配的输出视为最终输出，它将停止运行代理图，并且不会执行模型在此“最终”输出之后进行的任何工具调用。

如果您希望始终将代理图运行到完成，并流式传输模型流式响应和代理执行工具的所有事件，请改用带有 event_stream_handler（文档）的 agent.run() 或 agent.iter()（文档）。

流式文本

流式文本输出示例：

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.

1. 流式传输与标准的 Agent 类兼容，并且不需要任何特殊设置，只需要一个支持流式传输的模型（目前所有模型都支持流式传输）。
2. Agent.run_stream() 方法用于启动流式运行，该方法返回一个上下文管理器，以便在流完成时可以关闭连接。
3. 由 StreamedRunResult.stream_text() 产生的每个项目都是完整的文本响应，随着新数据的接收而扩展。

（此示例是完整的，可以“按原样”运行——您需要添加 asyncio.run(main()) 来运行 main）

我们也可以将文本作为增量（delta）而不是每个项目中包含全部文本来流式传输：

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.

1. 如果响应不是文本，stream_text 将会报错。

（此示例是完整的，可以“按原样”运行——您需要添加 asyncio.run(main()) 来运行 main）

输出消息不包含在 messages 中

如果您使用 .stream_text(delta=True)，最终的输出消息将不会被添加到结果消息中，更多信息请参见消息和聊天历史。

流式结构化输出

这是一个在构建用户个人资料时进行流式传输的示例：

streamed_user_profile.py

from datetime import date

from typing_extensions import NotRequired, TypedDict

from pydantic_ai import Agent


class UserProfile(TypedDict):
    name: str
    dob: NotRequired[date]
    bio: NotRequired[str]


agent = Agent(
    'openai:gpt-4o',
    output_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_output():
            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）

如果您想要对验证进行精细控制，可以使用以下模式来获取完整的、部分的 ModelResponse：

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', output_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_responses(debounce_by=0.01):  # (1)!
            try:
                profile = await result.validate_response_output(  # (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'}

1. stream_responses 以 ModelResponse 对象的形式流式传输数据，因此迭代不会因 ValidationError 而失败。
2. validate_response_output 验证数据，allow_partial=True 启用了 pydantic 在 TypeAdapter 上的 experimental_allow_partial 标志。

（此示例是完整的，可以“按原样”运行——您需要添加 asyncio.run(main()) 来运行 main）

示例

以下示例演示了如何在 Pydantic AI 中使用流式响应：

• 流式 Markdown
• 流式传输鲸鱼