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），仅供参考