Pydantic AI 流式处理指南选对 4 种实时输出模式告别界面干等【免费下载链接】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 的run_stream就是解决这件事的入口它让模型吐出的每一个片段都能实时推送到你的界面而不是攒成一坨再交给用户。本文带你把流式输出怎么跑起来、不同场景选哪种流、最容易踩的坑一次讲清照着改几行代码就能落地。原理速览数据从哪里来到哪里去一句话讲清楚模型按片段吐出内容run_stream把每个片段包装成一个可迭代的结果对象你在async for里逐块消费。数据流向很简单用户 prompt → 模型逐块返回 → 框架累积并可选地校验 → 你的消费循环拿到当前已有的全部文本或增量片段。理解这一点后面所有模式的区别就只是你在消费循环里拿到的是纯文本、是已校验的结构、还是原始模型事件。流式模式怎么选一张表对号入座Pydantic AI 给了你几个层次的流式消费方法选哪个取决于你最终要展示什么。业务场景推荐方法你在循环里拿到的典型用途聊天回复、逐字打字机效果stream_text(deltaTrue)增量文本片段对话界面实时打字纯文本但想看当前累积stream_text(deltaFalse)从头到当前点的完整文本简单 Markdown 渲染表格/列表等结构数据stream_output()已按 Schema 校验过的对象数据看板、实时监控需要感知工具调用、原始事件event_stream_handler或stream_response()模型事件 /ModelResponse智能体工作流、细粒度控制文本图片混合配合支持图像输出的模型 能力开关多模态片段多模态助手 一句话心法只展示文本就用stream_text要结构化数据就用stream_output要插手中间过程才上事件层。三步跑通流式输出下面是一个端到端、能直接跑的最小例子创建一个 Agent用run_stream打开流逐字打印最后取完整结果。import asyncio from pydantic_ai import Agent agent Agent(openai:gpt-5.2) # 换成你的模型与 API Key async def main(): prompt 用三句话介绍一下自己 async with agent.run_stream(prompt) as result: # 第一步逐块打印增量文本打字机效果 async for chunk in result.stream_text(deltaTrue): print(chunk, end, flushTrue) # 第二步流结束后拿经过完整校验的最终结果 final await result.get_output() print() print(最终结果:, final) asyncio.run(main())三步分别在做什么agent.run_stream(prompt)进入异步上下文框架开始向模型要流并把结果句柄交给result。async for chunk in result.stream_text(deltaTrue)deltaTrue时每次 yield 的是新到的那一段所以print出来就是打字机效果改成deltaFalse则是每次给从头到现在的全文。await result.get_output()流关闭后取经过最终校验的完整输出作为兜底保存。提示run_stream会在拿到第一段符合输出类型的结果后就停止后续工具调用。若你希望跑完整体流程的同时流式输出改用agent.iter()或配合event_stream_handler细节见 docs/output.md。避坑清单现象 → 原因 → 解法坑一结构化输出一直转圈到最后一行才刷新。现象表格明明已经生成大半界面却迟迟不更新。原因stream_output默认带debounce_by0.1的防抖且每个片段都要跑一次校验长结构下校验开销会拖慢刷新。解法调小或关掉防抖stream_output(debounce_by0.01)确认真实时间敏感时再用debounce_byNone。坑二stream_text()报UserError: can only be used with text responses。现象对结构化 Agent 调stream_text直接报错。原因stream_text只接受文本型输出你的 Agent 定义了结构化output_type。解法结构化数据请改用stream_output()别混用。坑三拿到半成品就落库数据不完整。现象把循环中最后一次拿到的对象存起来偶尔字段缺失。原因中间 yield 的是部分校验allow_partialTrue的快照字段可能还没齐。解法永远以get_output()或循环结束后最后一次完整 yield 作为落库依据中间值只用于展示。坑四长任务内存越跑越高。现象长时间流式任务进程内存缓慢上涨。原因不断累积中间快照未及时释放。解法循环里只保留当前最新一帧旧的交给 GC大结果集分页流式输出用async with确保流与连接被正确关闭。进阶调优性能与稳定性的关键参数debounce_by控制刷新节奏值越小越实时、CPU 校验越频繁值越大越省、但界面更新越钝。文本流对0.01~0.1之间做实测取舍。delta决定打印语义True是新片段打字机False是全文快照整体重绘。选错会导致重复打印或丢字。retries做断流兜底网络抖动导致的临时错误可用retries参数设置自动重试次数例如agent.run_stream(prompt, retries3)。复用连接、控制并发长会话尽量复用底层客户端连接减少握手高并发场景注意并发上限避免消费者端来不及处理造成堆积。用可观测性定位瓶颈接入 Logfire 后能看清每次模型请求耗时与 token 消耗快速判断慢在模型还是慢在自己消费端。结语流式输出的本质是把等一整坨变成边到边给选对stream_text/stream_output匹配你的展示形态用debounce_by和delta调好实时性与开销的平衡再用retries和完整结果落库兜住稳定性界面就能从干等转圈变成逐字出现。延伸阅读输出与流式 APIdocs/output.md结构化流式示例examples/pydantic_ai_examples/stream_whales.pyMarkdown 流式示例examples/pydantic_ai_examples/stream_markdown.py核心结果类源码pydantic_ai_slim/pydantic_ai/result.py【免费下载链接】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 流式处理指南：选对 4 种实时输出模式，告别界面干等
Pydantic AI 流式处理指南选对 4 种实时输出模式告别界面干等【免费下载链接】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 的run_stream就是解决这件事的入口它让模型吐出的每一个片段都能实时推送到你的界面而不是攒成一坨再交给用户。本文带你把流式输出怎么跑起来、不同场景选哪种流、最容易踩的坑一次讲清照着改几行代码就能落地。原理速览数据从哪里来到哪里去一句话讲清楚模型按片段吐出内容run_stream把每个片段包装成一个可迭代的结果对象你在async for里逐块消费。数据流向很简单用户 prompt → 模型逐块返回 → 框架累积并可选地校验 → 你的消费循环拿到当前已有的全部文本或增量片段。理解这一点后面所有模式的区别就只是你在消费循环里拿到的是纯文本、是已校验的结构、还是原始模型事件。流式模式怎么选一张表对号入座Pydantic AI 给了你几个层次的流式消费方法选哪个取决于你最终要展示什么。业务场景推荐方法你在循环里拿到的典型用途聊天回复、逐字打字机效果stream_text(deltaTrue)增量文本片段对话界面实时打字纯文本但想看当前累积stream_text(deltaFalse)从头到当前点的完整文本简单 Markdown 渲染表格/列表等结构数据stream_output()已按 Schema 校验过的对象数据看板、实时监控需要感知工具调用、原始事件event_stream_handler或stream_response()模型事件 /ModelResponse智能体工作流、细粒度控制文本图片混合配合支持图像输出的模型 能力开关多模态片段多模态助手 一句话心法只展示文本就用stream_text要结构化数据就用stream_output要插手中间过程才上事件层。三步跑通流式输出下面是一个端到端、能直接跑的最小例子创建一个 Agent用run_stream打开流逐字打印最后取完整结果。import asyncio from pydantic_ai import Agent agent Agent(openai:gpt-5.2) # 换成你的模型与 API Key async def main(): prompt 用三句话介绍一下自己 async with agent.run_stream(prompt) as result: # 第一步逐块打印增量文本打字机效果 async for chunk in result.stream_text(deltaTrue): print(chunk, end, flushTrue) # 第二步流结束后拿经过完整校验的最终结果 final await result.get_output() print() print(最终结果:, final) asyncio.run(main())三步分别在做什么agent.run_stream(prompt)进入异步上下文框架开始向模型要流并把结果句柄交给result。async for chunk in result.stream_text(deltaTrue)deltaTrue时每次 yield 的是新到的那一段所以print出来就是打字机效果改成deltaFalse则是每次给从头到现在的全文。await result.get_output()流关闭后取经过最终校验的完整输出作为兜底保存。提示run_stream会在拿到第一段符合输出类型的结果后就停止后续工具调用。若你希望跑完整体流程的同时流式输出改用agent.iter()或配合event_stream_handler细节见 docs/output.md。避坑清单现象 → 原因 → 解法坑一结构化输出一直转圈到最后一行才刷新。现象表格明明已经生成大半界面却迟迟不更新。原因stream_output默认带debounce_by0.1的防抖且每个片段都要跑一次校验长结构下校验开销会拖慢刷新。解法调小或关掉防抖stream_output(debounce_by0.01)确认真实时间敏感时再用debounce_byNone。坑二stream_text()报UserError: can only be used with text responses。现象对结构化 Agent 调stream_text直接报错。原因stream_text只接受文本型输出你的 Agent 定义了结构化output_type。解法结构化数据请改用stream_output()别混用。坑三拿到半成品就落库数据不完整。现象把循环中最后一次拿到的对象存起来偶尔字段缺失。原因中间 yield 的是部分校验allow_partialTrue的快照字段可能还没齐。解法永远以get_output()或循环结束后最后一次完整 yield 作为落库依据中间值只用于展示。坑四长任务内存越跑越高。现象长时间流式任务进程内存缓慢上涨。原因不断累积中间快照未及时释放。解法循环里只保留当前最新一帧旧的交给 GC大结果集分页流式输出用async with确保流与连接被正确关闭。进阶调优性能与稳定性的关键参数debounce_by控制刷新节奏值越小越实时、CPU 校验越频繁值越大越省、但界面更新越钝。文本流对0.01~0.1之间做实测取舍。delta决定打印语义True是新片段打字机False是全文快照整体重绘。选错会导致重复打印或丢字。retries做断流兜底网络抖动导致的临时错误可用retries参数设置自动重试次数例如agent.run_stream(prompt, retries3)。复用连接、控制并发长会话尽量复用底层客户端连接减少握手高并发场景注意并发上限避免消费者端来不及处理造成堆积。用可观测性定位瓶颈接入 Logfire 后能看清每次模型请求耗时与 token 消耗快速判断慢在模型还是慢在自己消费端。结语流式输出的本质是把等一整坨变成边到边给选对stream_text/stream_output匹配你的展示形态用debounce_by和delta调好实时性与开销的平衡再用retries和完整结果落库兜住稳定性界面就能从干等转圈变成逐字出现。延伸阅读输出与流式 APIdocs/output.md结构化流式示例examples/pydantic_ai_examples/stream_whales.pyMarkdown 流式示例examples/pydantic_ai_examples/stream_markdown.py核心结果类源码pydantic_ai_slim/pydantic_ai/result.py【免费下载链接】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），仅供参考