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 AIPython 的 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, flushTrue) # 每收到一段就刷一次屏跑起来后你会看到回答一个字一个字往外冒而不是最后整段蹦出。这里有两个值得留意的细节一是async with块结束时流才真正关闭别提前跳出二是循环结束后还能拿到result.usage查询本次运行的 token 用量和费用。上面示例的完整实现含多模型切换和 rich 渲染见 examples/pydantic_ai_examples/stream_markdown.py。3种流式输出方式按场景选方式1原始文本流——stream_text(deltaTrue)逐字增量stream_output()每次给你的是到目前为止的完整文本适合整块重绘界面。但如果你做的是打字机效果每次只想要新增的那几个字用stream_text并打开deltaTrueasync with agent.run_stream(写一篇短文) as result: async for text in result.stream_text(deltaTrue): console.print(text, end) # text 只是增量部分两者是同一底层流的两套视图deltaFalse默认给全量快照deltaTrue给增量片段。选哪种取决于你的前端是整块替换还是末尾追加。方式2结构化数据流——边生成边校验的实时表格这是 Pydantic AI 流式处理最有价值的能力。你在创建 Agent 时用output_type声明一个 Pydantic 模型或 TypedDict之后stream_output()每次 yield 的不再是字符串而是当前已能解析出来的、部分验证通过的数据。仓库里的鲸鱼示例演示了这个玩法模型生成 5 种鲸鱼的数据Rich 表格随着字段逐个到位而实时刷新——class Whale(TypedDict): name: str length: Annotated[float, Field(descriptionAverage length of an adult whale in meters.)] weight: NotRequired[Annotated[float, Field(ge50)]] # 可选字段可约束最小值 ocean: NotRequired[str] agent Agent(openai:gpt-5.2, output_typelist[Whale]) async with agent.run_stream(Generate me details of 5 species of Whale.) as result: async for whales in result.stream_output(debounce_by0.01): render_table(whales) # 每次拿到一个部分填好的列表未到位的字段显示省略号debounce_by0.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_handlerevent_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创作声明：本文部分内容由AI辅助生成（AIGC），仅供参考 Pydantic AI run_stream 流式输出指南：3条路径从首帧响应到结构化实时校验
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 AIPython 的 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, flushTrue) # 每收到一段就刷一次屏跑起来后你会看到回答一个字一个字往外冒而不是最后整段蹦出。这里有两个值得留意的细节一是async with块结束时流才真正关闭别提前跳出二是循环结束后还能拿到result.usage查询本次运行的 token 用量和费用。上面示例的完整实现含多模型切换和 rich 渲染见 examples/pydantic_ai_examples/stream_markdown.py。3种流式输出方式按场景选方式1原始文本流——stream_text(deltaTrue)逐字增量stream_output()每次给你的是到目前为止的完整文本适合整块重绘界面。但如果你做的是打字机效果每次只想要新增的那几个字用stream_text并打开deltaTrueasync with agent.run_stream(写一篇短文) as result: async for text in result.stream_text(deltaTrue): console.print(text, end) # text 只是增量部分两者是同一底层流的两套视图deltaFalse默认给全量快照deltaTrue给增量片段。选哪种取决于你的前端是整块替换还是末尾追加。方式2结构化数据流——边生成边校验的实时表格这是 Pydantic AI 流式处理最有价值的能力。你在创建 Agent 时用output_type声明一个 Pydantic 模型或 TypedDict之后stream_output()每次 yield 的不再是字符串而是当前已能解析出来的、部分验证通过的数据。仓库里的鲸鱼示例演示了这个玩法模型生成 5 种鲸鱼的数据Rich 表格随着字段逐个到位而实时刷新——class Whale(TypedDict): name: str length: Annotated[float, Field(descriptionAverage length of an adult whale in meters.)] weight: NotRequired[Annotated[float, Field(ge50)]] # 可选字段可约束最小值 ocean: NotRequired[str] agent Agent(openai:gpt-5.2, output_typelist[Whale]) async with agent.run_stream(Generate me details of 5 species of Whale.) as result: async for whales in result.stream_output(debounce_by0.01): render_table(whales) # 每次拿到一个部分填好的列表未到位的字段显示省略号debounce_by0.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_handlerevent_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创作声明：本文部分内容由AI辅助生成（AIGC），仅供参考