企业官网建设流程全解析

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-ai

AG-UI（Agent-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）**四大概念，让前端可以持有工具、维护共享状态、接收流式事件，从而构建"生成式 UI（Generative UI）"体验。

Pydantic AI 通过AGUIAdapter（位于pydantic_ai.ui.ag_ui）实现协议适配：前端把请求封装为 AG-UI 的RunAgentInput对象（包含消息历史、状态、可用工具），适配器将其转换为 Pydantic AI 内部类型交给 Agent 执行；Agent 产生的工具调用、状态更新等事件再被转换回 AG-UI 事件，以Server-Sent Events（SSE）流式返回给前端。一次用户请求可能需要客户端 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(title='Pydantic 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_KEY=<your 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', port=9000)

第二步：运行 AG-UI Dojo 前端

AG-UI Dojo 是 AG-UI 官方的调试面板，可以逐项演示协议特性：

1. 克隆 AG-UI 仓库：

   git clone https://github.com/ag-ui-protocol/ag-ui.git

2. 按官方说明安装前置依赖，然后从仓库根目录安装依赖并构建：

   cd ag-ui pnpm i pnpm build --projects=demo-viewer

3. 进入apps/dojo目录运行 Dojo 应用：

   cd apps/dojo pnpm dev

4. 浏览器访问 http://localhost:3000/pydantic-ai
5. 在侧边栏选择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(tz=tz).isoformat() async def run_agent(request: Request) -> Response: return await AGUIAdapter.dispatch_request(request, agent=agent) app = Starlette(routes=[Route('/', run_agent, methods=['POST'])])

端点本身非常简洁：单个POST /路由调用AGUIAdapter.dispatch_request(request, agent=agent)，其余协议细节全部由适配器接管。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和status（pending/completed）
• Plan：步骤列表
• JSONPatchOp：RFC 6902 JSON Patch 操作，用于表达状态增量

Agent 的指令强调"只使用工具、不输出多余文字"：

agent = Agent( 'openai:gpt-5-mini', instructions=dedent(""" 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(description=step) for step in steps]) return StateSnapshotEvent(type=EventType.STATE_SNAPSHOT, snapshot=plan.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(op='replace', path=f'/steps/{index}/description', value=description)) if status is not None: changes.append(JSONPatchOp(op='replace', path=f'/steps/{index}/status', value=status)) return StateDeltaEvent(type=EventType.STATE_DELTA, delta=changes)

实现要点：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', instructions=dedent(""" 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_type=StateDeps[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( type=EventType.CUSTOM, name='PredictState', 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, agent=agent, deps=replace(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_factory=Recipe, description='The current state of the recipe') agent = Agent('openai:gpt-5-mini', deps_type=StateDeps[RecipeSnapshot])

展示工具display_recipe返回StateSnapshotEvent，把整个配方快照同步给前端以图形化渲染：

@agent.tool_plain async def display_recipe(recipe: Recipe) -> StateSnapshotEvent: """Display the recipe to the user.""" return StateSnapshotEvent( type=EventType.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(indent=2)} """)

操作步骤：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_approval=True)声明危险工具，并把DeferredToolRequests加入output_type，这样当模型提议调用该工具时，运行会暂停而不是报错：

agent = Agent('openai:gpt-5-mini', output_type=[str, DeferredToolRequests]) @agent.tool_plain(requires_approval=True) def delete_file(path: str) -> str: """Delete a file. The run pauses here and waits for the user to approve before executing.""" return f'deleted {path}'

流程如下：模型提议调用 → 适配器以outcome.type == "interrupt"的RUN_FINISHED事件结束 SSE 流，outcome.interrupts[]描述每个待审批项 → 前端据此渲染审批 UI → 用户操作后前端 POST 携带resume[]数组（ResumeEntry）的下一个RunAgentInput。

适配器的字段映射（与 AG-UI Python SDK 字段名一致）总结如下（见 docs/ui/ag-ui.md）：

payload还会依据Interrupt.response_schema校验：approved字段必填；editedArgs、reason若给出但类型错误，即使approved=True也会被判定为拒绝。恢复轮次中 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 有三种方式，灵活度从高到低：

1. AGUIAdapter.run_stream()：对以RunAgentInput实例化的适配器调用，运行 Agent 并返回 AG-UI 事件流；支持Agent.iter()的可选参数（如deps）。适合非 Starlette 框架（Django、Flask）或需要自行加工输入/输出的场景。
2. AGUIAdapter.dispatch_request()类方法：接收 Starlette 请求（如来自 FastAPI）直接返回流式 Starlette 响应，可逐请求传入deps（如基于已认证用户）。它是from_request()、run_stream()、streaming_response()三者的便捷组合。
3. 独立 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', instructions='Be fun!') async def run_agent(request: Request) -> Response: return await AGUIAdapter.dispatch_request(request, agent=agent) 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 来保留outcome='failed'；客户端回传这些消息时适配器会恢复失败结局。这是历史连续性机制，不会设置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