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

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()stream_text()/stream_output()增量文本或逐块校验过的结构化数据聊天回复、Markdown 渲染、实时表格首选最省事run_stream()event_stream_handler回调运行过程中的事件工具调用、分片变化需要在最终输出前展示正在调用什么工具需要过程可视化时run_stream_events()完整事件流直到最终AgentRunResultEvent复杂多工具编排、需要自己拼装输出只有前两者不够用时才上一句话建议只要输出就是最终答案用run_stream()只要你需要看到工具执行的完整生命周期用run_stream_events()docs/agent.md。最小可行示例6 行代码跑通流式文本下面这段代码创建一个 Agent以流式方式提问并逐块打印回答保存为脚本后python stream_demo.py即可运行需要配置好对应模型的 API keyimport 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.mdstream_text()每次 yield 的是截至当前时刻的完整文本所以直接print会看到文本越长越长这正是流式效果。如果想要打字机式的逐 token 增量把上一行换成async for chunk in result.stream_text(deltaTrue): print(chunk, end, flushTrue)deltaTrue时每次只给新产生的片段适合前端逐字上屏但注意此时结果校验器不会被调用pydantic_ai_slim/pydantic_ai/result.py。进阶技巧4 个让流更稳的参数技巧 1用debounce_by控制刷新频率适用条件你输出的是结构化数据表格、JSON块到达非常频繁。做法stream_output(debounce_by0.1)会在时间窗口内合并数据块再统一做 Pydantic 校验默认 0.1 秒设为None则来一块处理一块。官方明确说这对长结构化输出能明显降低校验开销pydantic_ai_slim/pydantic_ai/result.py。技巧 2deltaTrue配合前端渲染适用条件界面按增量追加内容WebSocket 推给浏览器、终端打字机。做法用stream_text(deltaTrue)但要知道它绕过了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()模型响应会以stateinterrupted记入历史你可以决定丢弃还是保留这段半成品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_partialTrue做部分校验没验过的块会被静默跳过不是断流。解法这是设计行为只需以最后一次 yield完整校验为准别用中间态做持久化pydantic_ai_slim/pydantic_ai/result.py。现象取消流之后复用消息历史行为诡异。原因被中断的响应在历史里标记为stateinterrupted且带着半截内容。解法复用历史前检查该状态自行决定保留、丢弃还是重新生成docs/agent.md。真实案例复盘stream_whales 与天气 Agent 的设计取舍仓库 examples 里有两类值得拆解的示例。第一类是 examples/pydantic_ai_examples/stream_whales.py它用output_typelist[Whale]让 Agent 输出鲸鱼数据结构然后stream_output(debounce_by0.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.pyAgent 要先调get_lat_long再调get_weather才能答出气温。它没有用裸的run_stream()而是叠加event_stream_handler回调——工具开始执行、参数到达、结果返回时分别向界面发通知最终文本仍走流式输出。这个取舍说明了一个判断标准输出即答案的用第一层过程本身值得展示的用第二层若还要驱动多轮工具编排才升级到run_stream_events()。要点回顾与学习路径带走三条run_stream()返回StreamedRunResultstream_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创作声明：本文部分内容由AI辅助生成（AIGC），仅供参考