尧图网页设计 
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需要把模型回复逐字推给前端时用run()等整段生成完再渲染用户只能对着空白界面干等。Pydantic AI 的run_stream()改变了这一点模型吐出第一个字符就会推给你的 UI结构化输出也在接收过程中按 Schema 边校验边返回。一句话原理模型以异步迭代器逐块吐出 tokenrun_stream()的流层边收边累加。结构化输出则用 Pydantic 的 partial部分模式逐次按 Schema 校验所以 JSON 还没闭合你就能看到第一行有效数据。按场景选流纯文本逐字渲染run_stream 最小用法你的界面只需要文本增量聊天窗口、终端时不用定义 output_type直接用stream_text()。from pydantic_ai import Agent agent Agent() async with agent.run_stream(简要介绍 Pydantic 项目) as result: async for text in result.stream_text(deltaTrue): print(text, end, flushTrue)示例来源examples/pydantic_ai_examples/stream_markdown.pydeltaTrue每次只给新增片段适合往缓冲区追加要截至今时刻的全文就保持默认Falsedebounce_by默认 0.1 秒纯文本场景可传None关闭合并拿到最低延迟run_stream接受model参数临时换模型agent.run_stream(prompt, modelopenai:gpt-5.2)纯 Markdown 渲染参考示例里的Liverich.Markdown组合结构化数据实时表格stream_output 与 debounce_by输出是列表或对象、又想让第一行数据一可用就上屏时给 agent 指定output_type然后迭代stream_output()——每次 yield 的都是一份已按 Schema 校验过的当前数据。class Whale(TypedDict): name: str length: float agent Agent(output_typelist[Whale]) async with agent.run_stream(给我 5 种鲸鱼的信息) as result: async for whales in result.stream_output(debounce_by0.1): render_table(whales) # 每次 yield 都是校验后的快照示例来源examples/pydantic_ai_examples/stream_whales.pydebounce_by控制分块合并间隔越大校验次数越少、越省 CPU但上屏越慢校验器跑在 partial 模式靠前的 yield 可能缺可选项渲染前对字段做get兜底最后一次 yield 才是完整数据落库、写缓存以它为准长输出几十条记录建议从 0.2 起调0.1 秒级间隔下终端会明显闪烁工具调用中途出结果stream_response 看过程 agent 中途要调工具查天气、查库而你希望实时显示它调了什么工具用stream_response()。它 yield 的是ModelResponse快照从parts里挑出工具调用部分即可不需要等最终答案。from pydantic_ai.messages import ToolCallPart agent Agent() async with agent.run_stream(上海今天天气怎么样) as result: async for resp in result.stream_response(): for part in resp.parts: if isinstance(part, ToolCallPart): print(f正在调用工具: {part.tool_name}, 参数 {part.args})实现参考result.py 中的stream_responseresp.state在流进行中是incomplete最后一个是complete工具返回后流会继续跑结束后用result.get_output()拿最终结构化结果ToolCallPart带tool_call_id可以配对展示调用中 → 已返回同一轮里工具调用和文本增量都会出现在快照中按 part 类型分派即可容易踩的坑如果你发现结构化流的第一份数据缺字段原因是校验器跑在 partial 模式、早到的快照本来就不完整把每次 yield 当当前进度渲染、只信最后一次即可渲染前用whale.get(weight)兜底。如果你看到长结构化输出时界面疯狂重绘或 CPU 升高是默认debounce_by0.1下每个分块都触发一次 Schema 校验把debounce_by调到 0.2 以上纯文本渲染则传debounce_byNone少合并、少延迟。如果流到一半抛超时异常大概率是模型侧或网络层超时而不是你的代码 bug给整个run_stream套一层 retrytenacity 之类流不能断点续传重试就是整轮重跑。如果你迭代完stream_output()之后再迭代一次拿到空是因为流是一次性异步迭代器、消费完结果即标记完成需要的中间快照要在循环内自己收集别指望第二次迭代。性能与内存debounce_by默认 0.1s长结构化输出保持 0.1s 以上减少逐块校验开销纯文本渲染传None拿最低延迟stream_text(delta...)往缓冲区增量追加用True每次整体重绘如富文本 Live 区用Falsepartial 校验是结构化流内置的无法关闭schema 层级深时靠调大debounce_by摊薄校验次数流式结果会在内存里累加完整ModelResponse长对话持续上涨用户离开页面时调result.cancel()中断流并及时释放下一步先跑通 examples/pydantic_ai_examples/stream_whales.py 看表格逐行填充再回来看debounce_by和delta两个参数怎么调。【免费下载链接】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需要把模型回复逐字推给前端时用run()等整段生成完再渲染用户只能对着空白界面干等。Pydantic AI 的run_stream()改变了这一点模型吐出第一个字符就会推给你的 UI结构化输出也在接收过程中按 Schema 边校验边返回。一句话原理模型以异步迭代器逐块吐出 tokenrun_stream()的流层边收边累加。结构化输出则用 Pydantic 的 partial部分模式逐次按 Schema 校验所以 JSON 还没闭合你就能看到第一行有效数据。按场景选流纯文本逐字渲染run_stream 最小用法你的界面只需要文本增量聊天窗口、终端时不用定义 output_type直接用stream_text()。from pydantic_ai import Agent agent Agent() async with agent.run_stream(简要介绍 Pydantic 项目) as result: async for text in result.stream_text(deltaTrue): print(text, end, flushTrue)示例来源examples/pydantic_ai_examples/stream_markdown.pydeltaTrue每次只给新增片段适合往缓冲区追加要截至今时刻的全文就保持默认Falsedebounce_by默认 0.1 秒纯文本场景可传None关闭合并拿到最低延迟run_stream接受model参数临时换模型agent.run_stream(prompt, modelopenai:gpt-5.2)纯 Markdown 渲染参考示例里的Liverich.Markdown组合结构化数据实时表格stream_output 与 debounce_by输出是列表或对象、又想让第一行数据一可用就上屏时给 agent 指定output_type然后迭代stream_output()——每次 yield 的都是一份已按 Schema 校验过的当前数据。class Whale(TypedDict): name: str length: float agent Agent(output_typelist[Whale]) async with agent.run_stream(给我 5 种鲸鱼的信息) as result: async for whales in result.stream_output(debounce_by0.1): render_table(whales) # 每次 yield 都是校验后的快照示例来源examples/pydantic_ai_examples/stream_whales.pydebounce_by控制分块合并间隔越大校验次数越少、越省 CPU但上屏越慢校验器跑在 partial 模式靠前的 yield 可能缺可选项渲染前对字段做get兜底最后一次 yield 才是完整数据落库、写缓存以它为准长输出几十条记录建议从 0.2 起调0.1 秒级间隔下终端会明显闪烁工具调用中途出结果stream_response 看过程 agent 中途要调工具查天气、查库而你希望实时显示它调了什么工具用stream_response()。它 yield 的是ModelResponse快照从parts里挑出工具调用部分即可不需要等最终答案。from pydantic_ai.messages import ToolCallPart agent Agent() async with agent.run_stream(上海今天天气怎么样) as result: async for resp in result.stream_response(): for part in resp.parts: if isinstance(part, ToolCallPart): print(f正在调用工具: {part.tool_name}, 参数 {part.args})实现参考result.py 中的stream_responseresp.state在流进行中是incomplete最后一个是complete工具返回后流会继续跑结束后用result.get_output()拿最终结构化结果ToolCallPart带tool_call_id可以配对展示调用中 → 已返回同一轮里工具调用和文本增量都会出现在快照中按 part 类型分派即可容易踩的坑如果你发现结构化流的第一份数据缺字段原因是校验器跑在 partial 模式、早到的快照本来就不完整把每次 yield 当当前进度渲染、只信最后一次即可渲染前用whale.get(weight)兜底。如果你看到长结构化输出时界面疯狂重绘或 CPU 升高是默认debounce_by0.1下每个分块都触发一次 Schema 校验把debounce_by调到 0.2 以上纯文本渲染则传debounce_byNone少合并、少延迟。如果流到一半抛超时异常大概率是模型侧或网络层超时而不是你的代码 bug给整个run_stream套一层 retrytenacity 之类流不能断点续传重试就是整轮重跑。如果你迭代完stream_output()之后再迭代一次拿到空是因为流是一次性异步迭代器、消费完结果即标记完成需要的中间快照要在循环内自己收集别指望第二次迭代。性能与内存debounce_by默认 0.1s长结构化输出保持 0.1s 以上减少逐块校验开销纯文本渲染传None拿最低延迟stream_text(delta...)往缓冲区增量追加用True每次整体重绘如富文本 Live 区用Falsepartial 校验是结构化流内置的无法关闭schema 层级深时靠调大debounce_by摊薄校验次数流式结果会在内存里累加完整ModelResponse长对话持续上涨用户离开页面时调result.cancel()中断流并及时释放下一步先跑通 examples/pydantic_ai_examples/stream_whales.py 看表格逐行填充再回来看debounce_by和delta两个参数怎么调。【免费下载链接】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），仅供参考