企业官网建设流程全解析

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='', flush=True) 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_by=0.01): render(whales)

三个要点，都来自 docs/output.md 的 Streamed Results 一节：

• 每次 yield 是累积快照，不是增量。某个字段或列表项在数据不够时可能整体缺席，渲染要覆盖更新，不要追加。
• 流式场景下@agent.output_validator会被调用多次，用ctx.partial_output区分中间快照和最终输出，对中间值放宽校验，只在最后严格把关。
• 需要手动控制校验时，用stream_response()拿原始ModelResponse（迭代本身不会因校验失败抛异常），再调validate_response_output(response, allow_partial=response.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_handler=handler) 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会被标记state='interrupted'。复用这段历史时，未完成的部分工具调用参数会被自动修复。注意 Google、xAI、Hugging Face 的 SDK 只保证本地迭代器中断，不保证远端生成立刻停止。

实时性与性能的权衡

三个杠杆，按需选：

• debounce_by：把一段时间内的 chunk 合并成一次 yield，减少"每到一个 chunk 就校验一次"的开销。默认 0.1 秒；长结构化响应建议保持或加大；stream_text(delta=True)这类轻量消费可以直接传None求最低延迟。
• delta=True：发送增量而非全文快照，网络开销小很多；代价是最终文本不会进入result.messages，多轮对话历史需要自己管理。
• 及时取消：拿到足够信息就cancel()，比消费完再丢弃省 token 也省带宽。

收尾：参数对照与决策清单

几个高频踩坑点：stream_output的帧要覆盖渲染不能追加；流式下output_validator抛错默认不会自动转成重试提示；delta=True时最终文本不进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