企业官网建设流程全解析

Pydantic AI 流式输出指南：如何从 run_stream 到事件流一步步上手

【免费下载链接】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

写 AI 聊天应用时，你是不是也干过这种事：用户问了一个问题，界面转圈十秒，最后"啪"地把整段答案甩出来——哪怕答案其实一个字一个字早就生成了。Pydantic AI 的流式输出（streaming）功能就是为这个场景准备的：模型每生成一小块内容，你的界面立刻拿到并刷新。本文带你用 1 段 6 行的最小代码跑通流式文本，掌握 4 个进阶技巧，并对照 4 条避坑清单，让实时响应这件事不再踩坑。

快速认识：run_stream 到底帮你做了什么

把agent.run()想象成点外卖：下单后骑手提着整份餐送到你手上，一次给齐。而agent.run_stream()更像现场档口——师傅每炒好一勺就递到你碗里，你边吃边等。它返回一个StreamedRunResult上下文管理器，你在里面用async for消费"到碗里"的内容（docs/agent.md）。

它解决的核心问题只有两个：一是首字节延迟，用户不必干等整段生成；二是可中断性，中途可以随时cancel()，而不是让模型白烧 token。官方列出的五种运行方式中，run_stream()和run_stream_events()是唯二支持流式消费的（pydantic_ai_slim/pydantic_ai/agent/abstract.py）。

场景选型：流文本、流结构化数据还是流事件

Pydantic AI 提供三层粒度，选错层会让代码平白变复杂：

一句话建议：只要输出就是最终答案，用run_stream()；只要你需要看到工具执行的完整生命周期，用run_stream_events()（docs/agent.md）。

最小可行示例：6 行代码跑通流式文本

下面这段代码创建一个 Agent，以流式方式提问并逐块打印回答，保存为脚本后python stream_demo.py即可运行（需要配置好对应模型的 API key）：

import asyncio 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 result: async for text in result.stream_text(): print(text) asyncio.run(main())

逐步说明：Agent(...)声明使用的模型；run_stream(...)是异步上下文管理器，进入时才真正发起请求（docs/agent.md）；stream_text()每次 yield 的是"截至当前时刻的完整文本"，所以直接print会看到文本越长越长，这正是流式效果。

如果想要"打字机"式的逐 token 增量，把上一行换成：

async for chunk in result.stream_text(delta=True): print(chunk, end='', flush=True)

delta=True时每次只给新产生的片段，适合前端逐字上屏，但注意此时结果校验器不会被调用（pydantic_ai_slim/pydantic_ai/result.py）。

进阶技巧：4 个让流更稳的参数

技巧 1：用debounce_by控制刷新频率

适用条件：你输出的是结构化数据（表格、JSON），块到达非常频繁。做法：stream_output(debounce_by=0.1)会在时间窗口内合并数据块再统一做 Pydantic 校验，默认 0.1 秒；设为None则来一块处理一块。官方明确说这对长结构化输出能明显降低校验开销（pydantic_ai_slim/pydantic_ai/result.py）。

技巧 2：delta=True配合前端渲染

适用条件：界面按增量追加内容（WebSocket 推给浏览器、终端打字机）。做法：用stream_text(delta=True)，但要知道它绕过了TextOutput转换函数和校验器，适合纯展示场景；需要变换逻辑时改用stream_output()。

技巧 3：工具调用场景换成run_stream_events()

适用条件：你的 Agent 挂了工具，且期望"文本 + 工具"混合出现时工具照跑。做法：run_stream()在流到第一个匹配输出类型的内容时就锁定为最终结果，此时同批出现的函数工具调用不会执行（默认end_strategy下）；改成run_stream_events()才能看到并驱动完整的工具执行事件链（docs/agent.md、docs/output.md）。

技巧 4：主动cancel()省 token

适用条件：用户点了"停止"按钮。做法：在消费流时提前break后调用await result.cancel()，模型响应会以state='interrupted'记入历史，你可以决定丢弃还是保留这段半成品（docs/agent.md）。

避坑清单：断流、静默跳过、悬挂工具调用

现象：给output_type配了TextOutput转换函数，stream_text()的输出却没有生效。原因：流式模式下stream_text()明确不应用TextOutput函数。解法：需要变换逻辑时用stream_output()（docs/output.md）。

现象：模型一边吐文字一边发起工具调用，工具却没执行，输出直接结束了。原因：run_stream()把第一个匹配output_type的输出立刻锁定为最终结果，悬挂的工具调用被跳过。解法：改用run_stream_events()，或把end_strategy设为'graceful'/'exhaustive'让工具执行（docs/agent.md）。

现象：流式过程中偶发"缺帧"，最终结果却是好的。原因：中间块以allow_partial=True做部分校验，没验过的块会被静默跳过，不是断流。解法：这是设计行为，只需以最后一次 yield（完整校验）为准，别用中间态做持久化（pydantic_ai_slim/pydantic_ai/result.py）。

现象：取消流之后复用消息历史，行为诡异。原因：被中断的响应在历史里标记为state='interrupted'，且带着半截内容。解法：复用历史前检查该状态，自行决定保留、丢弃还是重新生成（docs/agent.md）。

真实案例复盘：stream_whales 与天气 Agent 的设计取舍

仓库 examples 里有两类值得拆解的示例。第一类是 examples/pydantic_ai_examples/stream_whales.py：它用output_type=list[Whale]让 Agent 输出鲸鱼数据结构，然后stream_output(debounce_by=0.01)边到边渲染 Rich 表格。它的取舍有三处：防抖窗口压到 0.01 秒，为了表格"逐行长出来"的观感而多付一点校验开销；模型字段大量用NotRequired加Field(description=...)，这样流式中途缺字段时部分校验能顺利通过；对可选值统一用whale.get('weight')加省略号占位，UI 不为半成品数据结构做特殊分支。

第二类是天气 Agent 系列（examples/pydantic_ai_examples/weather_agent.py、examples/pydantic_ai_examples/weather_agent_gradio.py）：Agent 要先调get_lat_long再调get_weather才能答出气温。它没有用裸的run_stream()，而是叠加event_stream_handler回调——工具开始执行、参数到达、结果返回时分别向界面发通知，最终文本仍走流式输出。这个取舍说明了一个判断标准：输出即答案的用第一层，过程本身值得展示的用第二层；若还要驱动多轮工具编排，才升级到run_stream_events()。

要点回顾与学习路径

带走三条：run_stream()返回StreamedRunResult，stream_text()/stream_output()是它的两个消费口；中间产出都是部分校验的，只有最后一次 yield 才是完整结果；涉及工具生命周期时，直接上run_stream_events()，别硬用第一层。

延伸学习：入口文档 docs/agent.md（五种运行方式与取消语义）、docs/output.md（结构化输出的流式细节）；核心实现在 pydantic_ai_slim/pydantic_ai/result.py 的AgentStream类，想理解防抖与部分校验的实现，从它读起。

【免费下载链接】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