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-aiPydantic AI 是 Python 生态里的 AI 智能体框架它能让大模型的回答一边生成一边流式推送给你并顺手完成 Pydantic 结构化校验。本文适合会基础 Python、想做出打字机效果聊天界面或实时数据面板的开发者跟着走一遍就能把run_stream用熟。一、先跑起来10 行代码流式输出流式输出说白了就是模型每生成一小段文字立刻推送到你的屏幕而不是等全部生成完再一次性给你。Pydantic AI 用agent.run_stream()这个异步上下文管理器提供这条通道它返回一个StreamedRunResult上面挂着你最常用的两个方法stream_text()逐次吐出文本默认每次是累计全文stream_output()逐次吐出经过校验的结构化数据10 行代码看到第一个字 把下面片段保存成stream_demo.py配好模型 API Key 即可运行from pydantic_ai import Agent agent Agent(openai:gpt-4o-mini) # 换成你手头有 Key 的模型 async def main(): async with agent.run_stream(用三句话介绍 Pydantic) as result: async for text in result.stream_text(): print(text, end, flushTrue) # 每次迭代拿到累计全文 print(\n用量:, result.usage) # 流结束后统计仍可用 import asyncio; asyncio.run(main())注意async with块流的生命周期被框在这个块里块结束流自动关闭不用手动清理。更多入口说明见 docs/agent.md。流结束后再取最终结果一个容易忽略的细节result.output在整个流消费完之前是拿不到的。你可以等循环自然结束后再读它也可以用await result.get_output()在循环里提前拿到已解析的输出。result.usage里的 token 统计同样在流结束后依然有效方便你记费用。二、流式结构化数据边到边校验如果你的输出不是纯文本而是表格、列表这类结构化数据stream_text()就不够用了。此时靠output_type声明结构Pydantic AI 会在数据流到的过程中对 JSON 做分段校验——先到的字段先验证不用等整段闭合。声明输出结构并流式接收from typing import TypedDict from pydantic_ai import Agent class Whale(TypedDict): name: str length: float # 成年鲸鱼平均长度米 agent Agent(openai:gpt-4o-mini, output_typelist[Whale]) async with agent.run_stream(给我 5 种鲸鱼的数据) as result: async for whales in result.stream_output(debounce_by0.05): print(whales) # 当前已校验通过的半成品数据跑起来的效果name到了先显示名字length到了再补上数值像填表一样一格格亮出来。完整可运行版本参考 examples/pydantic_ai_examples/stream_whales.py。stream_text 和 stream_output 怎么选方法拿到的是什么适合场景stream_text()累计的纯文本聊天回复、Markdown 渲染stream_output()部分校验通过的结构化对象仪表盘、表格、列表实时刷新run_stream_events()原始事件流工具调用、增量片段等需要展示正在调用哪个工具的过程stream_output在数据没凑齐之前可能抛出校验错误也可能返回不完整的对象如果你只要要么全对、要么别给我就改用非流式的agent.run()。三、出问题时别慌工具调用、断流与取消流式场景里最典型的三类状况模型中途要调工具、网络把流掐断了、用户想提前停止。三种各有各的接法。想看工具调用过程换 run_stream_eventsrun_stream()默认只关心最终输出。如果你的智能体要调天气、查数据库之类的工具想在前端显示正在查询…用run_stream_events()拿原始事件流from pydantic_ai import AgentRunResultEvent, FunctionToolCallEvent async with agent.run_stream_events(北京今天天气如何) as events: async for event in events: if isinstance(event, FunctionToolCallEvent): print(准备调用工具:, event.tool_name) elif isinstance(event, AgentRunResultEvent): print(完成:, event.result.output)事件流里还有PartStartEvent、PartDeltaEvent、PartEndEvent想自己拼文本时按这些增量拼接即可。断流与错误的兜底顺序瞬时网络错误给 Agent 加retries框架会自动重试详见 docs/retries.mdagent Agent(openai:gpt-4o-mini, retries3) # 失败最多自动重试 3 次模型输出不合法结构化流中校验失败会以异常抛出捕获后可以选择重新发起请求带上新提示词或降级为非流式调用长任务状态把已完成的中间结果写进会话历史messages重跑时传回去相当于断点续传用户点停止怎么办run_stream_events()返回的句柄上挂着cancel()在另一个任务比如 UI 的停止按钮回调里调用它继续迭代的协程会收到RunCancelled异常流被干净地关掉。注意取消后usage统计是尽力而为的别拿它做精确计费。四、把节奏调顺几个实用调优点 跑通之后体验好坏主要取决于刷新多频繁和资源怎么释放。以下三点不需要改架构改参数就行。调优点做法效果刷新节奏stream_text()/stream_output()都支持debounce_by秒控制界面刷新间隔平衡实时感和渲染开销增量 vs 全量stream_text(deltaTrue)只吐新增片段拼接自己管避免重复打印累计全文渲染节流配合 RichLive或前端节流渲染长文本时避免每字节都重绘几个补充建议长回答场景给debounce_by设个小值如0.05用户几乎无感渲染压力也小消费完的中间片段及时丢弃别把每个delta都攒在列表里async with块退出后流相关资源会随结果对象一起释放想直观看效果官方示例 examples/pydantic_ai_examples/stream_markdown.py 用 Rich 实时渲染 Markdown可照着改成自己的 UI五、选型速查与上线前检查清单 ✅按需求选入口你的需求推荐入口备注只要打字机式文本run_streamstream_text()最简单90% 场景够用要流式结构化数据run_streamstream_output()记得给 Agent 设output_type要展示工具调用过程run_stream_events()自己拼增量事件不需要实时性agent.run()/run_sync()拿到即校验省心非 async 环境run_stream_sync()同步版流式接口上线前检查清单确认async with agent.run_stream(...)覆盖了整个消费循环避免流悬挂确认模型支持流式与你的输出形态不支持图像输出等能力时框架会提前抛错给重试retries和超时设置上限防止个别请求拖死整个界面前端渲染加节流debounce_by与渲染帧率匹配日志里记录result.usage便于核对 token 成本把入口、校验、异常、节奏这四块对上流式功能就稳了。想深入结构化输出的各种形态可以继续看 docs/output.md 和 docs/agent.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 拿到实时文本与结构化结果
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-aiPydantic AI 是 Python 生态里的 AI 智能体框架它能让大模型的回答一边生成一边流式推送给你并顺手完成 Pydantic 结构化校验。本文适合会基础 Python、想做出打字机效果聊天界面或实时数据面板的开发者跟着走一遍就能把run_stream用熟。一、先跑起来10 行代码流式输出流式输出说白了就是模型每生成一小段文字立刻推送到你的屏幕而不是等全部生成完再一次性给你。Pydantic AI 用agent.run_stream()这个异步上下文管理器提供这条通道它返回一个StreamedRunResult上面挂着你最常用的两个方法stream_text()逐次吐出文本默认每次是累计全文stream_output()逐次吐出经过校验的结构化数据10 行代码看到第一个字 把下面片段保存成stream_demo.py配好模型 API Key 即可运行from pydantic_ai import Agent agent Agent(openai:gpt-4o-mini) # 换成你手头有 Key 的模型 async def main(): async with agent.run_stream(用三句话介绍 Pydantic) as result: async for text in result.stream_text(): print(text, end, flushTrue) # 每次迭代拿到累计全文 print(\n用量:, result.usage) # 流结束后统计仍可用 import asyncio; asyncio.run(main())注意async with块流的生命周期被框在这个块里块结束流自动关闭不用手动清理。更多入口说明见 docs/agent.md。流结束后再取最终结果一个容易忽略的细节result.output在整个流消费完之前是拿不到的。你可以等循环自然结束后再读它也可以用await result.get_output()在循环里提前拿到已解析的输出。result.usage里的 token 统计同样在流结束后依然有效方便你记费用。二、流式结构化数据边到边校验如果你的输出不是纯文本而是表格、列表这类结构化数据stream_text()就不够用了。此时靠output_type声明结构Pydantic AI 会在数据流到的过程中对 JSON 做分段校验——先到的字段先验证不用等整段闭合。声明输出结构并流式接收from typing import TypedDict from pydantic_ai import Agent class Whale(TypedDict): name: str length: float # 成年鲸鱼平均长度米 agent Agent(openai:gpt-4o-mini, output_typelist[Whale]) async with agent.run_stream(给我 5 种鲸鱼的数据) as result: async for whales in result.stream_output(debounce_by0.05): print(whales) # 当前已校验通过的半成品数据跑起来的效果name到了先显示名字length到了再补上数值像填表一样一格格亮出来。完整可运行版本参考 examples/pydantic_ai_examples/stream_whales.py。stream_text 和 stream_output 怎么选方法拿到的是什么适合场景stream_text()累计的纯文本聊天回复、Markdown 渲染stream_output()部分校验通过的结构化对象仪表盘、表格、列表实时刷新run_stream_events()原始事件流工具调用、增量片段等需要展示正在调用哪个工具的过程stream_output在数据没凑齐之前可能抛出校验错误也可能返回不完整的对象如果你只要要么全对、要么别给我就改用非流式的agent.run()。三、出问题时别慌工具调用、断流与取消流式场景里最典型的三类状况模型中途要调工具、网络把流掐断了、用户想提前停止。三种各有各的接法。想看工具调用过程换 run_stream_eventsrun_stream()默认只关心最终输出。如果你的智能体要调天气、查数据库之类的工具想在前端显示正在查询…用run_stream_events()拿原始事件流from pydantic_ai import AgentRunResultEvent, FunctionToolCallEvent async with agent.run_stream_events(北京今天天气如何) as events: async for event in events: if isinstance(event, FunctionToolCallEvent): print(准备调用工具:, event.tool_name) elif isinstance(event, AgentRunResultEvent): print(完成:, event.result.output)事件流里还有PartStartEvent、PartDeltaEvent、PartEndEvent想自己拼文本时按这些增量拼接即可。断流与错误的兜底顺序瞬时网络错误给 Agent 加retries框架会自动重试详见 docs/retries.mdagent Agent(openai:gpt-4o-mini, retries3) # 失败最多自动重试 3 次模型输出不合法结构化流中校验失败会以异常抛出捕获后可以选择重新发起请求带上新提示词或降级为非流式调用长任务状态把已完成的中间结果写进会话历史messages重跑时传回去相当于断点续传用户点停止怎么办run_stream_events()返回的句柄上挂着cancel()在另一个任务比如 UI 的停止按钮回调里调用它继续迭代的协程会收到RunCancelled异常流被干净地关掉。注意取消后usage统计是尽力而为的别拿它做精确计费。四、把节奏调顺几个实用调优点 跑通之后体验好坏主要取决于刷新多频繁和资源怎么释放。以下三点不需要改架构改参数就行。调优点做法效果刷新节奏stream_text()/stream_output()都支持debounce_by秒控制界面刷新间隔平衡实时感和渲染开销增量 vs 全量stream_text(deltaTrue)只吐新增片段拼接自己管避免重复打印累计全文渲染节流配合 RichLive或前端节流渲染长文本时避免每字节都重绘几个补充建议长回答场景给debounce_by设个小值如0.05用户几乎无感渲染压力也小消费完的中间片段及时丢弃别把每个delta都攒在列表里async with块退出后流相关资源会随结果对象一起释放想直观看效果官方示例 examples/pydantic_ai_examples/stream_markdown.py 用 Rich 实时渲染 Markdown可照着改成自己的 UI五、选型速查与上线前检查清单 ✅按需求选入口你的需求推荐入口备注只要打字机式文本run_streamstream_text()最简单90% 场景够用要流式结构化数据run_streamstream_output()记得给 Agent 设output_type要展示工具调用过程run_stream_events()自己拼增量事件不需要实时性agent.run()/run_sync()拿到即校验省心非 async 环境run_stream_sync()同步版流式接口上线前检查清单确认async with agent.run_stream(...)覆盖了整个消费循环避免流悬挂确认模型支持流式与你的输出形态不支持图像输出等能力时框架会提前抛错给重试retries和超时设置上限防止个别请求拖死整个界面前端渲染加节流debounce_by与渲染帧率匹配日志里记录result.usage便于核对 token 成本把入口、校验、异常、节奏这四块对上流式功能就稳了。想深入结构化输出的各种形态可以继续看 docs/output.md 和 docs/agent.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），仅供参考