企业官网建设流程全解析

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() 默认每次返回"当前为止的完整文本"，传 delta=True 则只拿新增片段；流结束时上下文管理器自动收尾。同步场景还有 run_stream_sync()，流式方法完全相同。仓库里 examples/pydantic_ai_examples/stream_markdown.py 是这个思路的可运行版本，用 rich 把 Markdown 实时渲染出来。

四种流，怎么选

Pydantic AI 的流式输出有四个入口，分别对应不同场景：

结构化输出流是结构化输出流式验证里最巧的一环。声明Agent('openai:gpt-5.2', output_type=list[Whale])后，数据没生成完就能拿到"已校验"的半成品：

async with agent.run_stream('列出 5 种鲸鱼') as result: async for whales in result.stream_output(debounce_by=0.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 五分钟能跑通，生产环境要过的关是这五条：

1. 部分结果当最终数据用了？stream_output() 给的是"校验中快照"，落库、发邮件这类动作只该发生在流结束后拿到的最终结果上。
2. 重试与降级有路吗？输出校验失败走retries={'output': N}预算让模型改，工具调用走retries={'tools': N}，模型整体不可用时用 FallbackModel 切备用模型兜底。各层重试如何叠加、预算怎么算，docs/retries.md 拆得很细。
3. 内存与连接管住了吗？长流的 debounce_by 保持合理值，避免每个 token 都触发一次校验；HTTP 连接保持复用；别在流式上下文里囤积大对象，资源释放交给上下文管理器。
4. 模型兼容性测过吗？各模型对流式输出、结构化输出的支持程度并不一致，换模型前先用真实输出类型冒烟；不兼容的场景退回完整 run()，别硬上流式。
5. 流式模式下的工具调用没丢吧？记住 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