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-ai用 Pydantic AI 做 AI 应用时流式输出基本是默认需求用户等不起整段回答生成完。这个框架把流式收敛到同一个入口Agent.run_stream()文本、结构化数据、工具调用、中断恢复都从它派生。下面按「先跑通 → 加验证 → 加工具 → 加容错」的顺序走一遍每步只加一个能力。从 run_stream 拿到实时文本最小形态run_stream(prompt)返回上下文管理器形式的结果对象在里面迭代stream_text()就能拿到文本。默认它每次 yield 的是「截至当前的完整文本」想要每次只拿新增片段就传deltaTrue。async with agent.run_stream(用一句话介绍 Pydantic AI) as result: async for text in result.stream_text(): print(text, end, flushTrue) print(result.usage)这段代码把累积文本直接打到终端退出上下文后用result.usage拿到本次运行的 token 用量。注意stream_text()只适用于文本输出output_type是结构化类型时直接调用会抛UserError而且它不执行TextOutput函数需要那套转换逻辑时改用下一节的stream_output()。结构化输出边生成边做增量验证把output_type设成 Pydantic 模型或 TypedDict 后stream_output()每次 yield 的都是一个已验证的对象。区别在于中间每次 yield 走的是部分校验内部按allow_partialTrue处理某次校验失败只会被静默跳过不中断循环最后一次 yield 则一定做一次完整校验并保证是最终结果。实践含义中间值只管上屏渲染落库、发通知这类动作要等最后一个 yield。agent Agent(openai:gpt-4o, output_typelist[Whale]) async with agent.run_stream(生成 5 种鲸鱼的信息) as result: async for whales in result.stream_output(debounce_by0.01): table.update(whales)这里debounce_by0.01表示把 10 毫秒内的分块合并成一次再校验官方示例 examples/pydantic_ai_examples/stream_whales.py 正是靠它把表格逐行刷新出来。debounce_by 到底设多大默认值是 0.1 秒None表示不合并。列表或对象越大每次校验的开销越高源码注释里明确建议对较长的结构化响应做合并延迟敏感的界面降到 0.01 左右普通结构化流用 0.1 就够了。另外如果你的输出函数或校验器里有副作用用RunContext.partial_output区分「中间值」和「最终值」——流式运行中它对每个部分结果为 True只有最终结果那次是 False。工具调用穿插进来时流会变成什么样agent 中途调用工具时当前这条模型响应结束、工具执行完再开新的响应消费端只需要保持迭代。需要比「验证后的输出」更细的信息时比如确认某次快照里出现了ToolCallPart改用stream_response()它每次 yield 一个ModelResponse快照流式进行中stateincomplete结束时补一个statecomplete被取消则是interrupted的最终快照。⚠️ 有一个容易踩的行为流式模式下run_stream()会把第一个匹配output_type的输出流到即提交相当于early结束策略——如果模型在那之后还想再调工具运行已经结束。需要保证工具全部跑完的场景应改用非流式入口或调整结束策略而不是在流里硬等。流到一半断了取消、重试与断点续传容错分三层框架里都有对应 API主动取消await result.cancel()停止消费当前模型响应并向提供方请求关闭远端是否真的停止生成取决于提供方 SDK但最终快照的state会落成interrupted方便调用方走清理分支。校验重试部分输出没过校验时框架会按重试策略自动发回模型重试ModelRetry上一节里「失败的中间块被跳过」就是这条路径在流上的表现。断点续传流本身是无状态的可靠做法是保存消息历史断线后带着message_history重新发起run_stream从断点继续UI 侧保持渲染即可。最后核对一遍模型兼容性不同提供方在流式下的行为并不完全一致有的提供方工具参数以增量方式给有的每次给全量这层差异框架已经抹平你不用处理真正要确认的是输出模态——流式图像输出依赖模型 profile 的支持不支持的模型在启动时就会抛UserError不会流到一半才发现。换提供方时建议把 whales 那个示例完整跑一遍表格能否稳定逐行刷新比任何单测都直观地反映该提供方流式契约的兼容性。可观测性方面可以接入 Logfire每个流式分块、重试和 token 用量都会落在同一条 trace 里。流式这部分基本就是结果对象上的三个方法stream_text()管纯文本stream_output()管带增量验证的结构化数据stream_response()管看原始快照容错靠cancel()、重试策略和消息历史兜底。想动手的话从 examples/pydantic_ai_examples/stream_markdown.py 这类示例起步参数细节查 docs/output.md 的流式章节和实现 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），仅供参考