Pydantic AI流式输出实战：逐字刷新、中断恢复与结果校验-北京尧图网络

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 的Agent.run_stream把这件事收敛成两步进入一个异步上下文管理器发起模型请求再用stream_output、stream_text逐块取回输出结束时还能拿到一份完整通过校验的结果。下面从最小可运行示例讲起覆盖参数怎么调、流断了怎么恢复、哪些坑会踩到。从一个聊天界面开始最小流式循环 ️先做一个终端聊天界面模型每说一句屏幕就刷新一次。只需要run_stream加一个async forfrom pydantic_ai import Agent agent Agent(openai:gpt-5.2) async def main(): async with agent.run_stream(给我看一个 Pydantic 用法示例) as res: async for chunk in res.stream_text(deltaTrue): print(chunk, end, flushTrue) # 把新到的增量追加到界面run_stream是异步上下文管理器进入时发出模型请求并产出一个StreamedRunResult退出时关闭底层流。这里deltaTrue表示每次迭代只产出新到的增量正好用于逐字追加改成默认的deltaFalse每次产出的是从开头到当前点的累积全文适合整块覆盖重绘的渲染方式。完整可运行的版本在 examples/pydantic_ai_examples/stream_markdown.py它用 rich 的Live把每次产出的文本重新渲染成 Markdown效果就是打字机式的逐段出现。机制拆解分块是怎么变成已校验数据的把stream_output当成每次迭代都重新校验一遍就够了细节有三点值得知道每次迭代都过一遍 Pydantic 校验。结构化输出场景下内部输出处理器以部分校验模式partial mode反复解析模型吐出的 JSON实现在 pydantic_ai_slim/pydantic_ai/_output.py。迭代器给你的不是原始 JSON 字符串而是此刻能解析出的部分值还没生成的字段是缺失的已生成的字段是真实类型。debounce_by控制校验合并窗口。默认0.1即 0.1 秒内到达的多个分块合并成一次校验传None关闭合并每个分块单独校验。校验本身有开销这个参数就是实时性和 CPU 之间的旋钮。迭代终止后还能读完整状态。退出上下文后res.get_output()给最终结果res.usage给本次运行的 token 用量适合做持久化和计费记录。校验入口的完整签名可以看 pydantic_ai_slim/pydantic_ai/result.py 里StreamedRunResult的三个stream_*方法。stream_output怎么用、debounce_by如何设置先给选型表再给结构化示例方法每次迭代产出适合场景stream_text(deltaTrue)新文本增量逐字追加的聊天界面stream_text()累积文本覆盖式重绘stream_output()结构化数据的当前部分值实时表格、仪表盘stream_response()ModelResponse快照含state自己消费模型事件await get_output()最终完整校验结果落库、传给下游结构化流式传输是 Pydantic AI 的强项声明一个 Pydantic 类型模型边生成你边看到行。以仓库里的鲸鱼示例为例agent Agent(openai:gpt-5.2, output_typelist[Whale]) async with agent.run_stream(Generate me details of 5 species of Whale.) as res: async for rows in res.stream_output(debounce_by0.01): redraw_table(rows) # rows 是部分列表缺字段先占位随后补齐debounce_by的取值经验默认0.1对大多数场景够用想要更灵敏的刷新就把值调小示例里用的0.01代价是校验次数变多纯文本且不在乎每次校验时可以传None跳过合并把开销压到最低。原始示例见 examples/pydantic_ai_examples/stream_whales.py其中用NotRequired声明了可选字段正好匹配字段后到的部分值。流式中断如何恢复一张排错表流式链路长断点可能出现在网络、模型或校验任何一环。按现象 → 原因 → 解法对照处理现象常见原因解法迭代中途抛异常流没了网络中断、供应商限流捕获异常后带上message_history重新发起运行历史消息可直接复用用户点了停止按钮需要主动终止调用res.cancel()发出中断结束后res.cancelled为真最终状态是interrupted结构化输出校验报错模型输出与 schema 不符调大retries重试预算或用NotRequired放宽必填字段界面更新卡顿、CPU 高每个分块都在跑校验调大debounce_by让多个分块合并成一次校验主动取消和收尾可以这样写async with agent.run_stream(prompt) as res: async for text in res.stream_text(): push_to_screen(text) final await res.get_output() # 走到这说明流正常结束 if res.cancelled: return # 用户主动取消不再落库注意区分cancel()是协作式的中断适合停止生成按钮而供应商侧的断流是异常靠上面的重试路径恢复。两者不要混用。get_output怎么拿到最终已校验的结果stream_*迭代出来的永远是某一刻的快照只有退出迭代后的get_output()保证完整、且通过了全部校验器。两个容易踩的差别文本输出函数不作用于stream_text如果你给 Agent 配了TextOutput之类的输出函数想让它生效要用stream_output()deltaTrue时结果校验器不会跑在文本上所以stream_text(deltaTrue)适合纯展示最终值仍以get_output()为准。一句话原则屏幕上展示快照落库前过一遍get_output()。下一步可以做什么想把流式接到 Web 界面读一下 examples/pydantic_ai_examples/chat_app.py 的 FastAPI 实现需要观察流式过程的 token 用量与耗时配合 Logfire 的监控面板docs/img/logfire-monitoring-pydanticai.png 里的指标体系做验证对长任务想做到断点续跑可以去看pydantic_ai_slim/pydantic_ai/durable_exec/下的持久化执行方案它把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创作声明：本文部分内容由AI辅助生成（AIGC），仅供参考

Pydantic AI流式输出实战：逐字刷新、中断恢复与结果校验

相关推荐

npx add-skill 实战：Agent Skill 安装与工程化指南

深入理解 @visx/curve：为 visx 折线与面积图注入 d3 曲线插值能力

matcher鸿蒙化适配实践：语义化断言与端侧质量验证

更多资讯

iOS 18安全机制升级详解：BlastDoor与PAC如何阻断越狱

Page Assist：怎么在每个网页上装一个本地 AI 助手

从论文到答辩PPT：AI辅助制作全流程实操指南

从聊天机器人到数字员工：WorkBuddy产品力拆解

OpenCode五大运行模式详解与应用场景

Mistral AI与Mozilla合作：开放模型与隐私优先的AI集成实践

今日更新

深入解析Transformer多头注意力机制与工程优化

OpenClaw 的 Skills 跑学习任务，模型通道改到 TaoToken 通道行不行？

ChatGPT报错Oops, an error occurred! 全链路排查指南

本周精选

深入解析Transformer多头注意力机制与工程优化

OpenClaw 的 Skills 跑学习任务，模型通道改到 TaoToken 通道行不行？

ChatGPT报错Oops, an error occurred! 全链路排查指南

本月热门

基于YOLO的焊缝缺陷检测：从数据标注到工业部署全流程实践

分布式服务的决策记录

创业产品的首版范围

读完文章，想聊聊您的网站？