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(, consoleconsole) as live: async with agent.run_stream(prompt, modelmodel) 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_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): table Table(titleSpecies 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(fcalling {event.part.tool_name})来源docs/hooks.md⚠️ 监听器看到的是发出时刻的事件如果后续 capability 改写或丢弃了它你记录的就不是最终投递流要投递流用run_stream_events()。踩坑清单现象迭代stream_output()后把各次结果拼接得到重复的长串。原因默认每次 yield 是到目前为止的全文不是增量片段。修复直接覆盖 UI要增量片段改用stream_text(deltaTrue)。现象stream_text()拿到的文本和最终输出格式对不上。原因它跳过TextOutput后处理deltaTrue时也不再调用 result validators。修复需要处理后文本就用stream_output()。现象流式迭代中途读result.usage数字不完整。原因usage 要等流结束才会汇总完整。修复退出run_stream上下文管理器之后再读usage。现象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创作声明：本文部分内容由AI辅助生成（AIGC），仅供参考 Pydantic AI 流式输出三个场景不翻车：逐字、结构化校验与工具回调
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(, consoleconsole) as live: async with agent.run_stream(prompt, modelmodel) 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_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): table Table(titleSpecies 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(fcalling {event.part.tool_name})来源docs/hooks.md⚠️ 监听器看到的是发出时刻的事件如果后续 capability 改写或丢弃了它你记录的就不是最终投递流要投递流用run_stream_events()。踩坑清单现象迭代stream_output()后把各次结果拼接得到重复的长串。原因默认每次 yield 是到目前为止的全文不是增量片段。修复直接覆盖 UI要增量片段改用stream_text(deltaTrue)。现象stream_text()拿到的文本和最终输出格式对不上。原因它跳过TextOutput后处理deltaTrue时也不再调用 result validators。修复需要处理后文本就用stream_output()。现象流式迭代中途读result.usage数字不完整。原因usage 要等流结束才会汇总完整。修复退出run_stream上下文管理器之后再读usage。现象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创作声明：本文部分内容由AI辅助生成（AIGC），仅供参考