Pydantic AI 流式输出：从首个 token 到完整校验的 4 步实践

Pydantic AI 流式输出：从首个 token 到完整校验的 4 步实践

【免费下载链接】pydantic-aiHow Python does AI. Agents, realtime voice, image generation, embeddings. Every model, every interface, typed end to end.项目地址: https://gitcode.com/GitHub_Trending/py/pydantic-ai

Pydantic AI 的agent.run_stream()是流式输出的入口：模型输出不再等整段生成完才返回，而是逐块推送到你的业务层，首个 token 就能先到前端。装好包之后不用改任何安装配置，直接看下面四步。

一个场景：整段等待和逐块返回的差别

假设你在做一个聊天界面，用户提问后要干等 5 秒，然后一大段文字一次性砸到屏幕上。换成流式推送后，界面上文字是"长出来"的，体感延迟明显下降。代价是你要处理两件事：中间状态的输出可能不合法（半截 JSON 永远校验不过），以及什么时候能拿到"最终版"。Pydantic AI 的流式输出把这两件事封装成了"部分校验 + 最终校验"两段式：中间块用宽松校验先推，结束那一刻再做一次严格校验。

四步跑通 run_stream 的流式输出

第一步：用 run_stream 拿到流式结果对象

这段代码在做什么：用异步上下文管理器启动流式运行，通过stream_text()逐块打印文本，最后用get_output()拿完整结果和用量。注意stream_text()默认delta=False，每次 yield 的是"到目前为止的全文"，不是增量片段。

from pydantic_ai import Agent agent = Agent('openai:gpt-5.2') async def main(): async with agent.run_stream('What is the capital of the UK?') as response: async for text in response.stream_text(): print(text) print(await response.get_output())

示例来源：docs/agent.md；get_output()的实现在pydantic_ai_slim/pydantic_ai/result.py

第二步：结构化数据也能"流"出来

这段代码在做什么：定义一个TypedDict作为输出类型，模型边生成边被校验，每凑出合法的部分就 yield 一次，前端表格随数据逐行"长"出来。stream_output()内部对中间块用allow_partial=True做校验，校验失败的块直接跳过，最后一定会再 yield 一次完整校验过的结果。

class Whale(TypedDict): name: str length: Annotated[float, Field(description='Average length in meters.')] weight: NotRequired[Annotated[float, Field(..., ge=50)]] agent = Agent('openai:gpt-5.2', output_type=list[Whale]) async def main(): async with agent.run_stream('Details of 5 species of Whale.') as result: async for whales in result.stream_output(debounce_by=0.01): render_table(whales) # your own Rich/HTML rendering

示例来源：examples/pydantic_ai_examples/stream_whales.py

第三步：想看到中间事件就接 event_stream_handler

这段代码在做什么：给run_stream()传一个event_stream_handler，在最终输出产生之前观察工具调用、thinking、文本增量等事件。完整的事件类型清单（PartStartEvent、PartDeltaEvent、FunctionToolCallEvent、FinalResultEvent等）见pydantic_ai_slim/pydantic_ai/messages.py。

async def event_stream_handler(ctx, event_stream): async for event in event_stream: if isinstance(event, FunctionToolCallEvent): print(f'Tool call: {event.part.tool_name}') elif isinstance(event, FinalResultEvent): print('Final result started') async with agent.run_stream(prompt, event_stream_handler=event_stream_handler) as run: async for text in run.stream_text(): print(text)

事件处理示例来源：docs/agent.md

这里有个容易误会的点：run_stream()把第一个匹配输出类型的结果当最终输出，模型在"最终输出"之后生成的工具调用默认不会被执行。如果你的 Agent 必须把工具全部跑完，改用agent.run_stream_events()或agent.iter()，或者把end_strategy设为'graceful'/'exhaustive'。

第四步：收尾时拿完整输出、消息历史和用量

response属性随时能拿当前响应快照（流式中state为'incomplete'，结束后为'complete'）；usage属性在流结束后才有完整 cost；想中途放弃就调cancel()，它只停当前这条模型响应，整个 run 还在继续。这几个属性都在同一个StreamedRunResult对象上，见pydantic_ai_slim/pydantic_ai/result.py。

避坑清单：症状 → 原因 → 修复

debounce_by：流式输出校验频率怎么选

debounce_by控制"合并多少秒内的块再校验/推送一次"，默认0.1秒，传None表示每个块都触发。它本质是用一点延迟换校验次数：结构化输出越长，默认 0.1 秒省下的校验开销越明显（stream_text()的 docstring 里也专门提了这一点）。

如果接了 logfire 之类的观测，可以直接看每次运行的追踪图确认请求、工具调用和输出的时序：

跑通examples/pydantic_ai_examples/stream_whales.py之后，把debounce_by从0.01改成None，对比表格刷新频率和终端的校验日志量，你会直观感受到这两个值的差别。更多参数语义见 docs/agent.md 的 Running Agents 一节和 docs/output.md 的结构化输出部分。

【免费下载链接】pydantic-aiHow Python does AI. Agents, realtime voice, image generation, embeddings. Every model, every interface, typed end to end.项目地址: https://gitcode.com/GitHub_Trending/py/pydantic-ai