企业官网建设流程全解析

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

模型还在说话，你想把表格实时刷到页面上，结果一跑 Pydantic 校验就报 JSON 不完整——流的第一批内容本来就只是半个对象。Pydantic AI 的run_stream里有三个开关：debounce_by、delta和 yield 语义，决定流式输出能不能用。读完能按场景挑对参数。

流式 yield 为什么是"累计快照"

Pydantic AI 的run_stream不转发裸 token，文本或结构化输出每次 yield 的都是"到目前为止的完整快照"。这样做有两点好处：消费端不用写累加逻辑；每次 yield 会走一次部分校验——pydantic 的 partial 模式容忍未闭合的 JSON（字符串字段按trailing-strings处理），模型输出到一半不会直接报错。

因为校验是按 yield 跑的，stream_output才有debounce_by参数（默认 0.1 秒）合并分块，避免长结构化输出每个 token 都跑一轮校验。最终结果仍要做完整校验：流结束后用result.get_output()拿校验过的对象，定义在 StreamedRunResult。

场景一：纯文本逐字输出（聊天界面）

适用于聊天 UI、终端等需要"边说边显示"的地方。

with Live('', console=console) as live: async with agent.run_stream(prompt, model=model) as result: async for message in result.stream_output(): live.update(Markdown(message))

来源：examples/pydantic_ai_examples/stream_markdown.py

⚠️message是"到目前为止的全文"而不是新片段，直接覆盖 UI 即可，再+=会得到重复串。

场景二：带类型校验的结构化流（仪表盘表格）

输出是列表、表格类型时，Pydantic AI 对每次 yield 做 partial 校验，表格可以逐行实时填充，字段类型不对当场暴露。

class Whale(TypedDict): name: str length: float # ... 其他字段 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): table = Table(title='Species of Whale') # ... 填表行逻辑 live.update(table)

来源：examples/pydantic_ai_examples/stream_whales.py

⚠️debounce_by默认 0.1 秒合并分块；设为None后每个 token 触发一轮 partial 校验，长输出会把 CPU 打满。

场景三：观察外部工具调用进度

流式过程中想知道模型正在调哪个工具，事件监听器是成本最低的方式。FunctionToolCallEvent是AgentStreamEvent联合类型之一，按类型注册后只分发你关心的事件。

@agent.on_event(FunctionToolCallEvent) async def track_tools(ctx: RunContext, event: FunctionToolCallEvent) -> None: print(f'calling {event.part.tool_name}')

来源：docs/hooks.md

⚠️ 监听器看到的是"发出时刻"的事件，如果后续 capability 改写或丢弃了它，你记录的就不是最终投递流；要投递流用run_stream_events()。

踩坑清单

1. 现象：迭代stream_output()后把各次结果拼接，得到重复的长串。原因：默认每次 yield 是"到目前为止的全文"，不是增量片段。修复：直接覆盖 UI；要增量片段改用stream_text(delta=True)。

2. 现象：stream_text()拿到的文本和最终输出格式对不上。原因：它跳过TextOutput后处理，delta=True时也不再调用 result validators。修复：需要处理后文本就用stream_output()。

3. 现象：流式迭代中途读result.usage，数字不完整。原因：usage 要等流结束才会汇总完整。修复：退出run_stream上下文管理器之后再读usage。

4. 现象：@agent.on_event记录的事件和下游 UI 实际收到的不一致。原因：监听器在"发出"环节，capability 之后仍可改写或丢弃事件。修复：需要最终投递流时消费run_stream_events()。

三句话收束：yield 的是累计快照、校验按 yield 做 partial 模式、最终结果以get_output()为准。下一步行动：把 stream_whales 示例 里的debounce_by从 0.01 改回 0.1，对比表格刷新频率与校验开销；完整参数说明见 docs/api/result.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