現(xiàn)與調(diào)參)
Pydantic AI 流式輸出三個場景的最小實(shí)現(xiàn)與調(diào)參【免費(fèi)下載鏈接】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需要把模型回復(fù)逐字推給前端時用run()等整段生成完再渲染用戶只能對著空白界面干等。Pydantic AI 的run_stream()改變了這一點(diǎn)模型吐出第一個字符就會推給你的 UI結(jié)構(gòu)化輸出也在接收過程中按 Schema 邊校驗(yàn)邊返回。一句話原理模型以異步迭代器逐塊吐出 tokenrun_stream()的流層邊收邊累加。結(jié)構(gòu)化輸出則用 Pydantic 的 partial部分模式逐次按 Schema 校驗(yàn)所以 JSON 還沒閉合你就能看到第一行有效數(shù)據(jù)。按場景選流純文本逐字渲染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每次只給新增片段適合往緩沖區(qū)追加要截至今時刻的全文就保持默認(rèn)Falsedebounce_by默認(rèn) 0.1 秒純文本場景可傳None關(guān)閉合并拿到最低延遲run_stream接受model參數(shù)臨時換模型agent.run_stream(prompt, modelopenai:gpt-5.2)純 Markdown 渲染參考示例里的Liverich.Markdown組合結(jié)構(gòu)化數(shù)據(jù)實(shí)時表格stream_output 與 debounce_by輸出是列表或?qū)ο?、又想讓第一行?shù)據(jù)一可用就上屏?xí)r給 agent 指定output_type然后迭代stream_output()——每次 yield 的都是一份已按 Schema 校驗(yàn)過的當(dāng)前數(shù)據(jù)。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 都是校驗(yàn)后的快照示例來源examples/pydantic_ai_examples/stream_whales.pydebounce_by控制分塊合并間隔越大校驗(yàn)次數(shù)越少、越省 CPU但上屏越慢校驗(yàn)器跑在 partial 模式靠前的 yield 可能缺可選項渲染前對字段做get兜底最后一次 yield 才是完整數(shù)據(jù)落庫、寫緩存以它為準(zhǔn)長輸出幾十條記錄建議從 0.2 起調(diào)0.1 秒級間隔下終端會明顯閃爍工具調(diào)用中途出結(jié)果stream_response 看過程 agent 中途要調(diào)工具查天氣、查庫而你希望實(shí)時顯示它調(diào)了什么工具用stream_response()。它 yield 的是ModelResponse快照從parts里挑出工具調(diào)用部分即可不需要等最終答案。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正在調(diào)用工具: {part.tool_name}, 參數(shù) {part.args})實(shí)現(xiàn)參考result.py 中的stream_responseresp.state在流進(jìn)行中是incomplete最后一個是complete工具返回后流會繼續(xù)跑結(jié)束后用result.get_output()拿最終結(jié)構(gòu)化結(jié)果ToolCallPart帶tool_call_id可以配對展示調(diào)用中 → 已返回同一輪里工具調(diào)用和文本增量都會出現(xiàn)在快照中按 part 類型分派即可容易踩的坑如果你發(fā)現(xiàn)結(jié)構(gòu)化流的第一份數(shù)據(jù)缺字段原因是校驗(yàn)器跑在 partial 模式、早到的快照本來就不完整把每次 yield 當(dāng)當(dāng)前進(jìn)度渲染、只信最后一次即可渲染前用whale.get(weight)兜底。如果你看到長結(jié)構(gòu)化輸出時界面瘋狂重繪或 CPU 升高是默認(rèn)debounce_by0.1下每個分塊都觸發(fā)一次 Schema 校驗(yàn)把debounce_by調(diào)到 0.2 以上純文本渲染則傳debounce_byNone少合并、少延遲。如果流到一半拋超時異常大概率是模型側(cè)或網(wǎng)絡(luò)層超時而不是你的代碼 bug給整個run_stream套一層 retrytenacity 之類流不能斷點(diǎn)續(xù)傳重試就是整輪重跑。如果你迭代完stream_output()之后再迭代一次拿到空是因?yàn)榱魇且淮涡援惒降?、消費(fèi)完結(jié)果即標(biāo)記完成需要的中間快照要在循環(huán)內(nèi)自己收集別指望第二次迭代。性能與內(nèi)存debounce_by默認(rèn) 0.1s長結(jié)構(gòu)化輸出保持 0.1s 以上減少逐塊校驗(yàn)開銷純文本渲染傳None拿最低延遲stream_text(delta...)往緩沖區(qū)增量追加用True每次整體重繪如富文本 Live 區(qū)用Falsepartial 校驗(yàn)是結(jié)構(gòu)化流內(nèi)置的無法關(guān)閉schema 層級深時靠調(diào)大debounce_by攤薄校驗(yàn)次數(shù)流式結(jié)果會在內(nèi)存里累加完整ModelResponse長對話持續(xù)上漲用戶離開頁面時調(diào)result.cancel()中斷流并及時釋放下一步先跑通 examples/pydantic_ai_examples/stream_whales.py 看表格逐行填充再回來看debounce_by和delta兩個參數(shù)怎么調(diào)。【免費(fèi)下載鏈接】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創(chuàng)作聲明：本文部分內(nèi)容由AI輔助生成（AIGC），僅供參考 Pydantic AI 流式輸出：三個場景的最小實(shí)現(xiàn)與調(diào)參
Pydantic AI 流式輸出三個場景的最小實(shí)現(xiàn)與調(diào)參【免費(fèi)下載鏈接】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需要把模型回復(fù)逐字推給前端時用run()等整段生成完再渲染用戶只能對著空白界面干等。Pydantic AI 的run_stream()改變了這一點(diǎn)模型吐出第一個字符就會推給你的 UI結(jié)構(gòu)化輸出也在接收過程中按 Schema 邊校驗(yàn)邊返回。一句話原理模型以異步迭代器逐塊吐出 tokenrun_stream()的流層邊收邊累加。結(jié)構(gòu)化輸出則用 Pydantic 的 partial部分模式逐次按 Schema 校驗(yàn)所以 JSON 還沒閉合你就能看到第一行有效數(shù)據(jù)。按場景選流純文本逐字渲染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每次只給新增片段適合往緩沖區(qū)追加要截至今時刻的全文就保持默認(rèn)Falsedebounce_by默認(rèn) 0.1 秒純文本場景可傳None關(guān)閉合并拿到最低延遲run_stream接受model參數(shù)臨時換模型agent.run_stream(prompt, modelopenai:gpt-5.2)純 Markdown 渲染參考示例里的Liverich.Markdown組合結(jié)構(gòu)化數(shù)據(jù)實(shí)時表格stream_output 與 debounce_by輸出是列表或?qū)ο?、又想讓第一行?shù)據(jù)一可用就上屏?xí)r給 agent 指定output_type然后迭代stream_output()——每次 yield 的都是一份已按 Schema 校驗(yàn)過的當(dāng)前數(shù)據(jù)。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 都是校驗(yàn)后的快照示例來源examples/pydantic_ai_examples/stream_whales.pydebounce_by控制分塊合并間隔越大校驗(yàn)次數(shù)越少、越省 CPU但上屏越慢校驗(yàn)器跑在 partial 模式靠前的 yield 可能缺可選項渲染前對字段做get兜底最后一次 yield 才是完整數(shù)據(jù)落庫、寫緩存以它為準(zhǔn)長輸出幾十條記錄建議從 0.2 起調(diào)0.1 秒級間隔下終端會明顯閃爍工具調(diào)用中途出結(jié)果stream_response 看過程 agent 中途要調(diào)工具查天氣、查庫而你希望實(shí)時顯示它調(diào)了什么工具用stream_response()。它 yield 的是ModelResponse快照從parts里挑出工具調(diào)用部分即可不需要等最終答案。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正在調(diào)用工具: {part.tool_name}, 參數(shù) {part.args})實(shí)現(xiàn)參考result.py 中的stream_responseresp.state在流進(jìn)行中是incomplete最后一個是complete工具返回后流會繼續(xù)跑結(jié)束后用result.get_output()拿最終結(jié)構(gòu)化結(jié)果ToolCallPart帶tool_call_id可以配對展示調(diào)用中 → 已返回同一輪里工具調(diào)用和文本增量都會出現(xiàn)在快照中按 part 類型分派即可容易踩的坑如果你發(fā)現(xiàn)結(jié)構(gòu)化流的第一份數(shù)據(jù)缺字段原因是校驗(yàn)器跑在 partial 模式、早到的快照本來就不完整把每次 yield 當(dāng)當(dāng)前進(jìn)度渲染、只信最后一次即可渲染前用whale.get(weight)兜底。如果你看到長結(jié)構(gòu)化輸出時界面瘋狂重繪或 CPU 升高是默認(rèn)debounce_by0.1下每個分塊都觸發(fā)一次 Schema 校驗(yàn)把debounce_by調(diào)到 0.2 以上純文本渲染則傳debounce_byNone少合并、少延遲。如果流到一半拋超時異常大概率是模型側(cè)或網(wǎng)絡(luò)層超時而不是你的代碼 bug給整個run_stream套一層 retrytenacity 之類流不能斷點(diǎn)續(xù)傳重試就是整輪重跑。如果你迭代完stream_output()之后再迭代一次拿到空是因?yàn)榱魇且淮涡援惒降?、消費(fèi)完結(jié)果即標(biāo)記完成需要的中間快照要在循環(huán)內(nèi)自己收集別指望第二次迭代。性能與內(nèi)存debounce_by默認(rèn) 0.1s長結(jié)構(gòu)化輸出保持 0.1s 以上減少逐塊校驗(yàn)開銷純文本渲染傳None拿最低延遲stream_text(delta...)往緩沖區(qū)增量追加用True每次整體重繪如富文本 Live 區(qū)用Falsepartial 校驗(yàn)是結(jié)構(gòu)化流內(nèi)置的無法關(guān)閉schema 層級深時靠調(diào)大debounce_by攤薄校驗(yàn)次數(shù)流式結(jié)果會在內(nèi)存里累加完整ModelResponse長對話持續(xù)上漲用戶離開頁面時調(diào)result.cancel()中斷流并及時釋放下一步先跑通 examples/pydantic_ai_examples/stream_whales.py 看表格逐行填充再回來看debounce_by和delta兩個參數(shù)怎么調(diào)。【免費(fèi)下載鏈接】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創(chuàng)作聲明：本文部分內(nèi)容由AI輔助生成（AIGC），僅供參考