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-aiAI 聊天应用里最常见的体验问题是用户发完提问后盯着光标干等十几秒回复才整段蹦出来。Pydantic AI 的流式处理run_stream配合stream_text/stream_output把模型输出变成边生成边送达的实时数据流并且只在输出通过校验时才交出最终结果。本文面向刚接触 Pydantic AI 的开发者带你跑通逐字显示与结构化数据边生成边渲染两个任务并附一份覆盖验证失败、断流、模型不兼容的排错清单。从一个真实场景切入想象一个聊天窗口用户问完帮我写个排序函数光标闪了十秒600 字回复一次性刷出来。这段空白期里用户在猜是不是卡死了等待感远大于实际耗时。流式处理解决的正是这段等待模型每生成一小段 token 就往外发Pydantic AI 把这些片段即时推给你的代码首字延迟从等全文缩短到等第一句。对结构化输出同样有效——第一行数据可以先渲染不必等整张表生成完。原理速览agent.run_stream()是一个异步上下文管理器进入时启动 agent 运行返回一个StreamedRunResult对象。模型侧的 HTTP 响应被拆成 token 片段等事件框架把它们累积进当前响应的快照你迭代stream_text()或stream_output()时拿到的是快照经过校验的版本。结构化数据在流式阶段使用 pydantic 的部分验证逐段试校验片段还拼不成合法 JSON 的帧会被静默跳过流的最后一帧改用完整校验保证最终结果一定通过 schema。核心结论中间帧只是尽量解析的预览最后一帧或await result.get_output()才是可信的完整结果。上手实践两个任务共用同一套 API。开始前uv add pydantic-ai并配好所用模型的 API Key代码细节见 pydantic_ai_slim/pydantic_ai/agent/abstract.py 中run_stream的参数说明。任务一让回复逐字显示适用场景聊天机器人、CLI 工具任何需要把纯文本 Markdown 实时刷上屏幕的界面。关键参数delta默认False每次迭代返回到目前为止的全部文本设True只返回新增片段长回复更省内存。debounce_by默认0.1秒把时间窗口内的片段合并成一次迭代None表示不合并。下面这段代码向模型要一段 Pydantic 用法示例并把流式文本逐字打印到终端async with agent.run_stream(Show me a short example of using Pydantic.) as result: async for text in result.stream_text(): print(text, end, flushTrue)验证方式终端里应看到文字一个字一个字地冒出来而不是等几秒后整段出现完整可运行版本用 rich 渲染 Markdown在 examples/pydantic_ai_examples/stream_markdown.py。最常见报错UserError: stream_text() can only be used with text responses。原因是 agent 配置了结构化输出文本流接口拒绝工作。处理改用任务二的stream_output()。任务二让结构化数据边生成边展示适用场景输出是 Pydantic 模型或 TypedDict 列表行情表、监控数据、报表行希望第一行到达就先上屏。关键参数debounce_by结构化输出每收一个 token 都可能触发一次校验长输出保持默认 0.1s 或调大减少重复校验开销。迭代值语义每帧都是当前能解析出的完整结构校验不通过的帧不产出最后一帧为完整校验结果。这段代码让模型生成 5 种鲸鱼的结构化数据边流边打印当前可展示的记录数agent Agent(openai:gpt-5.2, output_typelist[Whale]) async with agent.run_stream(Generate details of 5 species of whale.) as result: async for whales in result.stream_output(debounce_by0.01): print(len(whales), 条记录当前可展示)验证方式数字应从少量逐步涨到 5且最后一帧是完整校验过的list[Whale]官方完整示例rich 表格实时刷新在 examples/pydantic_ai_examples/stream_whales.py。最常见报错末帧抛OutputValidatorException——模型生成的 JSON 残缺或字段不符。处理run_stream(..., retriesN)让框架带着错误信息要求模型重答默认 1 次或放宽 schema把严格字段改为NotRequired部分验证逻辑在 pydantic_ai_slim/pydantic_ai/_output.py。避坑与排错按现象 → 原因 → 处理排查覆盖流式开发里最常踩的坑现象原因处理流式中途卡住几秒没新数据部分校验未通过或 debounce 分组等待中间帧被跳过属预期行为以最后一帧为准长时间无响应再查网络与模型配额最后一帧和前几帧数据不一致末帧用完整校验allow_partialFalse重算永远取最后一帧或await result.get_output()别拿中间帧落库工具/输出函数里再调run_stream_sync报UserError嵌套同步运行可能死锁框架直接拦截同步运行只能放在 agent 运行之外的应用代码说明见 docs/troubleshooting.mdA 模型能流式、B 模型报错或超时部分提供商不支持流式或有限流该路径回退agent.run()拿完整响应或换模型、加retries流中断后拿不到任何结果异常发生在迭代中途见下方兜底写法外层包 try 回退完整请求流断掉时的兜底写法——迭代中途异常就回退到完整请求try: async with agent.run_stream(prompt) as result: async for text in result.stream_text(): print(text, end, flushTrue) except Exception: final await agent.run(prompt) # 回退拿完整响应验证方式本地断开网络后运行确认走的是回退路径且异常没有被抛到终端。另外流级别的await result.cancel()只停当前这次模型响应整个运行的中止用cancellation_token接口见 pydantic_ai_slim/pydantic_ai/result.py。进阶调优合并窗口文本流保持默认 0.1s结构化长输出若 UI 刷新卡顿把debounce_by调大到 0.2s 以上每帧少做几次校验。内存stream_text(deltaFalse)每帧都是至今全文别把每帧都存下来需要累积就用deltaTrue自己拼接。刹车长回复场景给run_stream传usage_limitstoken / 请求次数上限与cancellation_token用户点停止时能真正终止。可观测接入logfire.instrument_pydantic_ai()后每次流式请求的耗时与 token 消耗可按请求拆开查看比肉眼猜卡在哪快得多。下一步行动清单跑通官方两个流式示例uv run -m pydantic_ai_examples.stream_markdown逐字文本与uv run -m pydantic_ai_examples.stream_whales结构化表格在终端确认增量刷新效果需要源码时执行git clone https://gitcode.com/GitHub_Trending/py/pydantic-ai。精读 docs/agent.md 的流式小节与 docs/output.md 的结构化输出部分补齐run_stream_events()事件流用法。把 docs/troubleshooting.md 加入书签遇到UserError、OutputValidatorException时按异常名直接定位。更多项目说明见 README.md带工具调用的流式天气智能体示例在 examples/pydantic_ai_examples/weather_agent.py。【免费下载链接】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），仅供参考