企业官网建设流程全解析

Pydantic AI run_stream 流式输出指南：3条路径从首帧响应到结构化实时校验

【免费下载链接】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（Python 的 AI Agent 框架）里写过聊天功能，多半体验过这样的等待：模型要生成完整的几十字甚至几百字，界面才一次性刷出来，用户以为卡死了。这篇文章帮你解决这个问题——用run_stream让模型边生成边输出，把首字延迟从"整段等完"压到"逐字到达"，同时讲清楚结构化数据（表格、列表）怎么在流式过程中逐行校验，以及卡住时的排查思路。

跑起来之前：装包和配好一个 API Key

准备工作分三步：先安装框架，再配一个模型服务的密钥，最后确认你在异步环境里。安装用pip install pydantic-ai或uv add pydantic-ai都行；密钥以最常用的 OpenAI 为例，设置OPENAI_API_KEY环境变量即可（用 Gemini、Groq 等就设对应变量，如GEMINI_API_KEY）。第三点容易忽略：run_stream是异步 API，所以你的入口代码要在asyncio.run(...)里执行，仓库里的示例全部以uv run -m pydantic_ai_examples.模块名方式启动，你可以照着跑。

最小可运行示例：10行代码拿到流式文本

下面这段代码演示了流式输出的最小组合：agent.run_stream打开一个流式运行，result.stream_output()每产生一段文本就 yield 一次，你只管渲染。它来自仓库自带的 Markdown 流式示例，把live.update(Markdown(message))换成你自己的打印或前端推送逻辑就能直接用。

agent = Agent() # 默认模型，或用 Agent('openai:gpt-5-mini') 指定 async with agent.run_stream('Show me a short example of using Pydantic.') as result: async for message in result.stream_output(): print(message, end='', flush=True) # 每收到一段就刷一次屏

跑起来后你会看到回答一个字一个字往外冒，而不是最后整段蹦出。这里有两个值得留意的细节：一是async with块结束时流才真正关闭，别提前跳出；二是循环结束后还能拿到result.usage，查询本次运行的 token 用量和费用。

上面示例的完整实现（含多模型切换和 rich 渲染）见 examples/pydantic_ai_examples/stream_markdown.py。

3种流式输出方式按场景选

方式1：原始文本流——stream_text(delta=True)逐字增量

stream_output()每次给你的是"到目前为止的完整文本"，适合整块重绘界面。但如果你做的是打字机效果，每次只想要"新增的那几个字"，用stream_text并打开delta=True：

async with agent.run_stream('写一篇短文') as result: async for text in result.stream_text(delta=True): console.print(text, end='') # text 只是增量部分

两者是同一底层流的两套视图：delta=False（默认）给全量快照，delta=True给增量片段。选哪种取决于你的前端是"整块替换"还是"末尾追加"。

方式2：结构化数据流——边生成边校验的实时表格

这是 Pydantic AI 流式处理最有价值的能力。你在创建 Agent 时用output_type声明一个 Pydantic 模型或 TypedDict，之后stream_output()每次 yield 的不再是字符串，而是当前已能解析出来的、部分验证通过的数据。仓库里的鲸鱼示例演示了这个玩法：模型生成 5 种鲸鱼的数据，Rich 表格随着字段逐个到位而实时刷新——

class Whale(TypedDict): name: str length: Annotated[float, Field(description='Average length of an adult whale in meters.')] weight: NotRequired[Annotated[float, Field(ge=50)]] # 可选字段，可约束最小值 ocean: NotRequired[str] agent = Agent('openai:gpt-5.2', output_type=list[Whale]) async with agent.run_stream('Generate me details of 5 species of Whale.') as result: async for whales in result.stream_output(debounce_by=0.01): render_table(whales) # 每次拿到一个"部分填好的列表"，未到位的字段显示省略号

debounce_by=0.01是节流间隔：10 毫秒内的多次更新合并成一次，避免网络快抖时你的渲染层被刷爆。想进一步减少刷新次数可以调大到 0.1（默认值）。这段机制的完整示例在 examples/pydantic_ai_examples/stream_whales.py，配合真实 Agent 的完整玩法可以看 天气 Agent 示例。

方式3：工具调用事件流——event_stream_handler看全过程

前两种只覆盖"最终回答"这一段，但 Agent 真正跑起来时中间还会调工具（查天气、查数据库）。run_stream接受一个event_stream_handler参数，把工具调用参数、工具返回结果、最终结果开始等每一步事件实时推给你。下面这个片段演示了如何按事件类型分发（精简自官方文档中的天气 Agent 演示）：

async def event_stream_handler(ctx, event_stream): async for event in event_stream: if isinstance(event, FunctionToolCallEvent): print(f'模型要调工具: {event.part.tool_name}({event.part.args})') elif isinstance(event, FunctionToolResultEvent): print(f'工具返回: {event.part.content}') elif isinstance(event, FinalResultEvent): print('模型开始输出最终结果') async with weather_agent.run_stream(prompt, event_stream_handler=event_stream_handler) as run: async for output in run.stream_text(): print(output, end='') # 事件和最终文本可以并行消费

事件类型（FunctionToolCallEvent、PartDeltaEvent等）的完整定义见 messages 模块，文档里有逐事件打印的完整日志样例（docs/agent.md 的 Streaming 章节）。

流式"不对劲"时的3步排查

🔧 先对照这三步，能覆盖绝大多数卡点。

第一步，界面抖得厉害或刷新太频繁：调debounce_by。这是节流阀，单位秒。默认 0.1 秒，调大则刷新更"迟钝"但省资源；调小（如 0.01）则更跟手，适合终端表格这类轻量渲染。

第二步，中间某次 yield 的数据不完整甚至校验失败：这是正常现象，不是 bug。流式校验分两层——中间快照用宽松的部分验证（允许字段还没到齐），只有流结束后的最后一个 yield 才是完整严格校验的结果。所以写代码时别把中间项当最终结果入库，以最后一次 yield 为准。这个"先部分验证、最后严格验证"的逻辑就在 pydantic_ai_slim/pydantic_ai/result.py 的stream_output里，想确认行为可以直接读源码。

第三步，流干脆不来数据或报错：先查模型，再查网络。并非所有模型都支持流式或全部输出类型（比如图像输出），Agent 会按模型能力做检查，不支持时抛UserError提示；网络波动导致的临时错误用 Pydantic AI 内置的 retries 机制兜底，把 Agent 配成自动重试几次即可。

收尾：接下来看哪里

• 想补全流式 API 的每个参数：docs/agent.md 的 Streaming 章节
• 想抄完整可运行案例：examples/pydantic_ai_examples/ 下的stream_markdown.py、stream_whales.py、weather_agent.py
• 想把结构化输出刷得更快（工具调用期间也提前出数据）：docs/output.md 的"Making structured responses appear faster"一节
• 关心流式过程的 token 用量统计：docs/usage.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