Pydantic AI 流式输出完整指南：如何用 run_stream 让 AI 边生成边响应-二趣网

Pydantic AI 流式输出完整指南：如何用 run_stream 让 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 的流式输出（streaming）让 AI 的回答随生成进程实时刷新，而不是憋到最后一整段吐给你。如果你在做聊天界面、实时仪表板，或者任何对首字延迟敏感的应用，run_stream()就是入口——它让响应在生成的第一秒就开始可见。

解决什么问题

两个最典型的场景：

聊天界面里，用户盯着空白光标等十几秒，体验直接崩掉；流式输出让首字延迟几乎降到零。
Agent 要生成几百行的 Pydantic 结构化数据，agent.run()得等完整 JSON 传完、解析、校验才能返回，中间过程完全不可见。

两个问题的根源都一样：非流式调用是"全有或全无"。流式模式把结果拆成不断刷新的小快照，边到边消费。

它是怎么工作的

把run_stream()想成打开一场直播。

模型每吐出一个片段，框架就把"目前为止的完整画面"递给你一张快照，而不是一张张小碎片。debounce_by控制递快照的节奏：默认攒 0.1 秒再交一张，传None则每个片段到达立刻交。

两条消费管道对应两种数据：stream_text()交的是累积文本（每次都是从头到当前点的全文），stream_output()交的是经过 Pydantic 校验的结构化数据。中间帧校验失败时框架直接跳过该帧继续等，流结束时再做一次严格终验——所以你看到的每一帧都相对可靠。

快速上手：从 0 到 1 的 4 步

配置环境变量（如OPENAI_API_KEY），并安装：pip install pydantic-ai
克隆示例仓库看可运行代码：git clone https://gitcode.com/GitHub_Trending/py/pydantic-ai
先跑通文本流式 demo：uv run -m pydantic_ai_examples.stream_markdown，终端里会用 rich 实时渲染 markdown
换成结构化数据时参考uv run -m pydantic_ai_examples.stream_whales：给 Agent 声明output_type，把stream_text()换成stream_output()即可

核心调用只有这几行，文本流式：

agent = Agent('openai:gpt-5.2') async with agent.run_stream('Where does "hello world" come from?') as result: async for message in result.stream_text(): print(message) # 每次是累积全文，随生成增长

示例逻辑参考：docs/output.md

结构化流式则把最后一行换成result.stream_output(debounce_by=0.01)，每帧拿到一个已经过 Pydantic 校验的部分结果，最后一帧是终验后的完整数据：

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): live.update(render_table(whales)) # 表格随生成逐步填满

来源文件：examples/pydantic_ai_examples/stream_whales.py

常见场景怎么选

场景	推荐用法	注意点
聊天 / 终端实时文本	`run_stream`+`stream_text()`	每帧是累积全文；要增量片段就传`delta=True`
结构化数据（Pydantic 模型 / TypedDict）	`run_stream`+`stream_output(debounce_by=...)`	中间帧可能缺字段，`debounce_by=None`会放大校验开销
终端渲染 markdown	`stream_text()`+ rich 的`Live`实时重绘	参考示例 examples/pydantic_ai_examples/stream_markdown.py
带副作用的落库 / 通知	输出函数里用`ctx.partial_output`判断	部分输出时跳过副作用，只对终帧执行

踩坑清单

问题	原因	解法
`stream_text()`抛`UserError`	该流方法只支持纯文本输出	声明了结构化`output_type`时改用`stream_output()`
中间帧突然消失、表格空了	部分输出未通过 Pydantic 校验，该帧被跳过	属正常行为，终帧保证完整；给可选字段留默认值更稳
`debounce_by=None`后卡顿	每帧都触发一次完整校验，长结构化响应下开销大	回到默认`0.1`，或用`0.01`之类的小值权衡
`delta=True`后对话历史里没结果	增量模式不会把结果拼成完整字符串存入历史	需要历史记录时用非 delta 模式
用户不想等了想掐断	流还挂着	调用`result.cancel()`停止本地消费并请求远端关闭

小结

Pydantic AI 的流式能力就三件事：run_stream()开流、stream_text()和stream_output()两条管道、debounce_by控制刷新节奏。机制细节见官方文档流式输出，管道实现在 pydantic_ai_slim/pydantic_ai/result.py。

创作声明：本文部分内容由AI辅助生成（AIGC），仅供参考

企业官网建设流程全解析