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打开聊天页面模型逐字吐出的内容要立刻推到屏幕上而最后拼完的字符串还得通过 JSON Schema 校验才能入库——这是做 AI 应用时很常见的处境。Pydantic AI 的流式输出接口run_stream就是为这件事设计的数据边到边推校验边到边做。官方仓库里的鲸鱼示例里表格每收到一个分片就重渲染一次不用盯着 spinner 等完整响应。它到底在干什么agent.run_stream()返回一个异步上下文管理器内部会先嗅探响应的前几个分片判断这是工具调用还是最终输出然后把流整体交给StreamedRunResult继续迭代。结构化结果走 Pydantic 的部分校验partial validation每个不完整的快照先宽松校验拼完才做全量校验——有点像先报比分、赛后再改错的实时解说。一次流式运行的 span 在追踪系统里长这样30 秒跑通最小示例import asyncio from pydantic_ai import Agent agent Agent(openai:gpt-5.2) async def main(): async with agent.run_stream(Where does hello world come from?) as result: async for message in result.stream_text(): print(message) asyncio.run(main())来源output.md关键是run_stream的上下文管理器写法——流结束时连接才会真正关闭而stream_text()每次 yield 的是累积的完整文本快照不是增量所以直接覆盖渲染即可。按场景拆用法 让表格数据边生成边刷新适合结构化数据要实时展示的仪表盘场景。agent Agent(openai:gpt-5.2, output_typelist[Whale]) async with agent.run_stream(Generate me details of 5 species of Whale.) as result: async for whales in result.stream_output(debounce_by0.01): render_table(whales) # 每次用完整快照重渲染来源stream_whales.py容易忽略的是debounce_by默认 0.1 秒合并一次分片长响应下能明显减少校验开销传None则是分片到了就 yield。注意output_type默认走 Tool Output 模式模型不支持流式工具参数时得改用 Native/Prompted Output否则流不动。逐字推送到屏幕聊天 UI 要打字机效果时用stream_text(deltaTrue)yield 的是原始增量适合直接append到界面。两个坑要心里有数delta 模式会跳过 output validator且最终消息不会写入result的消息历史——需要多轮上下文续接时还是用默认的快照模式。工具调用的中间结果要实时回传模型先调工具再出结论时run_stream只关心最终输出。想拿到工具调用的增量事件改用agent.iter()迭代每个节点的AgentStream看到FinalResultEvent后再切到stream.stream_output()拿校验过的部分结果。记住AgentStream是一次性迭代器切走后剩余原始事件就被消耗了想全量留痕就用run_stream_events()。踩坑与排障结构化结果迟迟不出现先查模型的流式工具参数支持。output_type默认是 Tool Output 模式模型不支持流式工具参数时输出根本流不出来切 Native Output 或 Prompted Output 即可细节见 docs/output.md。渲染内容越拼越多把快照当增量了。stream_output()每个值都是累积快照字段没凑够通过部分校验前会整个缺席正确姿势是每次用快照整体替换渲染而不是 append。流到一半断掉分层看重试。传输层、Provider SDK 层的重试对 agent 完全不可见agent 层的retries{output: N}每次重试都多花一次模型请求且各层预算相乘3×3×2 一次逻辑调用可能打出 18 个请求全貌见 docs/retries.md。点了停止还在扣 tokenresult.cancel()只断本地流。它关闭底层模型流但 Google、xAI、Hugging Face 的 SDK 不保证远端生成立刻停止计费以各 provider 文档为准。怎么选需求用哪个只要纯文本stream_text()快照模式打字机逐字效果stream_text(deltaTrue)结构化数据边生成边校验stream_output(debounce_by...)自己控制校验时机stream_response()validate_response_output(allow_partial...)流式拿工具调用中间事件agent.iter()或run_stream_events()⚠️ 判断顺序先看要的是文本还是结构再看要不要中间事件最后才调debounce_by这类参数。流式结果的核心实现在 pydantic_ai_slim/pydantic_ai/result.pyStreamedRunResult的每个方法签名都带了行为说明排障时对照着看最快。【免费下载链接】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），仅供参考