에이전트 사용자 상호작용

에이전트 사용자 상호작용 (AG-UI)

AG-UI Dojo 예제 앱과 함께 Pydantic AI 에이전트를 사용하는 예시예요.

AG-UI 통합에 대한 자세한 내용은 AG-UI 문서를 참고해요.

다음을 시연해요:

• AG-UI
• 툴

출처: 문서

본문

사전 요구사항

• OpenAI API 키

예제 실행하기

의존성이 설치되고 환경 변수가 설정되면 명령줄 창이 두 개 필요해요.

Pydantic AI AG-UI 백엔드

OpenAI API 키를 설정해요:

Terminal

export OPENAI_API_KEY=<your api key>

Pydantic AI AG-UI 예제 백엔드를 시작해요.

• pip
• uv

Terminal

python -m pydantic_ai_examples.ag_ui

Terminal

uv run -m pydantic_ai_examples.ag_ui

AG-UI Dojo 예제 프론트엔드

다음으로 AG-UI Dojo 예제 프론트엔드를 실행해요.

1. AG-UI 저장소를 클론해요

   Terminal

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

2. 공식 지침에 따라 사전 요구사항을 설치한 다음, 저장소 루트에서 의존성을 설치하고 프로젝트를 빌드해요:

   Terminal

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

3. apps/dojo 디렉토리로 이동해 Dojo 앱을 실행해요:

   Terminal

   cd apps/dojo
   pnpm dev
   

4. http://localhost:3000/pydantic-ai를 방문해요

5. 사이드바에서 Pydantic AI 뷰를 선택해요

기능 예시

에이전트형 채팅 (Agentic Chat)

Pydantic AI 서버 측 툴과 AG-UI 클라이언트 측 툴을 포함한 기본 에이전트 상호작용을 시연해요.

예제를 실행했다면 http://localhost:3000/pydantic-ai/feature/agentic_chat에서 볼 수 있어요.

에이전트 툴

• time - 시간대의 현재 시간을 확인하는 Pydantic AI 툴
• background - 클라이언트 창의 배경색을 설정하는 AG-UI 툴

에이전트 프롬프트

What is the time in New York?

Change the background to blue

AG-UI 툴과 Pydantic AI 툴을 모두 섞는 복잡한 예시:

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

에이전트형 채팅 - 코드

agentic_chat.py

from __future__ import annotations

from datetime import datetime
from zoneinfo import ZoneInfo

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-mini')


@agent.tool_plain
async def current_time(timezone: str = 'UTC') -> str:
    """Get the current time in ISO format.

    Args:
        timezone: The timezone to use.

    Returns:
        The current time in ISO format string.
    """
    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'])])

에이전트형 생성적 UI (Agentic Generative UI)

에이전트가 무슨 일이 일어나고 있는지 사용자에게 알리기 위해 프론트엔드로 업데이트를 보내는 오래 실행되는 작업을 시연해요.

예제를 실행했다면 http://localhost:3000/pydantic-ai/feature/agentic_generative_ui에서 볼 수 있어요.

계획 프롬프트

Create a plan for breakfast and execute it

에이전트형 생성적 UI - 코드

agentic_generative_ui.py

from __future__ import annotations

from textwrap import dedent
from typing import Any, Literal

from pydantic import BaseModel, Field
from starlette.applications import Starlette
from starlette.requests import Request
from starlette.responses import Response
from starlette.routing import Route

from ag_ui.core import EventType, StateDeltaEvent, StateSnapshotEvent
from pydantic_ai import Agent
from pydantic_ai.ui.ag_ui import AGUIAdapter

StepStatus = Literal['pending', 'completed']


class Step(BaseModel):
    """Represents a step in a plan."""

    description: str = Field(description='The description of the step')
    status: StepStatus = Field(
        default='pending',
        description='The status of the step (e.g., pending, completed)',
    )


class Plan(BaseModel):
    """Represents a plan with multiple steps."""

    steps: list[Step] = Field(
        default_factory=list[Step], description='The steps in the plan'
    )


class JSONPatchOp(BaseModel):
    """A class representing a JSON Patch operation (RFC 6902)."""

    op: Literal['add', 'remove', 'replace', 'move', 'copy', 'test'] = Field(
        description='The operation to perform: add, remove, replace, move, copy, or test',
    )
    path: str = Field(description='JSON Pointer (RFC 6901) to the target location')
    value: Any = Field(
        default=None,
        description='The value to apply (for add, replace operations)',
    )
    from_: str | None = Field(
        default=None,
        alias='from',
        description='Source path (for move, copy operations)',
    )


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
        - Do NOT confirm the creation or updates in a message
        - Do NOT ask the user for additional information or next steps

        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.
        """
    ),
)


@agent.tool_plain
async def create_plan(steps: list[str]) -> StateSnapshotEvent:
    """Create a plan with multiple steps.

    Args:
        steps: List of step descriptions to create the plan.

    Returns:
        StateSnapshotEvent containing the initial state of the steps.
    """
    plan: 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:
    """Update the plan with new steps or changes.

    Args:
        index: The index of the step to update.
        description: The new description for the step.
        status: The new status for the step.

    Returns:
        StateDeltaEvent containing the changes made to the plan.
    """
    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,
    )


async def run_agent(request: Request) -> Response:
    return await AGUIAdapter.dispatch_request(request, agent=agent)


app = Starlette(routes=[Route('/', run_agent, methods=['POST'])])

인간 개입(Human in the Loop)

에이전트가 계획을 세우고 사용자가 체크박스로 승인할 수 있는 단순한 인간 개입 워크플로를 시연해요.

작업 계획 툴

• generate_task_steps - 단계를 생성하고 확인하는 AG-UI 툴

작업 계획 프롬프트

Generate a list of steps for cleaning a car for me to review

인간 개입 - 코드

human_in_the_loop.py

from __future__ import annotations

from textwrap import dedent

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-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
        """
    ),
)


async def run_agent(request: Request) -> Response:
    return await AGUIAdapter.dispatch_request(request, agent=agent)


app = Starlette(routes=[Route('/', run_agent, methods=['POST'])])

예측 상태 업데이트 (Predictive State Updates)

에이전트 응답에 기반해 UI 상태를 업데이트하는 데 예측 상태 업데이트 기능을 사용하는 방법을 시연해요. 사용자 확인을 통한 사용자 상호작용도 포함해요.

예제를 실행했다면 http://localhost:3000/pydantic-ai/feature/predictive_state_updates에서 볼 수 있어요.

스토리 툴

• write_document - 문서를 창에 쓰는 AG-UI 툴
• document_predict_state - write_document 툴에 대한 문서 상태 예측을 가능하게 하는 Pydantic AI 툴

이것은 공유 상태 정보에 기반한 커스텀 지시사항을 사용하는 방법도 보여줘요.

스토리 예시

시작 문서 텍스트

Bruce was a good dog,

에이전트 프롬프트

Help me complete my story about bruce the dog, is should be no longer than a sentence.

예측 상태 업데이트 - 코드

predictive_state_updates.py

from __future__ import annotations

from dataclasses import replace
from textwrap import dedent

from pydantic import BaseModel
from starlette.applications import Starlette
from starlette.requests import Request
from starlette.responses import Response
from starlette.routing import Route

from ag_ui.core import CustomEvent, EventType
from pydantic_ai import Agent, RunContext
from pydantic_ai.ui import StateDeps
from pydantic_ai.ui.ag_ui import AGUIAdapter


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

    document: str = ''


agent = Agent('openai:gpt-5-mini', deps_type=StateDeps[DocumentState])


# AG-UI 이벤트를 반환하는 툴은 이벤트 스트림의 일부로 클라이언트에 보내져요.
# 단일 이벤트와 이벤트의 iterable 모두 지원돼요.
@agent.tool_plain
async def document_predict_state() -> list[CustomEvent]:
    """Enable document state prediction.

    Returns:
        CustomEvent containing the event to enable state prediction.
    """
    return [
        CustomEvent(
            type=EventType.CUSTOM,
            name='PredictState',
            value=[
                {
                    'state_key': 'document',
                    'tool': 'write_document',
                    'tool_argument': 'document',
                },
            ],
        ),
    ]


@agent.instructions()
async def story_instructions(ctx: RunContext[StateDeps[DocumentState]]) -> str:
    """Provide instructions for writing document if present.

    Args:
        ctx: The run context containing document state information.

    Returns:
        Instructions string for the document writing agent.
    """
    return dedent(
        f"""You are a helpful assistant for writing documents.

        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.

        When you have written the document, DO NOT repeat it as a message.
        If accepted briefly summarize the changes you made, 2 sentences
        max, otherwise ask the user to clarify what they want to change.

        This is the current document:

        {ctx.deps.state.document}
        """
    )


deps = StateDeps(DocumentState())


async def run_agent(request: Request) -> Response:
    # `dispatch_request`는 요청에서 `deps.state`를 변경하므로, 각 요청에 자체 사본을 준다.
    return await AGUIAdapter.dispatch_request(request, agent=agent, deps=replace(deps))


app = Starlette(routes=[Route('/', run_agent, methods=['POST'])])

공유 상태 (Shared State)

UI와 에이전트 사이의 공유 상태를 사용하는 방법을 시연해요.

에이전트로 보내진 상태는 함수 기반 지시사항으로 감지돼요. 그런 다음 커스텀 pydantic 모델로 데이터를 검증하고, 에이전트가 따를 지시사항을 만드는 데 사용한 다음 AG-UI 툴로 클라이언트에 보내요.

예제를 실행했다면 http://localhost:3000/pydantic-ai/feature/shared_state에서 볼 수 있어요.

레시피 툴

• display_recipe - 레시피를 그래픽 형식으로 표시하는 AG-UI 툴

레시피 예시

1. 레시피의 기본 설정을 커스터마이즈해요
2. Improve with AI를 클릭해요

공유 상태 - 코드

shared_state.py

from __future__ import annotations

from dataclasses import replace
from enum import Enum
from textwrap import dedent

from pydantic import BaseModel, Field
from starlette.applications import Starlette
from starlette.requests import Request
from starlette.responses import Response
from starlette.routing import Route

from ag_ui.core import EventType, StateSnapshotEvent
from pydantic_ai import Agent, RunContext
from pydantic_ai.ui import StateDeps
from pydantic_ai.ui.ag_ui import AGUIAdapter


class SkillLevel(str, Enum):
    """The level of skill required for the recipe."""

    BEGINNER = 'Beginner'
    INTERMEDIATE = 'Intermediate'
    ADVANCED = 'Advanced'


class SpecialPreferences(str, Enum):
    """Special preferences for the recipe."""

    HIGH_PROTEIN = 'High Protein'
    LOW_CARB = 'Low Carb'
    SPICY = 'Spicy'
    BUDGET_FRIENDLY = 'Budget-Friendly'
    ONE_POT_MEAL = 'One-Pot Meal'
    VEGETARIAN = 'Vegetarian'
    VEGAN = 'Vegan'


class CookingTime(str, Enum):
    """The cooking time of the recipe."""

    FIVE_MIN = '5 min'
    FIFTEEN_MIN = '15 min'
    THIRTY_MIN = '30 min'
    FORTY_FIVE_MIN = '45 min'
    SIXTY_PLUS_MIN = '60+ min'


class Ingredient(BaseModel):
    """A class representing an ingredient in a recipe."""

    icon: str = Field(
        default='ingredient',
        description="The icon emoji (not emoji code like '\U0001f35e', but the actual emoji like 🥕) of the ingredient",
    )
    name: str
    amount: str


class Recipe(BaseModel):
    """A class representing a recipe."""

    skill_level: SkillLevel = Field(
        default=SkillLevel.BEGINNER,
        description='The skill level required for the recipe',
    )
    special_preferences: list[SpecialPreferences] = Field(
        default_factory=list[SpecialPreferences],
        description='Any special preferences for the recipe',
    )
    cooking_time: CookingTime = Field(
        default=CookingTime.FIVE_MIN, description='The cooking time of the recipe'
    )
    ingredients: list[Ingredient] = Field(
        default_factory=list[Ingredient],
        description='Ingredients for the recipe',
    )
    instructions: list[str] = Field(
        default_factory=list[str], description='Instructions for the recipe'
    )


class RecipeSnapshot(BaseModel):
    """A class representing the state of the recipe."""

    recipe: Recipe = Field(
        default_factory=Recipe, description='The current state of the recipe'
    )


agent = Agent('openai:gpt-5-mini', deps_type=StateDeps[RecipeSnapshot])


@agent.tool_plain
async def display_recipe(recipe: Recipe) -> StateSnapshotEvent:
    """Display the recipe to the user.

    Args:
        recipe: The recipe to display.

    Returns:
        StateSnapshotEvent containing the recipe snapshot.
    """
    return StateSnapshotEvent(
        type=EventType.STATE_SNAPSHOT,
        snapshot={'recipe': recipe},
    )


@agent.instructions
async def recipe_instructions(ctx: RunContext[StateDeps[RecipeSnapshot]]) -> str:
    """Instructions for the recipe generation agent.

    Args:
        ctx: The run context containing recipe state information.

    Returns:
        Instructions string for the recipe generation agent.
    """
    return dedent(
        f"""
        You are a helpful assistant for creating recipes.

        IMPORTANT:
        - 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
        - Do NOT run the `display_recipe` tool multiple times in a row

        Once you have created the updated recipe and displayed it to the user,
        summarise the changes in one sentence, don't describe the recipe in
        detail or send it as a message to the user.

        The current state of the recipe is:

        {ctx.deps.state.recipe.model_dump_json(indent=2)}
        """,
    )


deps = StateDeps(RecipeSnapshot())


async def run_agent(request: Request) -> Response:
    # `dispatch_request`는 요청에서 `deps.state`를 변경하므로, 각 요청에 자체 사본을 준다.
    return await AGUIAdapter.dispatch_request(request, agent=agent, deps=replace(deps))


app = Starlette(routes=[Route('/', run_agent, methods=['POST'])])

툴 기반 생성적 UI (Tool Based Generative UI)

이 예시는 사용자 확인과 함께 툴 출력에 대한 커스터마이즈된 렌더링을 시연해요.

예제를 실행했다면 http://localhost:3000/pydantic-ai/feature/tool_based_generative_ui에서 볼 수 있어요.

하이쿠 툴

• generate_haiku - 영어와 일본어로 하이쿠를 표시하는 AG-UI 툴

하이쿠 프롬프트

Generate a haiku about formula 1

툴 기반 생성적 UI - 코드

tool_based_generative_ui.py

from __future__ import annotations

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-mini')


async def run_agent(request: Request) -> Response:
    return await AGUIAdapter.dispatch_request(request, agent=agent)


app = Starlette(routes=[Route('/', run_agent, methods=['POST'])])

더 알아보기 (Learn more)

• Pydantic AI 문서: 에이전트 사용자 상호작용 (AG-UI)