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聊天界面里用户盯着转圈 10 秒才等到 AI 把一整段回复吐出来——模型其实第一个 token 一秒钟就生成了。问题出在你等的是完整结果而不是正在生成的结果。Pydantic AI 的流式输出入口是Agent.run_stream()它返回StreamedRunResult让你一边接收增量数据一边做校验和渲染。这篇文章按实际动手的顺序展开先跑通最小示例再依次解决增量校验、工具调用事件、断流重试、实时性与性能这几个问题最后给一张参数对照表收尾。三步跑通 run_stream先建立能跑的信心。这段代码来自 examples/pydantic_ai_examples/stream_markdown.py把渲染部分换掉即可直接用import asyncio from pydantic_ai import Agent agent Agent(openai:gpt-5-mini) async def main(): async with agent.run_stream(用一句话介绍旧金山) as result: async for message in result.stream_text(): print(message, end, flushTrue) asyncio.run(main())run_stream是个上下文管理器退出async with时连接自动关闭stream_text()默认每次 yield 的是累积全文快照不是 delta——所以直接覆盖渲染不要拼接。跑通后你会发现第一段文字出现的时机取决于你消费什么。结构化流的增量校验哪些中间值可以相信结构化输出没生成完时JSON 是不完整的直接严格校验必然失败。Pydantic AI 的解法是 pydantic 的部分解析未闭合的字符串被当作已经写完的值源码里叫trailing-strings模式见 pydantic_ai_slim/pydantic_ai/_output.py 的ObjectOutputProcessor.validate。对应到 API 上stream_output()每次返回的是能通过部分校验的累积快照。async with agent.run_stream(5 种鲸鱼的详情) as result: async for whales in result.stream_output(debounce_by0.01): render(whales)三个要点都来自 docs/output.md 的 Streamed Results 一节每次 yield 是累积快照不是增量。某个字段或列表项在数据不够时可能整体缺席渲染要覆盖更新不要追加。流式场景下agent.output_validator会被调用多次用ctx.partial_output区分中间快照和最终输出对中间值放宽校验只在最后严格把关。需要手动控制校验时用stream_response()拿原始ModelResponse迭代本身不会因校验失败抛异常再调validate_response_output(response, allow_partialresponse.state incomplete)校验失败就跳过这一帧。allow_partial解决的正是还没跑完的结果该不该信它是 pydantic 的实验性部分校验开关True时未闭合字符串按已完成的值解析。工具调用场景中间事件怎么消费run_stream有个必须知道的取舍它会以第一个匹配output_type的输出作为最终结果不会执行模型在此之后发起的工具调用。要看到完整过程工具调用、工具结果、最终输出得走事件流docs/agent.md 给了两种方式async def handler(ctx, event_stream): async for event in event_stream: if isinstance(event, FunctionToolCallEvent): print(f调用 {event.part.tool_name}: {event.part.args}) elif isinstance(event, FunctionToolResultEvent): print(f结果: {event.part.content}) async with agent.run_stream(prompt, event_stream_handlerhandler) as run: async for text in run.stream_text(): ...常用事件类型都从pydantic_ai直接导出PartStartEvent某段内容开始、PartDeltaEvent文本/思考/工具参数增量、FunctionToolCallEvent模型发起工具调用、FunctionToolResultEvent工具返回、FinalResultEvent开始产出最终结果。事件流截图长这样另一种写法是agent.run_stream_events(prompt)它是异步上下文管理器yield 的事件以携带最终结果的AgentRunResultEvent收尾。注意它和event_stream_handler一样只给原始事件文本和结构化输出需要你自己从PartDeltaEvent拼起来。流断了怎么续重试、降级与取消重试在 Pydantic AI 里分三层各管一段docs/retries.md 有完整优先级说明provider SDK 层客户端的max_retries管 429、连接重置这类临时错误transport 层基于 tenacity 的重试 transport可针对特定异常类型和退避策略定制输出层retries{output: N}校验或output_validator拒绝最终答案时框架把失败原因写回上下文让模型自我纠正最多 N 次。agent Agent(openai:gpt-5-mini, retries{output: 2}) # 主模型彻底不可用时换到下一个模型 agent Agent(FallbackModel(openai:gpt-5-mini, anthropic:claude-sonnet-4-6))FallbackModel不重试同一个模型只在当前模型失败时切换下一个适合这个 provider 挂了而不是网络抖一下的场景。主动停止是另一件事。用户点停止生成时调result.cancel()通知模型停止生成并关闭连接随后result.cancelled为 True最终ModelResponse会被标记stateinterrupted。复用这段历史时未完成的部分工具调用参数会被自动修复。注意 Google、xAI、Hugging Face 的 SDK 只保证本地迭代器中断不保证远端生成立刻停止。实时性与性能的权衡三个杠杆按需选debounce_by把一段时间内的 chunk 合并成一次 yield减少每到一个 chunk 就校验一次的开销。默认 0.1 秒长结构化响应建议保持或加大stream_text(deltaTrue)这类轻量消费可以直接传None求最低延迟。deltaTrue发送增量而非全文快照网络开销小很多代价是最终文本不会进入result.messages多轮对话历史需要自己管理。及时取消拿到足够信息就cancel()比消费完再丢弃省 token 也省带宽。收尾参数对照与决策清单场景用什么备注只想要文本result.stream_text()默认 yield 累积快照deltaTrue收增量要边收边渲染结构化数据result.stream_output(debounce_by...)每帧是过部分校验的累积快照要看工具调用过程run(event_stream_handler...)或run_stream_events()run_stream不会执行最终输出之后的工具调用精细控制校验result.stream_response()validate_response_output(allow_partial...)迭代不抛校验异常流式校验器ctx.partial_output中间快照为 True最终输出为 False主动停止result.cancel()响应标记stateinterrupted临时错误providermax_retries/ tenacity transport管限流、连接重置、5xx答案不合格retries{output: N}让模型自我纠正最多 N 次模型整体不可用FallbackModel(a, b, ...)失败即切换不重试同一模型可选输出str | None流式result.get_output()stream_output()在这种情况下是空迭代器几个高频踩坑点stream_output的帧要覆盖渲染不能追加流式下output_validator抛错默认不会自动转成重试提示deltaTrue时最终文本不进result.messages。剩下的细节以 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），仅供参考