Pydantic AI 流式输出从首个 token 到完整校验的 4 步实践【免费下载链接】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-aiPydantic AI 的agent.run_stream()是流式输出的入口模型输出不再等整段生成完才返回而是逐块推送到你的业务层首个 token 就能先到前端。装好包之后不用改任何安装配置直接看下面四步。一个场景整段等待和逐块返回的差别假设你在做一个聊天界面用户提问后要干等 5 秒然后一大段文字一次性砸到屏幕上。换成流式推送后界面上文字是长出来的体感延迟明显下降。代价是你要处理两件事中间状态的输出可能不合法半截 JSON 永远校验不过以及什么时候能拿到最终版。Pydantic AI 的流式输出把这两件事封装成了部分校验 最终校验两段式中间块用宽松校验先推结束那一刻再做一次严格校验。四步跑通 run_stream 的流式输出第一步用 run_stream 拿到流式结果对象这段代码在做什么用异步上下文管理器启动流式运行通过stream_text()逐块打印文本最后用get_output()拿完整结果和用量。注意stream_text()默认deltaFalse每次 yield 的是到目前为止的全文不是增量片段。from pydantic_ai import Agent agent Agent(openai:gpt-5.2) async def main(): async with agent.run_stream(What is the capital of the UK?) as response: async for text in response.stream_text(): print(text) print(await response.get_output())示例来源docs/agent.mdget_output()的实现在pydantic_ai_slim/pydantic_ai/result.py第二步结构化数据也能流出来这段代码在做什么定义一个TypedDict作为输出类型模型边生成边被校验每凑出合法的部分就 yield 一次前端表格随数据逐行长出来。stream_output()内部对中间块用allow_partialTrue做校验校验失败的块直接跳过最后一定会再 yield 一次完整校验过的结果。class Whale(TypedDict): name: str length: Annotated[float, Field(descriptionAverage length in meters.)] weight: NotRequired[Annotated[float, Field(..., ge50)]] agent Agent(openai:gpt-5.2, output_typelist[Whale]) async def main(): async with agent.run_stream(Details of 5 species of Whale.) as result: async for whales in result.stream_output(debounce_by0.01): render_table(whales) # your own Rich/HTML rendering示例来源examples/pydantic_ai_examples/stream_whales.py第三步想看到中间事件就接 event_stream_handler这段代码在做什么给run_stream()传一个event_stream_handler在最终输出产生之前观察工具调用、thinking、文本增量等事件。完整的事件类型清单PartStartEvent、PartDeltaEvent、FunctionToolCallEvent、FinalResultEvent等见pydantic_ai_slim/pydantic_ai/messages.py。async def event_stream_handler(ctx, event_stream): async for event in event_stream: if isinstance(event, FunctionToolCallEvent): print(fTool call: {event.part.tool_name}) elif isinstance(event, FinalResultEvent): print(Final result started) async with agent.run_stream(prompt, event_stream_handlerevent_stream_handler) as run: async for text in run.stream_text(): print(text)事件处理示例来源docs/agent.md这里有个容易误会的点run_stream()把第一个匹配输出类型的结果当最终输出模型在最终输出之后生成的工具调用默认不会被执行。如果你的 Agent 必须把工具全部跑完改用agent.run_stream_events()或agent.iter()或者把end_strategy设为graceful/exhaustive。第四步收尾时拿完整输出、消息历史和用量response属性随时能拿当前响应快照流式中state为incomplete结束后为completeusage属性在流结束后才有完整 cost想中途放弃就调cancel()它只停当前这条模型响应整个 run 还在继续。这几个属性都在同一个StreamedRunResult对象上见pydantic_ai_slim/pydantic_ai/result.py。避坑清单症状 → 原因 → 修复症状原因修复结构化输出在中间几块时前端解析报 JSON 错误中间块是部分数据Pydantic 用宽松校验放行半截 JSON 本来就解析不了中间块只做展示用兜底值例子里的…占位只把最后一次yield 当权威结果流式模式下工具没被执行run_stream()遇到首个匹配输出的结果就结束之后的工具调用被丢弃用run_stream_events()/iter()或设置end_strategygraceful调用stream_text()抛UserErrorAgent 的output_type是结构化类型文本流只支持纯文本输出结构化类型改用stream_output()纯文本才用stream_text()deltaTrue时发现校验器没生效文档写明了deltaTrue时 result validators 不会被调用需要校验器参与就用stream_text(deltaFalse)或stream_output()debounce_by流式输出校验频率怎么选debounce_by控制合并多少秒内的块再校验/推送一次默认0.1秒传None表示每个块都触发。它本质是用一点延迟换校验次数结构化输出越长默认 0.1 秒省下的校验开销越明显stream_text()的 docstring 里也专门提了这一点。debounce_by 取值校验触发时机推送块数适用场景None每个块都校验最多短文本、极端低延迟需求0.01约 10ms 合并一次中等表格式结构化输出whales 示例的取值0.1默认约 100ms 合并一次最少常规聊天文本CPU 与体验平衡如果接了 logfire 之类的观测可以直接看每次运行的追踪图确认请求、工具调用和输出的时序跑通examples/pydantic_ai_examples/stream_whales.py之后把debounce_by从0.01改成None对比表格刷新频率和终端的校验日志量你会直观感受到这两个值的差别。更多参数语义见 docs/agent.md 的 Running Agents 一节和 docs/output.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），仅供参考