Pydantic AI 與 AG-UI 協議用 Agentic UI 構建人機協同的前端應用【免費下載鏈接】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-aiAG-UIAgent-User Interaction是 CopilotKit 團隊提出的開放協議用于標準化前端應用與 AI Agent 之間的通信方式。本文將以 Pydantic AI 倉庫中的 docs/examples/ag-ui.md 為骨架結合 AG-UI 集成文檔 與倉庫內的完整示例源碼完整講解如何把 Pydantic AI Agent 接入 AG-UI 生態、如何在本地通過 AG-UI Dojo 調試面板逐項驗證 Agentic Chat、Human in the Loop、共享狀態、預測式狀態更新等六大交互范式以及AGUIAdapter在底層如何完成協議轉換。讀完本文你將掌握AG-UI 后端的三種接入方式run_stream、dispatch_request、獨立 Starlette 應用、基于StateDeps的前后端狀態共享、基于工具事件與CustomEvent的流式進度推送以及基于DeferredToolRequests的工具審批interrupt流程。背景為什么需要 AG-UI以及 Pydantic AI 如何接入傳統 AI 應用的界面通常由服務端生成整段對話或整頁內容前端只能被動展示。AG-UI 協議則把通信拆分為**事件Events、消息Messages、狀態管理State、工具Tools**四大概念讓前端可以持有工具、維護共享狀態、接收流式事件從而構建生成式 UIGenerative UI體驗。Pydantic AI 通過AGUIAdapter位于pydantic_ai.ui.ag_ui實現協議適配前端把請求封裝為 AG-UI 的RunAgentInput對象包含消息歷史、狀態、可用工具適配器將其轉換為 Pydantic AI 內部類型交給 Agent 執行Agent 產生的工具調用、狀態更新等事件再被轉換回 AG-UI 事件以Server-Sent EventsSSE流式返回給前端。一次用戶請求可能需要客戶端 UI 與 Pydantic AI 服務端之間的多輪往返取決于工具和事件的需要見 docs/ui/ag-ui.md。該集成最初由 Rocket Science 團隊構建并與 Pydantic AI、CopilotKit 團隊合作貢獻見 AG-UI 集成文檔 中的說明。快速啟動在本地跑通 AG-UI 示例倉庫在 examples/pydantic_ai_examples/ag_ui/main.py 提供了一個基于 FastAPI 的 AG-UI 后端并在 examples/pydantic_ai_examples/ag_ui/init.py 中把每個 Feature 掛載為獨立子應用app FastAPI(titlePydantic AI AG-UI server) app.mount(/agentic_chat, agentic_chat_app, Agentic Chat) app.mount(/agentic_generative_ui, agentic_generative_ui_app, Agentic Generative UI) app.mount(/human_in_the_loop, human_in_the_loop_app, Human in the Loop) app.mount(/predictive_state_updates, predictive_state_updates_app, Predictive State Updates) app.mount(/shared_state, shared_state_app, Shared State) app.mount(/tool_approval, tool_approval_app, Tool Approval (interrupts)) app.mount(/tool_based_generative_ui, tool_based_generative_ui_app, Tool Based Generative UI)前置條件一個 OpenAI API Key已安裝項目依賴并設置好環境變量參見 docs/examples/setup.md需要兩個命令行窗口分別運行前后端。第一步啟動 Pydantic AI AG-UI 后端設置 API Key 并啟動示例后端export OPENAI_API_KEYyour api key python/uv-run -m pydantic_ai_examples.ag_ui__main__.py內部通過 uvicorn 在9000端口啟動服務if __name__ __main__: import uvicorn uvicorn.run(pydantic_ai_examples.ag_ui:app, port9000)第二步運行 AG-UI Dojo 前端AG-UI Dojo 是 AG-UI 官方的調試面板可以逐項演示協議特性克隆 AG-UI 倉庫git clone https://github.com/ag-ui-protocol/ag-ui.git按官方說明安裝前置依賴然后從倉庫根目錄安裝依賴并構建cd ag-ui pnpm i pnpm build --projectsdemo-viewer進入apps/dojo目錄運行 Dojo 應用cd apps/dojo pnpm dev瀏覽器訪問 http://localhost:3000/pydantic-ai在側邊欄選擇Pydantic AI視圖每個 Feature 的訪問地址為http://localhost:3000/pydantic-ai/feature/feature_name下文逐一說明。六大交互范式詳解基于倉庫示例源碼Agentic Chat服務端工具與客戶端工具同場協作這是最基本的 Agent 交互范式演示 Pydantic AI 服務端工具與 AG-UI 客戶端工具如何協同工作。訪問地址http://localhost:3000/pydantic-ai/feature/agentic_chat。該示例包含兩個工具time——Pydantic AI 服務端工具查詢指定時區的當前時間background——AG-UI 客戶端工具修改客戶端窗口的背景色對應的示例源碼為 examples/pydantic_ai_examples/ag_ui/api/agentic_chat.py。服務端工具用agent.tool_plain聲明內部通過zoneinfo.ZoneInfo處理時區并返回 ISO 格式時間agent Agent(openai:gpt-5-mini) agent.tool_plain async def current_time(timezone: str UTC) - str: Get the current time in ISO format. tz: ZoneInfo ZoneInfo(timezone) return datetime.now(tztz).isoformat() async def run_agent(request: Request) - Response: return await AGUIAdapter.dispatch_request(request, agentagent) app Starlette(routes[Route(/, run_agent, methods[POST])])端點本身非常簡潔單個POST /路由調用AGUIAdapter.dispatch_request(request, agentagent)其余協議細節全部由適配器接管。background這類客戶端工具不會出現在服務端代碼里而是由 AG-UI 前端在請求中聲明適配器會把客戶端工具透傳給模型由模型決定何時調用。可以嘗試的提示詞What is the time in New York?Change the background to blue更復雜的混合示例——讓模型在兩個工具間交替執行并計算耗時Perform the following steps, waiting for the response of each step before continuing: 1. Get the time 2. Set the background to red 3. Get the time 4. Report how long the background set took by diffing the two times這個示例直觀展示了 AG-UI 的客戶端工具能力工具執行結果由前端渲染服務端只負責推理決策真正實現了 UI 能力的分布。Agentic Generative UI長任務中的流式狀態更新該示例演示一個長時間運行的任務Agent 邊執行邊把進度推送給前端讓用戶實時看到正在發生什么。訪問地址http://localhost:3000/pydantic-ai/feature/agentic_generative_ui。示例源碼為 examples/pydantic_ai_examples/ag_ui/api/agentic_generative_ui.py。它用 Pydantic 模型描述計劃結構Step單個步驟含description和statuspending/completedPlan步驟列表JSONPatchOpRFC 6902 JSON Patch 操作用于表達狀態增量Agent 的指令強調只使用工具、不輸出多余文字agent Agent( openai:gpt-5-mini, instructionsdedent( When planning use tools only, without any other messages. IMPORTANT: - Use the create_plan tool to set the initial state of the steps - Use the update_plan_step tool to update the status of each step - Do NOT repeat the plan or summarise it in a message ... Only one plan can be active at a time, so do not call the create_plan tool again until all the steps in current plan are completed. ), )兩個核心工具直接返回 AG-UI 事件對象create_plan返回StateSnapshotEvent狀態快照把整個計劃一次性同步給前端update_plan_step返回StateDeltaEvent狀態增量攜帶 JSON Patch 操作數組只推送變更部分agent.tool_plain async def create_plan(steps: list[str]) - StateSnapshotEvent: plan Plan(steps[Step(descriptionstep) for step in steps]) return StateSnapshotEvent(typeEventType.STATE_SNAPSHOT, snapshotplan.model_dump()) agent.tool_plain async def update_plan_step(index: int, description: str | None None, status: StepStatus | None None) - StateDeltaEvent: changes: list[JSONPatchOp] [] if description is not None: changes.append(JSONPatchOp(opreplace, pathf/steps/{index}/description, valuedescription)) if status is not None: changes.append(JSONPatchOp(opreplace, pathf/steps/{index}/status, valuestatus)) return StateDeltaEvent(typeEventType.STATE_DELTA, deltachanges)實現要點Pydantic AI 工具可以直接返回 AG-UI 的BaseEvent或事件迭代器適配器會把這些事件作為工具結果的一部分隨事件流發給前端。這與ctx.emit()的即時事件不同——工具返回事件屬于消息的一部分能隨消息歷史往返適合前端需要重建的狀態更新詳見 docs/ui/ag-ui.md。嘗試提示詞Create a plan for breakfast and execute it前端會看到一個逐步勾選pending → completed的計劃卡片而不是一段純文本回復。Human in the Loop讓用戶審批 Agent 提出的計劃該示例演示簡單的人機協同流程Agent 生成計劃用戶在界面上用復選框確認。訪問地址http://localhost:3000/pydantic-ai/feature/human_in_the_loop。示例源碼為 examples/pydantic_ai_examples/ag_ui/api/human_in_the_loop.py。這個 Feature 依賴的是 AG-UI 的客戶端工具generate_task_steps——它由前端實現用于展示并確認步驟。服務端只需在指令中約束行為agent Agent( openai:gpt-5-mini, instructionsdedent( When planning tasks use tools only, without any other messages. IMPORTANT: - Use the generate_task_steps tool to display the suggested steps to the user - Never repeat the plan, or send a message detailing steps - If accepted, confirm the creation of the plan and the number of selected (enabled) steps only - If not accepted, ask the user for more information, DO NOT use the generate_task_steps tool again ), )嘗試提示詞Generate a list of steps for cleaning a car for me to review值得留意的是該文件 docstring 中的一句話No special handling is required for this feature.——人機協同的核心邏輯完全由 AG-UI 協議客戶端工具承擔服務端只需告訴模型什么時候該用工具、什么時候不該用。Predictive State Updates預測式狀態更新該示例演示如何基于 Agent 的響應預測性更新 UI 狀態包括通過用戶確認進行交互。訪問地址http://localhost:3000/pydantic-ai/feature/predictive_state_updates。示例源碼為 examples/pydantic_ai_examples/ag_ui/api/predictive_state_updates.py。它定義了一個DocumentState含document字段作為前后端共享狀態通過StateDeps注入class DocumentState(BaseModel): State for the document being written. document: str agent Agent(openai:gpt-5-mini, deps_typeStateDeps[DocumentState])關鍵工具document_predict_state返回一個名為PredictState的CustomEvent聲明write_document工具的document參數會更新document狀態鍵——前端據此在工具執行前就預測性地渲染新文檔agent.tool_plain async def document_predict_state() - list[CustomEvent]: Enable document state prediction. return [ CustomEvent( typeEventType.CUSTOM, namePredictState, value[ { state_key: document, tool: write_document, tool_argument: document, }, ], ), ]示例還展示了基于共享狀態的自定義指令agent.instructions()裝飾器把當前文檔內容動態注入指令讓模型接著寫而不是重寫agent.instructions() async def story_instructions(ctx: RunContext[StateDeps[DocumentState]]) - str: return dedent(f... Before you start writing, you MUST call the document_predict_state tool to enable state prediction. To present the document to the user for review, you MUST use the write_document tool. ... This is the current document: {ctx.deps.state.document} )啟動文檔內容為Bruce was a good dog,嘗試提示詞Help me complete my story about bruce the dog, is should be no longer than a sentence.注意請求處理時的一個關鍵細節dispatch_request會就地修改deps.state因此每個請求都要用dataclasses.replace生成獨立副本避免請求間狀態串擾deps StateDeps(DocumentState()) async def run_agent(request: Request) - Response: # dispatch_request mutates deps.state from the request, so give each request its own copy. return await AGUIAdapter.dispatch_request(request, agentagent, depsreplace(deps))Shared State前后端共享狀態該示例演示 UI 與 Agent 之間的狀態共享發送給 Agent 的狀態被一個基于函數的指令檢測到先用自定義 Pydantic 模型校驗數據再據此生成指令讓 Agent 遵循最后通過 AG-UI 工具把結果發回客戶端。訪問地址http://localhost:3000/pydantic-ai/feature/shared_state。示例源碼為 examples/pydantic_ai_examples/ag_ui/api/shared_state.py。它用枚舉定義SkillLevel、SpecialPreferences、CookingTime用Recipe/RecipeSnapshot兩個 Pydantic 模型承載配方結構class RecipeSnapshot(BaseModel): recipe: Recipe Field(default_factoryRecipe, descriptionThe current state of the recipe) agent Agent(openai:gpt-5-mini, deps_typeStateDeps[RecipeSnapshot])展示工具display_recipe返回StateSnapshotEvent把整個配方快照同步給前端以圖形化渲染agent.tool_plain async def display_recipe(recipe: Recipe) - StateSnapshotEvent: Display the recipe to the user. return StateSnapshotEvent( typeEventType.STATE_SNAPSHOT, snapshot{recipe: recipe}, )recipe_instructions同樣基于當前狀態動態生成指令把已有配方以 JSON 形式注入上下文agent.instructions async def recipe_instructions(ctx: RunContext[StateDeps[RecipeSnapshot]]) - str: return dedent(f... - Create a complete recipe using the existing ingredients - Append new ingredients to the existing ones - Use the display_recipe tool to present the recipe to the user - Do NOT repeat the recipe in the message, use the tool instead ... The current state of the recipe is: {ctx.deps.state.recipe.model_dump_json(indent2)} )操作步驟1. 自定義配方的初始設置技能等級、偏好、烹飪時長、食材2. 點擊Improve with AI觀察 Agent 在既有狀態上增量優化配方并通過display_recipe展示。Tool Based Generative UI工具輸出的定制渲染該示例演示帶用戶確認的工具輸出定制渲染。訪問地址http://localhost:3000/pydantic-ai/feature/tool_based_generative_ui。示例源碼為 examples/pydantic_ai_examples/ag_ui/api/tool_based_generative_ui.py。與服務端示例不同這里的generate_haiku是一個 AG-UI 客戶端工具負責以英文和日文雙語卡片形式渲染俳句——定制渲染邏輯完全發生在前端。嘗試提示詞Generate a haiku about formula 1延伸Tool Approval工具審批 / Interrupts除了 Dojo 六大 Feature 之外倉庫還提供了 examples/pydantic_ai_examples/ag_ui/api/tool_approval.py 演示 AG-UI 的 interrupt 生命周期該能力在 docs/ui/ag-ui.md 中有完整說明需要ag-ui-protocol 0.1.19。核心思路用agent.tool_plain(requires_approvalTrue)聲明危險工具并把DeferredToolRequests加入output_type這樣當模型提議調用該工具時運行會暫停而不是報錯agent Agent(openai:gpt-5-mini, output_type[str, DeferredToolRequests]) agent.tool_plain(requires_approvalTrue) def delete_file(path: str) - str: Delete a file. The run pauses here and waits for the user to approve before executing. return fdeleted {path}流程如下模型提議調用 → 適配器以outcome.type interrupt的RUN_FINISHED事件結束 SSE 流outcome.interrupts[]描述每個待審批項 → 前端據此渲染審批 UI → 用戶操作后前端 POST 攜帶resume[]數組ResumeEntry的下一個RunAgentInput。適配器的字段映射與 AG-UI Python SDK 字段名一致總結如下見 docs/ui/ag-ui.mdAG-UI 方向Pydantic AI 來源 / 去向Interrupt.reason對requires_approvalTrue工具恒為tool_callInterrupt.tool_call_id提議調用的ToolCallPart.tool_call_idInterrupt.idfint-{tool_call_id}resume 時還原為 tool_call_idInterrupt.metadataDeferredToolRequests.metadata.get(tool_call_id)payload.approvedTrueToolApprovedpayload.editedArgsToolApproved.override_args整體替換提議參數payload.approvedFalseToolDeniedmessagepayload.reasonstatuscancelledToolDeniedmessageCancelled by user.payload還會依據Interrupt.response_schema校驗approved字段必填editedArgs、reason若給出但類型錯誤即使approvedTrue也會被判定為拒絕。恢復輪次中 Agent 會以原始tool_call_id重新執行工具因此只會發出該 id 的TOOL_CALL_RESULT事件而不會重復TOOL_CALL_START從而保留 AG-UI 規范要求的審計軌跡。底層原語DeferredToolRequests不依賴 AG-UI 也能獨立使用詳見 docs/deferred-tools.md。底層原理AGUIAdapter 的三種接入方式與事件流轉從 docs/ui/ag-ui.md 可知運行基于 AG-UI 輸入的 Agent 有三種方式靈活度從高到低AGUIAdapter.run_stream()對以RunAgentInput實例化的適配器調用運行 Agent 并返回 AG-UI 事件流支持Agent.iter()的可選參數如deps。適合非 Starlette 框架Django、Flask或需要自行加工輸入/輸出的場景。AGUIAdapter.dispatch_request()類方法接收 Starlette 請求如來自 FastAPI直接返回流式 Starlette 響應可逐請求傳入deps如基于已認證用戶。它是from_request()、run_stream()、streaming_response()三者的便捷組合。獨立 Starlette 應用單個/路由調用dispatch_request()同一應用還能以子應用方式掛載到既有 FastAPI見 FastAPI 子應用文檔。最小可用實現方式 3只需十幾行from starlette.applications import Starlette from starlette.requests import Request from starlette.responses import Response from starlette.routing import Route from pydantic_ai import Agent from pydantic_ai.ui.ag_ui import AGUIAdapter agent Agent(openai:gpt-5.2, instructionsBe fun!) async def run_agent(request: Request) - Response: return await AGUIAdapter.dispatch_request(request, agentagent) app Starlette(routes[Route(/, run_agent, methods[POST])])啟動uvicorn ag_ui_app:app若需完全掌控請求解析與響應生成方式 1可組合build_run_input()把請求體字節解析為RunAgentInput校驗失敗返回422、run_stream()與encode_stream()按 Accept 頭編碼為 SSE 字符串完整示例見 docs/ui/ag-ui.md。取消語義當一次運行以第一方取消結束ctx.cancel()、AgentRun.cancel()或取消端點觸發的CancellationToken時適配器會關閉未完成的文本/工具事件并發出一個不帶 outcome 的RUN_FINISHED——AG-UI 目前沒有 cancelled 結局因此取消不會被報告為RUN_ERROR。可以傳入on_cancel回調用RunCancelled.all_messages()持久化可恢復的消息歷史。需要注意客戶端斷開連接屬于外部取消服務端看到的是asyncio.CancelledError不會觸發上述RUN_FINISHED與on_cancel。要捕獲停止手勢應保持流連接并通過單獨的取消端點觸發CancellationToken進行第一方取消詳見 docs/agent.md。信任模型與安全邊界AG-UI 的RunAgentInput.messages完全由客戶端控制。AGUIAdapter會應用默認策略剝離不可信部分系統提示、文件 URL 協議、上傳文件、未決工具調用等allow_uploaded_files控制上傳文件門禁但這些默認并不等于客戶端歷史可信詳見 docs/ui/overview.md 與 docs/message-history.md 中的信任邊界討論。此外AG-UI 客戶端可發送context數組description/value對描述其認為與本次運行相關的信息來源平臺、請求用戶、頻道常駐指令等。這些條目不會被自動傳入模型也不應被拼進instructions——指令帶有操作者權威把客戶端文本拼進去會讓提示注入繼承這種權威正確做法是把它們作為數據交付給模型例如通過一個frontend_context工具暴露給 Agent 讀取見 docs/ui/ag-ui.md 與 docs/ui/overview.md。系統提示詞與指令的歸屬Pydantic AI 區分兩種引導方式system_prompt持久化在消息歷史中作為SystemPromptPart與instructions每次請求新鮮注入、從不持久化。服務端可控時推薦默認使用instructions——無論 AG-UI 消息歷史如何它總是生效。若確實使用system_prompt可通過AGUIAdapter的manage_system_prompt參數選擇歸屬server默認Agent 配置的system_prompt具有權威性前端發來的SystemMessage會被剝離并告警同時通過ReinjectSystemPrompt能力在首次請求頭部重新注入。client前端擁有系統提示詞前端SystemMessage原樣保留Agent 配置的system_prompt不再注入若想回退到配置內容可為 Agent 加上ReinjectSystemPrompt能力。示例見 docs/ui/ag-ui.md。協議版本兼容與失敗工具結果保留Pydantic AI 支持ag-ui-protocol從0.1.10起的所有版本新特性按已安裝版本雙向門控向外舊協議無法表達的內容會被降級或省略見AGUIAdapter.ag_ui_version的協商閾值向內當前安裝的ag-ui-protocol沒有對應類的消息role或內容type會被跳過并發出UserWarning例如網關轉發的多模態圖片內容其余請求繼續運行。跳過僅針對結構合法的條目消息必須仍帶字符串id格式錯誤、role/type非字符串、非法 JSON 等仍會以422拒絕詳見 docs/ui/ag-ui.md。關于失敗工具結果AG-UI 的ToolCallResultEvent沒有 error/outcome 字段Pydantic AI 在ag-ui-protocol 0.1.11下使用ReasoningEncryptedValueEvent的encrypted_value附件機制攜帶命名空間化的 payload 來保留outcomefailed客戶端回傳這些消息時適配器會恢復失敗結局。這是歷史連續性機制不會設置ToolMessage.error也不保證前端把結果渲染為錯誤詳見 docs/ui/ag-ui.md。結語通過 docs/examples/ag-ui.md 與倉庫示例可以看到 Pydantic AI 對 AG-UI 的集成覆蓋了協議的全部核心能力事件、消息、狀態管理與工具。從最簡單的dispatch_request單路由接入到StateDeps驅動的共享狀態、工具返回的StateSnapshotEvent/StateDeltaEvent、requires_approval觸發的 interrupt 審批流再到manage_system_prompt與preserve_file_data等細粒度控制AGUIAdapter把協議細節封裝得足夠薄讓開發者可以專注于 Agent 本身的業務邏輯。想要深入了解各 Feature 的完整實現可以直接閱讀倉庫內的 examples/pydantic_ai_examples/ag_ui/api/ 目錄agentic_chat.py、agentic_generative_ui.py、human_in_the_loop.py、predictive_state_updates.py、shared_state.py、tool_based_generative_ui.py、tool_approval.py或參考 AG-UI 集成文檔 中更系統的 API 說明若要讓同一個 Agent 同時服務 Slack 等消息平臺docs/ui/ag-ui.md 中的 CopilotKit Channels 一節提供了從 Slack 到 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創作聲明：本文部分內容由AI輔助生成（AIGC），僅供參考 Pydantic AI 與 AG-UI 協議：用 Agentic UI 構建人機協同的前端應用
Pydantic AI 與 AG-UI 協議用 Agentic UI 構建人機協同的前端應用【免費下載鏈接】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-aiAG-UIAgent-User Interaction是 CopilotKit 團隊提出的開放協議用于標準化前端應用與 AI Agent 之間的通信方式。本文將以 Pydantic AI 倉庫中的 docs/examples/ag-ui.md 為骨架結合 AG-UI 集成文檔 與倉庫內的完整示例源碼完整講解如何把 Pydantic AI Agent 接入 AG-UI 生態、如何在本地通過 AG-UI Dojo 調試面板逐項驗證 Agentic Chat、Human in the Loop、共享狀態、預測式狀態更新等六大交互范式以及AGUIAdapter在底層如何完成協議轉換。讀完本文你將掌握AG-UI 后端的三種接入方式run_stream、dispatch_request、獨立 Starlette 應用、基于StateDeps的前后端狀態共享、基于工具事件與CustomEvent的流式進度推送以及基于DeferredToolRequests的工具審批interrupt流程。背景為什么需要 AG-UI以及 Pydantic AI 如何接入傳統 AI 應用的界面通常由服務端生成整段對話或整頁內容前端只能被動展示。AG-UI 協議則把通信拆分為**事件Events、消息Messages、狀態管理State、工具Tools**四大概念讓前端可以持有工具、維護共享狀態、接收流式事件從而構建生成式 UIGenerative UI體驗。Pydantic AI 通過AGUIAdapter位于pydantic_ai.ui.ag_ui實現協議適配前端把請求封裝為 AG-UI 的RunAgentInput對象包含消息歷史、狀態、可用工具適配器將其轉換為 Pydantic AI 內部類型交給 Agent 執行Agent 產生的工具調用、狀態更新等事件再被轉換回 AG-UI 事件以Server-Sent EventsSSE流式返回給前端。一次用戶請求可能需要客戶端 UI 與 Pydantic AI 服務端之間的多輪往返取決于工具和事件的需要見 docs/ui/ag-ui.md。該集成最初由 Rocket Science 團隊構建并與 Pydantic AI、CopilotKit 團隊合作貢獻見 AG-UI 集成文檔 中的說明。快速啟動在本地跑通 AG-UI 示例倉庫在 examples/pydantic_ai_examples/ag_ui/main.py 提供了一個基于 FastAPI 的 AG-UI 后端并在 examples/pydantic_ai_examples/ag_ui/init.py 中把每個 Feature 掛載為獨立子應用app FastAPI(titlePydantic AI AG-UI server) app.mount(/agentic_chat, agentic_chat_app, Agentic Chat) app.mount(/agentic_generative_ui, agentic_generative_ui_app, Agentic Generative UI) app.mount(/human_in_the_loop, human_in_the_loop_app, Human in the Loop) app.mount(/predictive_state_updates, predictive_state_updates_app, Predictive State Updates) app.mount(/shared_state, shared_state_app, Shared State) app.mount(/tool_approval, tool_approval_app, Tool Approval (interrupts)) app.mount(/tool_based_generative_ui, tool_based_generative_ui_app, Tool Based Generative UI)前置條件一個 OpenAI API Key已安裝項目依賴并設置好環境變量參見 docs/examples/setup.md需要兩個命令行窗口分別運行前后端。第一步啟動 Pydantic AI AG-UI 后端設置 API Key 并啟動示例后端export OPENAI_API_KEYyour api key python/uv-run -m pydantic_ai_examples.ag_ui__main__.py內部通過 uvicorn 在9000端口啟動服務if __name__ __main__: import uvicorn uvicorn.run(pydantic_ai_examples.ag_ui:app, port9000)第二步運行 AG-UI Dojo 前端AG-UI Dojo 是 AG-UI 官方的調試面板可以逐項演示協議特性克隆 AG-UI 倉庫git clone https://github.com/ag-ui-protocol/ag-ui.git按官方說明安裝前置依賴然后從倉庫根目錄安裝依賴并構建cd ag-ui pnpm i pnpm build --projectsdemo-viewer進入apps/dojo目錄運行 Dojo 應用cd apps/dojo pnpm dev瀏覽器訪問 http://localhost:3000/pydantic-ai在側邊欄選擇Pydantic AI視圖每個 Feature 的訪問地址為http://localhost:3000/pydantic-ai/feature/feature_name下文逐一說明。六大交互范式詳解基于倉庫示例源碼Agentic Chat服務端工具與客戶端工具同場協作這是最基本的 Agent 交互范式演示 Pydantic AI 服務端工具與 AG-UI 客戶端工具如何協同工作。訪問地址http://localhost:3000/pydantic-ai/feature/agentic_chat。該示例包含兩個工具time——Pydantic AI 服務端工具查詢指定時區的當前時間background——AG-UI 客戶端工具修改客戶端窗口的背景色對應的示例源碼為 examples/pydantic_ai_examples/ag_ui/api/agentic_chat.py。服務端工具用agent.tool_plain聲明內部通過zoneinfo.ZoneInfo處理時區并返回 ISO 格式時間agent Agent(openai:gpt-5-mini) agent.tool_plain async def current_time(timezone: str UTC) - str: Get the current time in ISO format. tz: ZoneInfo ZoneInfo(timezone) return datetime.now(tztz).isoformat() async def run_agent(request: Request) - Response: return await AGUIAdapter.dispatch_request(request, agentagent) app Starlette(routes[Route(/, run_agent, methods[POST])])端點本身非常簡潔單個POST /路由調用AGUIAdapter.dispatch_request(request, agentagent)其余協議細節全部由適配器接管。background這類客戶端工具不會出現在服務端代碼里而是由 AG-UI 前端在請求中聲明適配器會把客戶端工具透傳給模型由模型決定何時調用。可以嘗試的提示詞What is the time in New York?Change the background to blue更復雜的混合示例——讓模型在兩個工具間交替執行并計算耗時Perform the following steps, waiting for the response of each step before continuing: 1. Get the time 2. Set the background to red 3. Get the time 4. Report how long the background set took by diffing the two times這個示例直觀展示了 AG-UI 的客戶端工具能力工具執行結果由前端渲染服務端只負責推理決策真正實現了 UI 能力的分布。Agentic Generative UI長任務中的流式狀態更新該示例演示一個長時間運行的任務Agent 邊執行邊把進度推送給前端讓用戶實時看到正在發生什么。訪問地址http://localhost:3000/pydantic-ai/feature/agentic_generative_ui。示例源碼為 examples/pydantic_ai_examples/ag_ui/api/agentic_generative_ui.py。它用 Pydantic 模型描述計劃結構Step單個步驟含description和statuspending/completedPlan步驟列表JSONPatchOpRFC 6902 JSON Patch 操作用于表達狀態增量Agent 的指令強調只使用工具、不輸出多余文字agent Agent( openai:gpt-5-mini, instructionsdedent( When planning use tools only, without any other messages. IMPORTANT: - Use the create_plan tool to set the initial state of the steps - Use the update_plan_step tool to update the status of each step - Do NOT repeat the plan or summarise it in a message ... Only one plan can be active at a time, so do not call the create_plan tool again until all the steps in current plan are completed. ), )兩個核心工具直接返回 AG-UI 事件對象create_plan返回StateSnapshotEvent狀態快照把整個計劃一次性同步給前端update_plan_step返回StateDeltaEvent狀態增量攜帶 JSON Patch 操作數組只推送變更部分agent.tool_plain async def create_plan(steps: list[str]) - StateSnapshotEvent: plan Plan(steps[Step(descriptionstep) for step in steps]) return StateSnapshotEvent(typeEventType.STATE_SNAPSHOT, snapshotplan.model_dump()) agent.tool_plain async def update_plan_step(index: int, description: str | None None, status: StepStatus | None None) - StateDeltaEvent: changes: list[JSONPatchOp] [] if description is not None: changes.append(JSONPatchOp(opreplace, pathf/steps/{index}/description, valuedescription)) if status is not None: changes.append(JSONPatchOp(opreplace, pathf/steps/{index}/status, valuestatus)) return StateDeltaEvent(typeEventType.STATE_DELTA, deltachanges)實現要點Pydantic AI 工具可以直接返回 AG-UI 的BaseEvent或事件迭代器適配器會把這些事件作為工具結果的一部分隨事件流發給前端。這與ctx.emit()的即時事件不同——工具返回事件屬于消息的一部分能隨消息歷史往返適合前端需要重建的狀態更新詳見 docs/ui/ag-ui.md。嘗試提示詞Create a plan for breakfast and execute it前端會看到一個逐步勾選pending → completed的計劃卡片而不是一段純文本回復。Human in the Loop讓用戶審批 Agent 提出的計劃該示例演示簡單的人機協同流程Agent 生成計劃用戶在界面上用復選框確認。訪問地址http://localhost:3000/pydantic-ai/feature/human_in_the_loop。示例源碼為 examples/pydantic_ai_examples/ag_ui/api/human_in_the_loop.py。這個 Feature 依賴的是 AG-UI 的客戶端工具generate_task_steps——它由前端實現用于展示并確認步驟。服務端只需在指令中約束行為agent Agent( openai:gpt-5-mini, instructionsdedent( When planning tasks use tools only, without any other messages. IMPORTANT: - Use the generate_task_steps tool to display the suggested steps to the user - Never repeat the plan, or send a message detailing steps - If accepted, confirm the creation of the plan and the number of selected (enabled) steps only - If not accepted, ask the user for more information, DO NOT use the generate_task_steps tool again ), )嘗試提示詞Generate a list of steps for cleaning a car for me to review值得留意的是該文件 docstring 中的一句話No special handling is required for this feature.——人機協同的核心邏輯完全由 AG-UI 協議客戶端工具承擔服務端只需告訴模型什么時候該用工具、什么時候不該用。Predictive State Updates預測式狀態更新該示例演示如何基于 Agent 的響應預測性更新 UI 狀態包括通過用戶確認進行交互。訪問地址http://localhost:3000/pydantic-ai/feature/predictive_state_updates。示例源碼為 examples/pydantic_ai_examples/ag_ui/api/predictive_state_updates.py。它定義了一個DocumentState含document字段作為前后端共享狀態通過StateDeps注入class DocumentState(BaseModel): State for the document being written. document: str agent Agent(openai:gpt-5-mini, deps_typeStateDeps[DocumentState])關鍵工具document_predict_state返回一個名為PredictState的CustomEvent聲明write_document工具的document參數會更新document狀態鍵——前端據此在工具執行前就預測性地渲染新文檔agent.tool_plain async def document_predict_state() - list[CustomEvent]: Enable document state prediction. return [ CustomEvent( typeEventType.CUSTOM, namePredictState, value[ { state_key: document, tool: write_document, tool_argument: document, }, ], ), ]示例還展示了基于共享狀態的自定義指令agent.instructions()裝飾器把當前文檔內容動態注入指令讓模型接著寫而不是重寫agent.instructions() async def story_instructions(ctx: RunContext[StateDeps[DocumentState]]) - str: return dedent(f... Before you start writing, you MUST call the document_predict_state tool to enable state prediction. To present the document to the user for review, you MUST use the write_document tool. ... This is the current document: {ctx.deps.state.document} )啟動文檔內容為Bruce was a good dog,嘗試提示詞Help me complete my story about bruce the dog, is should be no longer than a sentence.注意請求處理時的一個關鍵細節dispatch_request會就地修改deps.state因此每個請求都要用dataclasses.replace生成獨立副本避免請求間狀態串擾deps StateDeps(DocumentState()) async def run_agent(request: Request) - Response: # dispatch_request mutates deps.state from the request, so give each request its own copy. return await AGUIAdapter.dispatch_request(request, agentagent, depsreplace(deps))Shared State前后端共享狀態該示例演示 UI 與 Agent 之間的狀態共享發送給 Agent 的狀態被一個基于函數的指令檢測到先用自定義 Pydantic 模型校驗數據再據此生成指令讓 Agent 遵循最后通過 AG-UI 工具把結果發回客戶端。訪問地址http://localhost:3000/pydantic-ai/feature/shared_state。示例源碼為 examples/pydantic_ai_examples/ag_ui/api/shared_state.py。它用枚舉定義SkillLevel、SpecialPreferences、CookingTime用Recipe/RecipeSnapshot兩個 Pydantic 模型承載配方結構class RecipeSnapshot(BaseModel): recipe: Recipe Field(default_factoryRecipe, descriptionThe current state of the recipe) agent Agent(openai:gpt-5-mini, deps_typeStateDeps[RecipeSnapshot])展示工具display_recipe返回StateSnapshotEvent把整個配方快照同步給前端以圖形化渲染agent.tool_plain async def display_recipe(recipe: Recipe) - StateSnapshotEvent: Display the recipe to the user. return StateSnapshotEvent( typeEventType.STATE_SNAPSHOT, snapshot{recipe: recipe}, )recipe_instructions同樣基于當前狀態動態生成指令把已有配方以 JSON 形式注入上下文agent.instructions async def recipe_instructions(ctx: RunContext[StateDeps[RecipeSnapshot]]) - str: return dedent(f... - Create a complete recipe using the existing ingredients - Append new ingredients to the existing ones - Use the display_recipe tool to present the recipe to the user - Do NOT repeat the recipe in the message, use the tool instead ... The current state of the recipe is: {ctx.deps.state.recipe.model_dump_json(indent2)} )操作步驟1. 自定義配方的初始設置技能等級、偏好、烹飪時長、食材2. 點擊Improve with AI觀察 Agent 在既有狀態上增量優化配方并通過display_recipe展示。Tool Based Generative UI工具輸出的定制渲染該示例演示帶用戶確認的工具輸出定制渲染。訪問地址http://localhost:3000/pydantic-ai/feature/tool_based_generative_ui。示例源碼為 examples/pydantic_ai_examples/ag_ui/api/tool_based_generative_ui.py。與服務端示例不同這里的generate_haiku是一個 AG-UI 客戶端工具負責以英文和日文雙語卡片形式渲染俳句——定制渲染邏輯完全發生在前端。嘗試提示詞Generate a haiku about formula 1延伸Tool Approval工具審批 / Interrupts除了 Dojo 六大 Feature 之外倉庫還提供了 examples/pydantic_ai_examples/ag_ui/api/tool_approval.py 演示 AG-UI 的 interrupt 生命周期該能力在 docs/ui/ag-ui.md 中有完整說明需要ag-ui-protocol 0.1.19。核心思路用agent.tool_plain(requires_approvalTrue)聲明危險工具并把DeferredToolRequests加入output_type這樣當模型提議調用該工具時運行會暫停而不是報錯agent Agent(openai:gpt-5-mini, output_type[str, DeferredToolRequests]) agent.tool_plain(requires_approvalTrue) def delete_file(path: str) - str: Delete a file. The run pauses here and waits for the user to approve before executing. return fdeleted {path}流程如下模型提議調用 → 適配器以outcome.type interrupt的RUN_FINISHED事件結束 SSE 流outcome.interrupts[]描述每個待審批項 → 前端據此渲染審批 UI → 用戶操作后前端 POST 攜帶resume[]數組ResumeEntry的下一個RunAgentInput。適配器的字段映射與 AG-UI Python SDK 字段名一致總結如下見 docs/ui/ag-ui.mdAG-UI 方向Pydantic AI 來源 / 去向Interrupt.reason對requires_approvalTrue工具恒為tool_callInterrupt.tool_call_id提議調用的ToolCallPart.tool_call_idInterrupt.idfint-{tool_call_id}resume 時還原為 tool_call_idInterrupt.metadataDeferredToolRequests.metadata.get(tool_call_id)payload.approvedTrueToolApprovedpayload.editedArgsToolApproved.override_args整體替換提議參數payload.approvedFalseToolDeniedmessagepayload.reasonstatuscancelledToolDeniedmessageCancelled by user.payload還會依據Interrupt.response_schema校驗approved字段必填editedArgs、reason若給出但類型錯誤即使approvedTrue也會被判定為拒絕。恢復輪次中 Agent 會以原始tool_call_id重新執行工具因此只會發出該 id 的TOOL_CALL_RESULT事件而不會重復TOOL_CALL_START從而保留 AG-UI 規范要求的審計軌跡。底層原語DeferredToolRequests不依賴 AG-UI 也能獨立使用詳見 docs/deferred-tools.md。底層原理AGUIAdapter 的三種接入方式與事件流轉從 docs/ui/ag-ui.md 可知運行基于 AG-UI 輸入的 Agent 有三種方式靈活度從高到低AGUIAdapter.run_stream()對以RunAgentInput實例化的適配器調用運行 Agent 并返回 AG-UI 事件流支持Agent.iter()的可選參數如deps。適合非 Starlette 框架Django、Flask或需要自行加工輸入/輸出的場景。AGUIAdapter.dispatch_request()類方法接收 Starlette 請求如來自 FastAPI直接返回流式 Starlette 響應可逐請求傳入deps如基于已認證用戶。它是from_request()、run_stream()、streaming_response()三者的便捷組合。獨立 Starlette 應用單個/路由調用dispatch_request()同一應用還能以子應用方式掛載到既有 FastAPI見 FastAPI 子應用文檔。最小可用實現方式 3只需十幾行from starlette.applications import Starlette from starlette.requests import Request from starlette.responses import Response from starlette.routing import Route from pydantic_ai import Agent from pydantic_ai.ui.ag_ui import AGUIAdapter agent Agent(openai:gpt-5.2, instructionsBe fun!) async def run_agent(request: Request) - Response: return await AGUIAdapter.dispatch_request(request, agentagent) app Starlette(routes[Route(/, run_agent, methods[POST])])啟動uvicorn ag_ui_app:app若需完全掌控請求解析與響應生成方式 1可組合build_run_input()把請求體字節解析為RunAgentInput校驗失敗返回422、run_stream()與encode_stream()按 Accept 頭編碼為 SSE 字符串完整示例見 docs/ui/ag-ui.md。取消語義當一次運行以第一方取消結束ctx.cancel()、AgentRun.cancel()或取消端點觸發的CancellationToken時適配器會關閉未完成的文本/工具事件并發出一個不帶 outcome 的RUN_FINISHED——AG-UI 目前沒有 cancelled 結局因此取消不會被報告為RUN_ERROR。可以傳入on_cancel回調用RunCancelled.all_messages()持久化可恢復的消息歷史。需要注意客戶端斷開連接屬于外部取消服務端看到的是asyncio.CancelledError不會觸發上述RUN_FINISHED與on_cancel。要捕獲停止手勢應保持流連接并通過單獨的取消端點觸發CancellationToken進行第一方取消詳見 docs/agent.md。信任模型與安全邊界AG-UI 的RunAgentInput.messages完全由客戶端控制。AGUIAdapter會應用默認策略剝離不可信部分系統提示、文件 URL 協議、上傳文件、未決工具調用等allow_uploaded_files控制上傳文件門禁但這些默認并不等于客戶端歷史可信詳見 docs/ui/overview.md 與 docs/message-history.md 中的信任邊界討論。此外AG-UI 客戶端可發送context數組description/value對描述其認為與本次運行相關的信息來源平臺、請求用戶、頻道常駐指令等。這些條目不會被自動傳入模型也不應被拼進instructions——指令帶有操作者權威把客戶端文本拼進去會讓提示注入繼承這種權威正確做法是把它們作為數據交付給模型例如通過一個frontend_context工具暴露給 Agent 讀取見 docs/ui/ag-ui.md 與 docs/ui/overview.md。系統提示詞與指令的歸屬Pydantic AI 區分兩種引導方式system_prompt持久化在消息歷史中作為SystemPromptPart與instructions每次請求新鮮注入、從不持久化。服務端可控時推薦默認使用instructions——無論 AG-UI 消息歷史如何它總是生效。若確實使用system_prompt可通過AGUIAdapter的manage_system_prompt參數選擇歸屬server默認Agent 配置的system_prompt具有權威性前端發來的SystemMessage會被剝離并告警同時通過ReinjectSystemPrompt能力在首次請求頭部重新注入。client前端擁有系統提示詞前端SystemMessage原樣保留Agent 配置的system_prompt不再注入若想回退到配置內容可為 Agent 加上ReinjectSystemPrompt能力。示例見 docs/ui/ag-ui.md。協議版本兼容與失敗工具結果保留Pydantic AI 支持ag-ui-protocol從0.1.10起的所有版本新特性按已安裝版本雙向門控向外舊協議無法表達的內容會被降級或省略見AGUIAdapter.ag_ui_version的協商閾值向內當前安裝的ag-ui-protocol沒有對應類的消息role或內容type會被跳過并發出UserWarning例如網關轉發的多模態圖片內容其余請求繼續運行。跳過僅針對結構合法的條目消息必須仍帶字符串id格式錯誤、role/type非字符串、非法 JSON 等仍會以422拒絕詳見 docs/ui/ag-ui.md。關于失敗工具結果AG-UI 的ToolCallResultEvent沒有 error/outcome 字段Pydantic AI 在ag-ui-protocol 0.1.11下使用ReasoningEncryptedValueEvent的encrypted_value附件機制攜帶命名空間化的 payload 來保留outcomefailed客戶端回傳這些消息時適配器會恢復失敗結局。這是歷史連續性機制不會設置ToolMessage.error也不保證前端把結果渲染為錯誤詳見 docs/ui/ag-ui.md。結語通過 docs/examples/ag-ui.md 與倉庫示例可以看到 Pydantic AI 對 AG-UI 的集成覆蓋了協議的全部核心能力事件、消息、狀態管理與工具。從最簡單的dispatch_request單路由接入到StateDeps驅動的共享狀態、工具返回的StateSnapshotEvent/StateDeltaEvent、requires_approval觸發的 interrupt 審批流再到manage_system_prompt與preserve_file_data等細粒度控制AGUIAdapter把協議細節封裝得足夠薄讓開發者可以專注于 Agent 本身的業務邏輯。想要深入了解各 Feature 的完整實現可以直接閱讀倉庫內的 examples/pydantic_ai_examples/ag_ui/api/ 目錄agentic_chat.py、agentic_generative_ui.py、human_in_the_loop.py、predictive_state_updates.py、shared_state.py、tool_based_generative_ui.py、tool_approval.py或參考 AG-UI 集成文檔 中更系統的 API 說明若要讓同一個 Agent 同時服務 Slack 等消息平臺docs/ui/ag-ui.md 中的 CopilotKit Channels 一節提供了從 Slack 到 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創作聲明：本文部分內容由AI輔助生成（AIGC），僅供參考