企业官网建设流程全解析

从 0 到 1：Pydantic 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 AI（Python 的 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_partial=True做部分校验，JSON 不完整时也能给出可解析的中间值；流结束时再做一次allow_partial=False的全量校验，保证最终结果一定通过 Pydantic 模型。

🚀 三步跑通第一个流式响应

最小可运行示例：定义一个 Agent，run_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='', flush=True) # 文本随模型输出逐段刷新到终端 print('\nToken 用量:', result.usage) # 流结束后读取完整统计 asyncio.run(main())

代码示例来源：examples/pydantic_ai_examples/stream_markdown.py

预期输出：Asking: Show me a short example...之后，终端逐段打印出模型生成的 Markdown 代码示例（而非一次性整段出现），最后一行打印本次运行的 token 用量，例如requests=1, input_tokens=84, output_tokens=126, total_tokens=210。

🎯 四种取数姿势：按你的界面选通道

StreamedRunResult提供三条取数方法，按场景取舍：

场景一：聊天界面逐字打字机效果

适用：纯文本回复、终端演示，界面只关心「下一个词」。用stream_text(delta=True)只拿增量，不重复前文：

async with agent.run_stream(prompt) as result: async for text in result.stream_text(delta=True): # 只取新增片段，省去自己 diff ui.append(text)

代码示例来源：examples/pydantic_ai_examples/stream_markdown.py

取舍：delta=True最轻量，但拿到的是裸字符串，无法做结构化处理。

场景二：数据表格边生成边渲染

适用：结构化输出（列表、表格、表单），需要每刷新一行就更新 UI。给 Agent 声明output_type，用stream_output()拿「部分校验成功」的中间对象：

agent = Agent('openai:gpt-5.2', output_type=list[Whale]) # 声明结构化输出类型 async with agent.run_stream('Generate me details of 5 species of Whale.') as result: async for whales in result.stream_output(debounce_by=0.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='', flush=True)

代码示例来源：pydantic_ai_slim/pydantic_ai/result.py

取舍：API 形态与异步版一致，但同步上下文内不能再并发发起其他异步调用。

🛠 五个高频报错的排查路线

1. 现象：stream_output抛出验证异常。原因：中间批次的 JSON 片段恰好切在字段中间，部分校验失败（常见于未声明NotRequired的必填字段）。处理：把可选字段标成NotRequired，并在消费端对中间批次的异常做try/except跳过，只把最后一次产出当权威结果。
2. 现象：前端画面卡顿、掉帧。原因：debounce_by=None关闭合并后，每个分片都触发一次重绘。处理：恢复默认debounce_by=0.1（100ms 窗口），重绘压力高的界面可再放大到 0.2。
3. 现象：迭代几轮后连接关闭、输出不完整。原因：把run_stream的async with块拆散了，上下文提前退出，流被截断。处理：消费循环必须完整写在async with内部；需要中途放弃时显式调用result.cancel()。
4. 现象：stream_text产出的片段拼接后与最终回答不一致。原因：delta=False时每次产出的是「到目前为止的全文」，叠加拼接必然重复。处理：要么统一用delta=True取增量，要么每轮用新值整体替换而不是追加。
5. 现象：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_type=Deps, retries=2) # 工具调用失败自动重试 2 次 async with weather_agent.run_stream( 'What is the weather like in London and in Wiltshire?', deps=deps ) 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…」的步骤提示，而不是白屏等待；retries=2兜住工具端偶发 5xx，最终答复由流结束时的全量校验保证类型正确。

• run_stream()必须在async with中消费，退出即关闭流。
• 三条通道按粒度选：stream_text（词）/stream_output（校验过的对象）/stream_response（原始响应）。
• debounce_by默认 0.1 秒，是「流畅度 vs 重绘开销」的合并窗口，别随手设None。
• 中间批次允许校验失败，最终一次allow_partial=False校验才是权威结果。
• 同步场景用run_sync的同名迭代器，API 形态不变。

延伸阅读：docs/run.md（run_stream 全参数）与 pydantic_ai_slim/pydantic_ai/result.py（StreamedRunResult 源码）。

【免费下载链接】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