Pydantic AI 流式输出实战指南：3 步拿到实时且已校验的结果

Pydantic AI 流式输出实战指南3 步拿到实时且已校验的结果【免费下载链接】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-aiPydantic AI 是一个端到端类型化的 Python AI 框架覆盖 Agent、实时语音、图像生成与嵌入模型。流式输出是它最实用的能力之一模型还没说完你就能拿到边生成边校验的结果而不必干等最终回复。本文面向刚接触框架的开发者讲清run_stream的机制并给出一套可照做的接入、校验、调优方法。原理一图懂管道分两路校验在末端一句话结论流式输出 模型响应流 校验器框架在每一帧上做尽力校验在最后一帧上做严格校验。把run_stream想象成一条双车道管道左车道是stream_response()吐的是原始ModelResponse快照不做任何校验最快右车道是stream_output()吐的是已按你的output_type校验过的对象。它内部调用 result.py 中的validate_response_output(response, allow_partialTrue)允许 JSON 字符串还没闭合源码层面交给 Pydantic Core 的allow_partialtrailing-strings见 _output.py。中间帧校验失败不会抛异常——框架直接跳过这一帧except (ValidationError, ModelRetry): pass等下一帧数据更完整再试。只有流结束时的最后一帧以allow_partialFalse做完整校验因此迭代器吐出的最后一个值一定可用。这是整个机制里最值钱的一句承诺。分步上手第 1 步接入——用 run_stream 打开流结论run_stream是异步上下文管理器进入后拿到StreamedRunResult按需选择stream_text()或stream_output()。文本输出最简async with agent.run_stream(用一句话解释 Pydantic AI) as result: async for text in result.stream_text(): print(text, end, flushTrue)stream_text()默认每次吐出到目前为止的全量文本deltaFalse适合直接覆盖式刷新界面传deltaTrue则每次只吐增量片段。注意它要求 Agent 的输出类型是纯文本否则会抛UserError。结构化输出Pydantic 模型或TypedDict列表则用stream_output()框架帮你逐帧校验。完整示例见 examples/pydantic_ai_examples/stream_whales.py。第 2 步校验——理解两档 allow_partial结论中间帧走宽松档末帧走严格档你通常不需要自己写任何校验重试逻辑。分档逻辑在 result.py 里可以直接读到流进行中每帧allow_partialTrue尝试构造目标对象失败就静默跳过流结束对self.response做allow_partialFalse的完整校验失败才会真正抛ValidationError。如果你的输出校验依赖完整性比如字段间的业务约束可以在输出函数里检查run_context.partial_output标志对中间帧跳过这类重校验只在末帧执行。第 3 步调优——debounce 与取消结论debounce_by控制多久合并一帧cancel()控制提前收流。debounce_by默认0.1秒把间隔内的 token 合并后再触发一次校验。对长结构化响应每 token 校验一次开销很大适当调大能明显降低 CPU 占用设None表示不合并、逐帧触发。不需要跑完可以await result.cancel()它会停止当前模型响应并请求上游关闭运行本身不会报错终止。网络抖动这类临时错误在Agent(model, retries2)上配置重试即可见 examples/pydantic_ai_examples/weather_agent.py。如果某模型不支持你要的输出形态比如图像输出框架会依据ModelProfile提前检查定义在 profiles/init.py。避坑与调优问题表现解决办法界面刷新卡顿每个 token 都触发一次 Pydantic 校验调大debounce_by如 0.05~0.2s中间帧拿到不完整数据列表缺字段、字符串被截断属正常现象只在末帧做依赖完整性的业务逻辑末帧校验抛 ValidationError整次运行失败检查output_type与模型实际输出格式是否一致必要时加输出函数约束流被提前放弃后内存滞留事件积压用cancel()停止消费或drain()丢弃剩余事件误用 stream_text 拿结构化数据抛出 UserError改用stream_output()或用stream_response()自行解析几点补充中间帧校验失败被跳过意味着迭代间隔不均匀UI 端别假设帧率恒定run_stream以第一个匹配输出类型的结果为终点之后的工具调用不会发生需要全程事件流时改用run_stream_events()详见 docs/agent.md。快速验证3 分钟跑通以下片段用一个 Pydantic 模型约束输出观察表格数据长出来的过程。只需一个 OpenAI 兼容的 API Keyimport asyncio from pydantic import BaseModel from pydantic_ai import Agent class City(BaseModel): name: str population: int agent Agent(openai:gpt-5-mini, output_typelist[City]) async def main(): async with agent.run_stream(列出 3 个欧洲城市及其人口) as result: async for cities in result.stream_output(debounce_by0.1): print(cities) # 中间帧是已校验的部分结果 asyncio.run(main())运行后你会看到同一列表被逐步补全打印最后一行是完整校验通过的版本。如果不想真实调用模型可以参照 examples/pydantic_ai_examples/stream_markdown.py 换用本地 Ollama 模型名即可离线体验。资源地图流式 API 总览docs/agent.mdrun_stream、事件流、取消机制结构化输出与校验细节docs/output.md流式结果实现pydantic_ai_slim/pydantic_ai/result.py部分校验实现pydantic_ai_slim/pydantic_ai/_output.py可运行示例examples/pydantic_ai_examples/stream_whales.py、examples/pydantic_ai_examples/stream_markdown.py当你的应用开始同时消费事件流与结构化输出或需要把流接到 Web UI 时可以接着看agent.iter()与 UI 适配器那会打开流式处理的下一层空间。【免费下载链接】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），仅供参考