企业官网建设流程全解析

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

Pydantic AI 是 Python 生态里的 AI 智能体框架，它能让大模型的回答一边生成一边流式推送给你，并顺手完成 Pydantic 结构化校验。本文适合会基础 Python、想做出"打字机效果"聊天界面或实时数据面板的开发者，跟着走一遍就能把run_stream用熟。

一、先跑起来：10 行代码流式输出

流式输出说白了就是：模型每生成一小段文字，立刻推送到你的屏幕，而不是等全部生成完再一次性给你。Pydantic AI 用agent.run_stream()这个异步上下文管理器提供这条通道，它返回一个StreamedRunResult，上面挂着你最常用的两个方法：

• stream_text()：逐次吐出文本（默认每次是累计全文）
• stream_output()：逐次吐出经过校验的结构化数据

10 行代码看到第一个字

🚀 把下面片段保存成stream_demo.py，配好模型 API Key 即可运行：

from pydantic_ai import Agent agent = Agent('openai:gpt-4o-mini') # 换成你手头有 Key 的模型 async def main(): async with agent.run_stream('用三句话介绍 Pydantic') as result: async for text in result.stream_text(): print(text, end='', flush=True) # 每次迭代拿到累计全文 print('\n用量:', result.usage) # 流结束后统计仍可用 import asyncio; asyncio.run(main())

注意async with块：流的生命周期被框在这个块里，块结束流自动关闭，不用手动清理。更多入口说明见 docs/agent.md。

流结束后再取最终结果

一个容易忽略的细节：result.output在整个流消费完之前是拿不到的。你可以等循环自然结束后再读它，也可以用await result.get_output()在循环里提前拿到已解析的输出。result.usage里的 token 统计同样在流结束后依然有效，方便你记费用。

二、流式结构化数据：边到边校验

如果你的输出不是纯文本，而是表格、列表这类结构化数据，stream_text()就不够用了。此时靠output_type声明结构，Pydantic AI 会在数据流到的过程中对 JSON 做分段校验——先到的字段先验证，不用等整段闭合。

声明输出结构并流式接收

from typing import TypedDict from pydantic_ai import Agent class Whale(TypedDict): name: str length: float # 成年鲸鱼平均长度（米） agent = Agent('openai:gpt-4o-mini', output_type=list[Whale]) async with agent.run_stream('给我 5 种鲸鱼的数据') as result: async for whales in result.stream_output(debounce_by=0.05): print(whales) # 当前已校验通过的"半成品"数据

跑起来的效果：name到了先显示名字，length到了再补上数值，像填表一样一格格亮出来。完整可运行版本参考 examples/pydantic_ai_examples/stream_whales.py。

stream_text 和 stream_output 怎么选

stream_output在数据没凑齐之前，可能抛出校验错误，也可能返回不完整的对象；如果你只要"要么全对、要么别给我"，就改用非流式的agent.run()。

三、出问题时别慌：工具调用、断流与取消

流式场景里最典型的三类状况：模型中途要调工具、网络把流掐断了、用户想提前停止。三种各有各的接法。

想看工具调用过程，换 run_stream_events

run_stream()默认只关心"最终输出"。如果你的智能体要调天气、查数据库之类的工具，想在前端显示"正在查询…"，用run_stream_events()拿原始事件流：

from pydantic_ai import AgentRunResultEvent, FunctionToolCallEvent async with agent.run_stream_events('北京今天天气如何？') as events: async for event in events: if isinstance(event, FunctionToolCallEvent): print('准备调用工具:', event.tool_name) elif isinstance(event, AgentRunResultEvent): print('完成:', event.result.output)

事件流里还有PartStartEvent、PartDeltaEvent、PartEndEvent，想自己拼文本时按这些增量拼接即可。

断流与错误的兜底顺序

• 瞬时网络错误：给 Agent 加retries，框架会自动重试，详见 docs/retries.md

agent = Agent('openai:gpt-4o-mini', retries=3) # 失败最多自动重试 3 次

• 模型输出不合法：结构化流中校验失败会以异常抛出，捕获后可以选择重新发起请求（带上新提示词）或降级为非流式调用
• 长任务状态：把已完成的中间结果写进会话历史（messages），重跑时传回去，相当于断点续传

用户点"停止"怎么办

run_stream_events()返回的句柄上挂着cancel()，在另一个任务（比如 UI 的停止按钮回调）里调用它，继续迭代的协程会收到RunCancelled异常，流被干净地关掉。注意取消后usage统计是尽力而为的，别拿它做精确计费。

四、把节奏调顺：几个实用调优点 📊

跑通之后，体验好坏主要取决于"刷新多频繁"和"资源怎么释放"。以下三点不需要改架构，改参数就行。

几个补充建议：

• 长回答场景给debounce_by设个小值（如0.05），用户几乎无感，渲染压力也小
• 消费完的中间片段及时丢弃，别把每个delta都攒在列表里；async with块退出后流相关资源会随结果对象一起释放
• 想直观看效果，官方示例 examples/pydantic_ai_examples/stream_markdown.py 用 Rich 实时渲染 Markdown，可照着改成自己的 UI

五、选型速查与上线前检查清单 ✅

按需求选入口

上线前检查清单

• 确认async with agent.run_stream(...)覆盖了整个消费循环，避免流悬挂
• 确认模型支持流式与你的输出形态；不支持图像输出等能力时框架会提前抛错
• 给重试（retries）和超时设置上限，防止个别请求拖死整个界面
• 前端渲染加节流，debounce_by与渲染帧率匹配
• 日志里记录result.usage，便于核对 token 成本

把入口、校验、异常、节奏这四块对上，流式功能就稳了。想深入结构化输出的各种形态，可以继续看 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