Pydantic AI 流式处理让加载转圈消失的选型实操【免费下载链接】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 流式处理解决这个问题用 run_stream 打开流把文本、结构化数据和工具调用进度逐块推给前端。本文讲四种流式输出怎么选、怎么用以及上线前容易踩的坑。为什么转圈流式最小示例 普通的 run() 要等回答完全生成才返回长文本意味着长时间空白。流式的机制其实一句话run_stream() 是一个异步上下文管理器模型每生成一点新内容就从流里 yield 一块——像水龙头匀速放水而不是一次性倒满桶。from pydantic_ai import Agent agent Agent(openai:gpt-5.2) async def main(): async with agent.run_stream(为什么天空是蓝色的) as result: async for text in result.stream_text(): print(text) # 当前为止的完整文本这段演示了 run_stream 用法的基础形态stream_text() 默认每次返回当前为止的完整文本传 deltaTrue 则只拿新增片段流结束时上下文管理器自动收尾。同步场景还有 run_stream_sync()流式方法完全相同。仓库里 examples/pydantic_ai_examples/stream_markdown.py 是这个思路的可运行版本用 rich 把 Markdown 实时渲染出来。四种流怎么选Pydantic AI 的流式输出有四个入口分别对应不同场景流类型入口拿到什么典型场景文本流result.stream_text()文本块完整或增量聊天回复、代码生成结构化输出流result.stream_output()已部分校验的结构化对象仪表盘、表格、表单实时渲染工具调用流agent.run_stream_events()完整的中间事件序列进度展示、审计日志多模态流output_type 声明多模态类型文本与图像混合内容图文报告、卡片生成结构化输出流是结构化输出流式验证里最巧的一环。声明Agent(openai:gpt-5.2, output_typelist[Whale])后数据没生成完就能拿到已校验的半成品async with agent.run_stream(列出 5 种鲸鱼) as result: async for whales in result.stream_output(debounce_by0.01): render_table(whales)stream_output() 每轮用部分校验模式过一遍 schema字段完整且合法的立刻吐出来中间不合法的批次被静默跳过最终结果再做一次严格校验。debounce_by 控制两次校验间的合并间隔None 表示不合并输出越长合并越省开销。examples/pydantic_ai_examples/stream_whales.py 就是拿 0.01 秒的间隔实时刷新鲸鱼数据表格。工具调用流要理解一个边界run_stream() 把第一个匹配 output_type 的输出当作最终结果模型在这之后又生成的工具调用默认不会执行end_strategy 设为 graceful 或 exhaustive 才例外。所以要看清工具执行过程该用 run_stream_events() 收事件序列它以一个携带最终结果的 AgentRunResultEvent 收尾官方说明见 docs/agent.md。多模态流则和文本流同套路在 output_type 里声明含图像的模型先确认目标模型支持图像输出不支持就退回非流式。上线前自查清单五个问题 ✅Demo 五分钟能跑通生产环境要过的关是这五条部分结果当最终数据用了stream_output() 给的是校验中快照落库、发邮件这类动作只该发生在流结束后拿到的最终结果上。重试与降级有路吗输出校验失败走retries{output: N}预算让模型改工具调用走retries{tools: N}模型整体不可用时用 FallbackModel 切备用模型兜底。各层重试如何叠加、预算怎么算docs/retries.md 拆得很细。内存与连接管住了吗长流的 debounce_by 保持合理值避免每个 token 都触发一次校验HTTP 连接保持复用别在流式上下文里囤积大对象资源释放交给上下文管理器。模型兼容性测过吗各模型对流式输出、结构化输出的支持程度并不一致换模型前先用真实输出类型冒烟不兼容的场景退回完整 run()别硬上流式。流式模式下的工具调用没丢吧记住 end_strategy 那条边界依赖工具执行完再让模型总结的流程不要建在 run_stream() 上改用 run_stream_events() 或调 end_strategy。案例搭一份实时数据报表场景运营打开报表页期望 agent 查完数据库、调完图表工具后报告的分节内容随生成随上屏而不是干等一整页。设计思路分四点双通道run_stream_events() 盯工具调用事件前端实时显示正在查销售表 / 正在渲染图表报告正文另用 stream_output() 消费。结构化输出当主体output_type 声明报告结构摘要、图表清单、结论每节出现就渲染一块。容错先行debounce_by 取 0.1 左右校验失败交给 output 重试预算消化重试耗尽再切 FallbackModel 或回退到上一份缓存报告。资源轻工具中间结果只展示不落盘只有流末的 AgentRunResultEvent 才入库。async with agent.run_stream_events(汇总本季度销售) as events: async for event in events: if isinstance(event, FunctionToolCallEvent): print(f工具调用: {event!r})这段是进度通道每冒出一个 FunctionToolCallEvent前端就知道 agent 正走到哪一步。把文本流给对话、结构化流给报表、事件流给进度条、多模态流给图文混排选型对了大部分卡顿和怪问题就消失了。想继续深挖从 docs/output.md 的输出类型与流式验证、docs/retries.md 的重试分层以及 examples/pydantic_ai_examples/ 下的示例代码入手。【免费下载链接】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），仅供参考