从 0 到 1Pydantic 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当你把 AI 对话做成 Web 页面时用户盯着一个转圈图标干等 5 到 20 秒体感很差——而模型其实从第一个词开始就在吐数据只是你没接住。Pydantic AIPython 的 AI Agent 框架强类型端到端把「接住这些词」这件事收敛为一条run_stream调用它打开一个异步上下文结果对象通过异步迭代器持续推送增量直到最终输出通过 Pydantic 校验。本文带你用 1 个最小示例跑通流式响应掌握 4 种取数姿势、5 个高频排错点把首字延迟从「等完整响应」压到几百毫秒。 一次 run_stream 调用里发生了什么一句话结论run_stream()返回的是流式结果对象StreamedRunResult它把模型的事件流分组、合并、再逐份校验后交给你校验采用「先接受部分结果、最后一次全量确认」的两段式策略。三个关键组件事件流stream_response最底层的取数通道产出模型响应的原始分片文本增量、工具调用等适合需要观察中间步骤的场景。时序分组器debounce_by默认 0.1 秒的合并窗口把高频分片攒成一批再产出避免 UI 每 10 毫秒重绘一次。输出处理器validate_response_output中间批次以allow_partialTrue做部分校验JSON 不完整时也能给出可解析的中间值流结束时再做一次allow_partialFalse的全量校验保证最终结果一定通过 Pydantic 模型。 三步跑通第一个流式响应最小可运行示例定义一个 Agentrun_stream打开上下文用stream_output()迭代增量输出本例无output_type输出即纯文本import asyncio from pydantic_ai import Agent agent Agent() # 不指定模型时走默认配置可用 model 参数覆盖 async def main(): prompt Show me a short example of using Pydantic. async with agent.run_stream(prompt) as result: # 打开流式上下文退出即关闭连接 async for message in result.stream_output(): # 逐批产出已部分校验的输出 print(message, end, flushTrue) # 文本随模型输出逐段刷新到终端 print(\nToken 用量:, result.usage) # 流结束后读取完整统计 asyncio.run(main())代码示例来源examples/pydantic_ai_examples/stream_markdown.py预期输出Asking: Show me a short example...之后终端逐段打印出模型生成的 Markdown 代码示例而非一次性整段出现最后一行打印本次运行的 token 用量例如requests1, input_tokens84, output_tokens126, total_tokens210。 四种取数姿势按你的界面选通道StreamedRunResult提供三条取数方法按场景取舍场景一聊天界面逐字打字机效果适用纯文本回复、终端演示界面只关心「下一个词」。用stream_text(deltaTrue)只拿增量不重复前文async with agent.run_stream(prompt) as result: async for text in result.stream_text(deltaTrue): # 只取新增片段省去自己 diff ui.append(text)代码示例来源examples/pydantic_ai_examples/stream_markdown.py取舍deltaTrue最轻量但拿到的是裸字符串无法做结构化处理。场景二数据表格边生成边渲染适用结构化输出列表、表格、表单需要每刷新一行就更新 UI。给 Agent 声明output_type用stream_output()拿「部分校验成功」的中间对象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): # 10ms 合并窗口控制刷新频率 render_table(whales) # 每批重绘表格缺省字段显示占位符代码示例来源examples/pydantic_ai_examples/stream_whales.py取舍部分校验靠 Pydantic 的NotRequired字段撑住不完整 JSON换来「数据没齐也能先画」。场景三审计与监控中间步骤适用多轮工具调用的 Agent需要把「调了哪个工具、模型说了什么」落到监控面板。用stream_response()直接消费原始响应对象async with agent.run_stream(prompt) as result: async for response in result.stream_response(): # 每个响应对象含 parts文本/工具调用 log_event(response.parts) # 自行解析 parts 记录工具名与参数代码示例来源pydantic_ai_slim/pydantic_ai/result.py取舍拿到的是模型响应全量对象信息最完整但解析 parts 的样板代码也最多。场景四同步代码里嵌流式输出适用在普通同步函数如脚本、非 async 的 API 层里消费流。同步版本的StreamedRunResultSync提供同名方法返回同步迭代器with agent.run_sync(prompt) as result: for text in result.stream_text(): # 同步迭代器内部桥接事件循环 print(text, end, flushTrue)代码示例来源pydantic_ai_slim/pydantic_ai/result.py取舍API 形态与异步版一致但同步上下文内不能再并发发起其他异步调用。维度stream_textstream_outputstream_response同步版适用场景纯文本打字机结构化表格/表单工具调用审计、监控同步调用栈内嵌流复杂度低中需定义 output_type高要解析 parts低性能最轻纯字符串中含逐批部分校验中无校验开销但对象重略高于对应异步版选择建议界面只显示文字时首选输出可 Pydantic 化时首选需要完整链路可观测时无法改 async 时用 五个高频报错的排查路线现象stream_output抛出验证异常。原因中间批次的 JSON 片段恰好切在字段中间部分校验失败常见于未声明NotRequired的必填字段。处理把可选字段标成NotRequired并在消费端对中间批次的异常做try/except跳过只把最后一次产出当权威结果。现象前端画面卡顿、掉帧。原因debounce_byNone关闭合并后每个分片都触发一次重绘。处理恢复默认debounce_by0.1100ms 窗口重绘压力高的界面可再放大到 0.2。现象迭代几轮后连接关闭、输出不完整。原因把run_stream的async with块拆散了上下文提前退出流被截断。处理消费循环必须完整写在async with内部需要中途放弃时显式调用result.cancel()。现象stream_text产出的片段拼接后与最终回答不一致。原因deltaFalse时每次产出的是「到目前为止的全文」叠加拼接必然重复。处理要么统一用deltaTrue取增量要么每轮用新值整体替换而不是追加。现象UserError: Image output is not supported by this model.一类报错。原因输出类型要求如图像超过所选模型的ModelProfile能力声明。处理换支持对应输出类型的模型或在构建 Agent 前用 profile 检查能力位而不是等运行时报错。 实战给天气 Agent 加上边算边说把前文技巧串成一个真实形态多城市天气查询。Agent 依次调用get_lat_lng和get_weather两个工具代码见 examples/pydantic_ai_examples/weather_agent.py流式改造只需两步weather_agent Agent(openai:gpt-5-mini, deps_typeDeps, retries2) # 工具调用失败自动重试 2 次 async with weather_agent.run_stream( What is the weather like in London and in Wiltshire?, depsdeps ) as result: async for response in result.stream_response(): # 工具调用阶段实时播报中间步骤 for part in response.parts: if part.tool_name: ui.note(f正在执行 {part.tool_name}...) ui.finish(result.output) # 流结束拿最终全量校验过的答复代码示例来源examples/pydantic_ai_examples/weather_agent.py效果工具执行的 10 到 30 秒空窗期里界面持续滚动「正在执行 get_lat_lng… 正在执行 get_weather…」的步骤提示而不是白屏等待retries2兜住工具端偶发 5xx最终答复由流结束时的全量校验保证类型正确。run_stream()必须在async with中消费退出即关闭流。三条通道按粒度选stream_text词/stream_output校验过的对象/stream_response原始响应。debounce_by默认 0.1 秒是「流畅度 vs 重绘开销」的合并窗口别随手设None。中间批次允许校验失败最终一次allow_partialFalse校验才是权威结果。同步场景用run_sync的同名迭代器API 形态不变。延伸阅读docs/run.mdrun_stream 全参数与 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），仅供参考 从 0 到 1：Pydantic AI 流式输出与实时结果获取完整指南
从 0 到 1Pydantic 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当你把 AI 对话做成 Web 页面时用户盯着一个转圈图标干等 5 到 20 秒体感很差——而模型其实从第一个词开始就在吐数据只是你没接住。Pydantic AIPython 的 AI Agent 框架强类型端到端把「接住这些词」这件事收敛为一条run_stream调用它打开一个异步上下文结果对象通过异步迭代器持续推送增量直到最终输出通过 Pydantic 校验。本文带你用 1 个最小示例跑通流式响应掌握 4 种取数姿势、5 个高频排错点把首字延迟从「等完整响应」压到几百毫秒。 一次 run_stream 调用里发生了什么一句话结论run_stream()返回的是流式结果对象StreamedRunResult它把模型的事件流分组、合并、再逐份校验后交给你校验采用「先接受部分结果、最后一次全量确认」的两段式策略。三个关键组件事件流stream_response最底层的取数通道产出模型响应的原始分片文本增量、工具调用等适合需要观察中间步骤的场景。时序分组器debounce_by默认 0.1 秒的合并窗口把高频分片攒成一批再产出避免 UI 每 10 毫秒重绘一次。输出处理器validate_response_output中间批次以allow_partialTrue做部分校验JSON 不完整时也能给出可解析的中间值流结束时再做一次allow_partialFalse的全量校验保证最终结果一定通过 Pydantic 模型。 三步跑通第一个流式响应最小可运行示例定义一个 Agentrun_stream打开上下文用stream_output()迭代增量输出本例无output_type输出即纯文本import asyncio from pydantic_ai import Agent agent Agent() # 不指定模型时走默认配置可用 model 参数覆盖 async def main(): prompt Show me a short example of using Pydantic. async with agent.run_stream(prompt) as result: # 打开流式上下文退出即关闭连接 async for message in result.stream_output(): # 逐批产出已部分校验的输出 print(message, end, flushTrue) # 文本随模型输出逐段刷新到终端 print(\nToken 用量:, result.usage) # 流结束后读取完整统计 asyncio.run(main())代码示例来源examples/pydantic_ai_examples/stream_markdown.py预期输出Asking: Show me a short example...之后终端逐段打印出模型生成的 Markdown 代码示例而非一次性整段出现最后一行打印本次运行的 token 用量例如requests1, input_tokens84, output_tokens126, total_tokens210。 四种取数姿势按你的界面选通道StreamedRunResult提供三条取数方法按场景取舍场景一聊天界面逐字打字机效果适用纯文本回复、终端演示界面只关心「下一个词」。用stream_text(deltaTrue)只拿增量不重复前文async with agent.run_stream(prompt) as result: async for text in result.stream_text(deltaTrue): # 只取新增片段省去自己 diff ui.append(text)代码示例来源examples/pydantic_ai_examples/stream_markdown.py取舍deltaTrue最轻量但拿到的是裸字符串无法做结构化处理。场景二数据表格边生成边渲染适用结构化输出列表、表格、表单需要每刷新一行就更新 UI。给 Agent 声明output_type用stream_output()拿「部分校验成功」的中间对象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): # 10ms 合并窗口控制刷新频率 render_table(whales) # 每批重绘表格缺省字段显示占位符代码示例来源examples/pydantic_ai_examples/stream_whales.py取舍部分校验靠 Pydantic 的NotRequired字段撑住不完整 JSON换来「数据没齐也能先画」。场景三审计与监控中间步骤适用多轮工具调用的 Agent需要把「调了哪个工具、模型说了什么」落到监控面板。用stream_response()直接消费原始响应对象async with agent.run_stream(prompt) as result: async for response in result.stream_response(): # 每个响应对象含 parts文本/工具调用 log_event(response.parts) # 自行解析 parts 记录工具名与参数代码示例来源pydantic_ai_slim/pydantic_ai/result.py取舍拿到的是模型响应全量对象信息最完整但解析 parts 的样板代码也最多。场景四同步代码里嵌流式输出适用在普通同步函数如脚本、非 async 的 API 层里消费流。同步版本的StreamedRunResultSync提供同名方法返回同步迭代器with agent.run_sync(prompt) as result: for text in result.stream_text(): # 同步迭代器内部桥接事件循环 print(text, end, flushTrue)代码示例来源pydantic_ai_slim/pydantic_ai/result.py取舍API 形态与异步版一致但同步上下文内不能再并发发起其他异步调用。维度stream_textstream_outputstream_response同步版适用场景纯文本打字机结构化表格/表单工具调用审计、监控同步调用栈内嵌流复杂度低中需定义 output_type高要解析 parts低性能最轻纯字符串中含逐批部分校验中无校验开销但对象重略高于对应异步版选择建议界面只显示文字时首选输出可 Pydantic 化时首选需要完整链路可观测时无法改 async 时用 五个高频报错的排查路线现象stream_output抛出验证异常。原因中间批次的 JSON 片段恰好切在字段中间部分校验失败常见于未声明NotRequired的必填字段。处理把可选字段标成NotRequired并在消费端对中间批次的异常做try/except跳过只把最后一次产出当权威结果。现象前端画面卡顿、掉帧。原因debounce_byNone关闭合并后每个分片都触发一次重绘。处理恢复默认debounce_by0.1100ms 窗口重绘压力高的界面可再放大到 0.2。现象迭代几轮后连接关闭、输出不完整。原因把run_stream的async with块拆散了上下文提前退出流被截断。处理消费循环必须完整写在async with内部需要中途放弃时显式调用result.cancel()。现象stream_text产出的片段拼接后与最终回答不一致。原因deltaFalse时每次产出的是「到目前为止的全文」叠加拼接必然重复。处理要么统一用deltaTrue取增量要么每轮用新值整体替换而不是追加。现象UserError: Image output is not supported by this model.一类报错。原因输出类型要求如图像超过所选模型的ModelProfile能力声明。处理换支持对应输出类型的模型或在构建 Agent 前用 profile 检查能力位而不是等运行时报错。 实战给天气 Agent 加上边算边说把前文技巧串成一个真实形态多城市天气查询。Agent 依次调用get_lat_lng和get_weather两个工具代码见 examples/pydantic_ai_examples/weather_agent.py流式改造只需两步weather_agent Agent(openai:gpt-5-mini, deps_typeDeps, retries2) # 工具调用失败自动重试 2 次 async with weather_agent.run_stream( What is the weather like in London and in Wiltshire?, depsdeps ) as result: async for response in result.stream_response(): # 工具调用阶段实时播报中间步骤 for part in response.parts: if part.tool_name: ui.note(f正在执行 {part.tool_name}...) ui.finish(result.output) # 流结束拿最终全量校验过的答复代码示例来源examples/pydantic_ai_examples/weather_agent.py效果工具执行的 10 到 30 秒空窗期里界面持续滚动「正在执行 get_lat_lng… 正在执行 get_weather…」的步骤提示而不是白屏等待retries2兜住工具端偶发 5xx最终答复由流结束时的全量校验保证类型正确。run_stream()必须在async with中消费退出即关闭流。三条通道按粒度选stream_text词/stream_output校验过的对象/stream_response原始响应。debounce_by默认 0.1 秒是「流畅度 vs 重绘开销」的合并窗口别随手设None。中间批次允许校验失败最终一次allow_partialFalse校验才是权威结果。同步场景用run_sync的同名迭代器API 形态不变。延伸阅读docs/run.mdrun_stream 全参数与 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），仅供参考