代理-用户交互 (AG-UI) 协议

代理-用户交互 (AG-UI) 协议是由 CopilotKit 团队引入的一项开放标准，它标准化了前端应用程序与 AI 代理的通信方式，并支持流式传输、前端工具、共享状态和自定义事件。

注意

AG-UI 集成最初由 Rocket Science 团队构建，并与 Pydantic AI 和 CopilotKit 团队合作贡献。感谢 Rocket Science！

安装

唯一的依赖项是

ag-ui-protocol：提供 AG-UI 类型和编码器。
starlette：用于处理来自 FastAPI 等框架的 ASGI 请求。

你可以通过安装 Pydantic AI 的 ag-ui 附加项来确保你拥有所有必需的 AG-UI 依赖

pipuv

pip install 'pydantic-ai-slim[ag-ui]'

uv add 'pydantic-ai-slim[ag-ui]'

要运行这些示例，你还需要

uvicorn 或其他 ASGI 兼容的服务器

pipuv

pip install uvicorn

uv add uvicorn

用法

有三种方法可以基于 AG-UI 运行输入来运行 Pydantic AI 代理，并以流式 AG-UI 事件作为输出，按灵活性从高到低排列。如果你正在使用像 FastAPI 这样基于 Starlette 的 Web 框架，通常会选择第二种方法。

run_ag_ui() 接收一个代理和一个 AG-UI 的 RunAgentInput 对象，并返回一个编码为字符串的 AG-UI 事件流。它还接受可选的 Agent.iter() 参数，包括 deps。如果你正在使用非基于 Starlette 的 Web 框架（例如 Django 或 Flask），或者想要以某种方式修改输入或输出，请使用此方法。
handle_ag_ui_request() 接收一个代理和一个来自 AG-UI 前端的 Starlette 请求（例如来自 FastAPI），并返回一个流式的 AG-UI 事件 Starlette 响应，你可以直接从你的端点返回它。它也接受可选的 Agent.iter() 参数，包括 deps，你可以为每个请求改变这些参数（例如，基于已认证的用户）。
Agent.to_ag_ui() 返回一个 ASGI 应用程序，它通过运行代理来处理每个 AG-UI 请求。它也接受可选的 Agent.iter() 参数，包括 deps，但除了在状态管理下描述的被注入的 AG-UI 状态外，这些参数对于每个请求都是相同的。这个 ASGI 应用可以被挂载到一个现有 FastAPI 应用的给定路径上。

直接处理运行输入和输出

此示例使用 run_ag_ui() 并执行自己的请求解析和响应生成。这可以被修改以适用于任何 Web 框架。

run_ag_ui.py

import json
from http import HTTPStatus

from ag_ui.core import RunAgentInput
from fastapi import FastAPI
from fastapi.requests import Request
from fastapi.responses import Response, StreamingResponse
from pydantic import ValidationError

from pydantic_ai import Agent
from pydantic_ai.ag_ui import SSE_CONTENT_TYPE, run_ag_ui

agent = Agent('openai:gpt-4.1', instructions='Be fun!')

app = FastAPI()


@app.post('/')
async def run_agent(request: Request) -> Response:
    accept = request.headers.get('accept', SSE_CONTENT_TYPE)
    try:
        run_input = RunAgentInput.model_validate(await request.json())
    except ValidationError as e:  # pragma: no cover
        return Response(
            content=json.dumps(e.json()),
            media_type='application/json',
            status_code=HTTPStatus.UNPROCESSABLE_ENTITY,
        )

    event_stream = run_ag_ui(agent, run_input, accept=accept)

    return StreamingResponse(event_stream, media_type=accept)

由于 app 是一个 ASGI 应用程序，它可以与任何 ASGI 服务器一起使用

uvicorn run_ag_ui:app

这将把代理暴露为一个 AG-UI 服务器，你的前端就可以开始向它发送请求了。

处理 Starlette 请求

此示例使用 handle_ag_ui_request() 直接处理一个 FastAPI 请求并返回一个响应。与此类似的方法将适用于任何基于 Starlette 的 Web 框架。

handle_ag_ui_request.py

from fastapi import FastAPI
from starlette.requests import Request
from starlette.responses import Response

from pydantic_ai import Agent
from pydantic_ai.ag_ui import handle_ag_ui_request

agent = Agent('openai:gpt-4.1', instructions='Be fun!')

app = FastAPI()

@app.post('/')
async def run_agent(request: Request) -> Response:
    return await handle_ag_ui_request(agent, request)

由于 app 是一个 ASGI 应用程序，它可以与任何 ASGI 服务器一起使用

uvicorn handle_ag_ui_request:app

这将把代理暴露为一个 AG-UI 服务器，你的前端就可以开始向它发送请求了。

独立的 ASGI 应用

此示例使用 Agent.to_ag_ui() 将代理转变为一个独立的 ASGI 应用程序

agent_to_ag_ui.py

from pydantic_ai import Agent

agent = Agent('openai:gpt-4.1', instructions='Be fun!')
app = agent.to_ag_ui()

由于 app 是一个 ASGI 应用程序，它可以与任何 ASGI 服务器一起使用

uvicorn agent_to_ag_ui:app

这将把代理暴露为一个 AG-UI 服务器，你的前端就可以开始向它发送请求了。

设计

Pydantic AI AG-UI 集成支持该规范的所有功能

该集成以 RunAgentInput 对象的形式接收消息，该对象描述了请求的代理运行的详细信息，包括消息历史、状态和可用工具。

这些消息被转换为 Pydantic AI 类型并传递给代理的运行方法。来自代理的事件，包括工具调用，被转换为 AG-UI 事件，并以服务器发送事件 (SSE) 的形式流式传输回调用者。

一个用户请求可能需要在客户端 UI 和 Pydantic AI 服务器之间进行多次往返，这取决于所需的工具和事件。

特性

状态管理

该集成完全支持 AG-UI 状态管理，它实现了代理和前端应用程序之间的实时同步。

在下面的示例中，我们有一个文档状态，它使用 StateDeps 依赖类型在 UI 和服务器之间共享，该类型可用于自动验证包含在 RunAgentInput.state 中的状态，验证时使用一个指定为泛型参数的 Pydantic BaseModel。

带有 AG-UI 状态的自定义依赖类型

如果你想使用自己的依赖类型来保存 AG-UI 状态以及其他东西，它需要实现 StateHandler 协议，这意味着它需要是一个带有非可选 state 字段的 dataclass。这使得 Pydantic AI 能够通过每次都构建一个新的依赖对象来确保状态在请求之间得到适当的隔离。

如果 state 字段的类型是 Pydantic BaseModel 的子类，请求中的原始状态字典会被自动验证。如果不是，你可以在你的依赖数据类（dataclass）的 __post_init__ 方法中自己验证原始值。

ag_ui_state.py

from pydantic import BaseModel

from pydantic_ai import Agent
from pydantic_ai.ag_ui import StateDeps


class DocumentState(BaseModel):
    """State for the document being written."""

    document: str = ''


agent = Agent(
    'openai:gpt-4.1',
    instructions='Be fun!',
    deps_type=StateDeps[DocumentState],
)
app = agent.to_ag_ui(deps=StateDeps(DocumentState()))

由于 app 是一个 ASGI 应用程序，它可以与任何 ASGI 服务器一起使用

uvicorn ag_ui_state:app --host 0.0.0.0 --port 9000

工具

AG-UI 前端工具被无缝地提供给 Pydantic AI 代理，从而实现了具有前端用户界面的丰富用户体验。

事件

Pydantic AI 工具可以发送 AG-UI 事件，只需定义一个返回 BaseEvent（或其子类）的工具即可，这允许自定义事件和状态更新。

ag_ui_tool_events.py

from ag_ui.core import CustomEvent, EventType, StateSnapshotEvent
from pydantic import BaseModel

from pydantic_ai import Agent, RunContext
from pydantic_ai.ag_ui import StateDeps


class DocumentState(BaseModel):
    """State for the document being written."""

    document: str = ''


agent = Agent(
    'openai:gpt-4.1',
    instructions='Be fun!',
    deps_type=StateDeps[DocumentState],
)
app = agent.to_ag_ui(deps=StateDeps(DocumentState()))


@agent.tool
async def update_state(ctx: RunContext[StateDeps[DocumentState]]) -> StateSnapshotEvent:
    return StateSnapshotEvent(
        type=EventType.STATE_SNAPSHOT,
        snapshot=ctx.deps.state,
    )


@agent.tool_plain
async def custom_events() -> list[CustomEvent]:
    return [
        CustomEvent(
            type=EventType.CUSTOM,
            name='count',
            value=1,
        ),
        CustomEvent(
            type=EventType.CUSTOM,
            name='count',
            value=2,
        ),
    ]

由于 app 是一个 ASGI 应用程序，它可以与任何 ASGI 服务器一起使用

uvicorn ag_ui_tool_events:app --host 0.0.0.0 --port 9000

示例

有关如何使用 to_ag_ui() 的更多示例，请参见 pydantic_ai_examples.ag_ui，其中包含一个可与 AG-UI Dojo 一起使用的服务器。