AG-UI 사용하여 상태 관리

C에서 State-Aware 에이전트 만들기#

상태 모델 정의

먼저 상태 구조에 대한 클래스를 정의합니다.

using System.Text.Json.Serialization;

namespace RecipeAssistant;

// State response wrapper
internal sealed class RecipeResponse
{
    [JsonPropertyName("recipe")]
    public RecipeState Recipe { get; set; } = new();
}

// Recipe state model
internal sealed class RecipeState
{
    [JsonPropertyName("title")]
    public string Title { get; set; } = string.Empty;

    [JsonPropertyName("cuisine")]
    public string Cuisine { get; set; } = string.Empty;

    [JsonPropertyName("ingredients")]
    public List<string> Ingredients { get; set; } = [];

    [JsonPropertyName("steps")]
    public List<string> Steps { get; set; } = [];

    [JsonPropertyName("prep_time_minutes")]
    public int PrepTimeMinutes { get; set; }

    [JsonPropertyName("cook_time_minutes")]
    public int CookTimeMinutes { get; set; }

    [JsonPropertyName("skill_level")]
    public string SkillLevel { get; set; } = string.Empty;
}

// JSON serialization context
[JsonSerializable(typeof(RecipeResponse))]
[JsonSerializable(typeof(RecipeState))]
[JsonSerializable(typeof(System.Text.Json.JsonElement))]
internal sealed partial class RecipeSerializerContext : JsonSerializerContext;

상태 관리 미들웨어 구현

클라이언트가 상태를 보내는 시기를 감지하고 에이전트의 응답을 조정하여 상태 관리를 처리하는 미들웨어를 만듭니다.

using System.Runtime.CompilerServices;
using System.Text.Json;
using Microsoft.Agents.AI;
using Microsoft.Extensions.AI;

internal sealed class SharedStateAgent : DelegatingAIAgent
{
    private readonly JsonSerializerOptions _jsonSerializerOptions;

    public SharedStateAgent(AIAgent innerAgent, JsonSerializerOptions jsonSerializerOptions)
        : base(innerAgent)
    {
        this._jsonSerializerOptions = jsonSerializerOptions;
    }

    protected override Task<AgentResponse> RunCoreAsync(
        IEnumerable<ChatMessage> messages,
        AgentSession? session = null,
        AgentRunOptions? options = null,
        CancellationToken cancellationToken = default)
    {
        return this.RunStreamingAsync(messages, session, options, cancellationToken)
            .ToAgentResponseAsync(cancellationToken);
    }

    protected override async IAsyncEnumerable<AgentResponseUpdate> RunCoreStreamingAsync(
        IEnumerable<ChatMessage> messages,
        AgentSession? session = null,
        AgentRunOptions? options = null,
        [EnumeratorCancellation] CancellationToken cancellationToken = default)
    {
        // Check if the client sent state in the request
        if (options is not ChatClientAgentRunOptions { ChatOptions.AdditionalProperties: { } properties } chatRunOptions ||
            !properties.TryGetValue("ag_ui_state", out object? stateObj) ||
            stateObj is not JsonElement state ||
            state.ValueKind != JsonValueKind.Object)
        {
            // No state management requested, pass through to inner agent
            await foreach (var update in this.InnerAgent.RunStreamingAsync(messages, session, options, cancellationToken).ConfigureAwait(false))
            {
                yield return update;
            }
            yield break;
        }

        // Check if state has properties (not empty {})
        bool hasProperties = false;
        foreach (JsonProperty _ in state.EnumerateObject())
        {
            hasProperties = true;
            break;
        }

        if (!hasProperties)
        {
            // Empty state - treat as no state
            await foreach (var update in this.InnerAgent.RunStreamingAsync(messages, session, options, cancellationToken).ConfigureAwait(false))
            {
                yield return update;
            }
            yield break;
        }

        // First run: Generate structured state update
        var firstRunOptions = new ChatClientAgentRunOptions
        {
            ChatOptions = chatRunOptions.ChatOptions.Clone(),
            AllowBackgroundResponses = chatRunOptions.AllowBackgroundResponses,
            ContinuationToken = chatRunOptions.ContinuationToken,
            ChatClientFactory = chatRunOptions.ChatClientFactory,
        };

        // Configure JSON schema response format for structured state output
        firstRunOptions.ChatOptions.ResponseFormat = ChatResponseFormat.ForJsonSchema<RecipeResponse>(
            schemaName: "RecipeResponse",
            schemaDescription: "A response containing a recipe with title, skill level, cooking time, preferences, ingredients, and instructions");

        // Add current state to the conversation - state is already a JsonElement
        ChatMessage stateUpdateMessage = new(
            ChatRole.System,
            [
                new TextContent("Here is the current state in JSON format:"),
                new TextContent(JsonSerializer.Serialize(state, this._jsonSerializerOptions.GetTypeInfo(typeof(JsonElement)))),
                new TextContent("The new state is:")
            ]);

        var firstRunMessages = messages.Append(stateUpdateMessage);

        // Collect all updates from first run
        var allUpdates = new List<AgentResponseUpdate>();
        await foreach (var update in this.InnerAgent.RunStreamingAsync(firstRunMessages, session, firstRunOptions, cancellationToken).ConfigureAwait(false))
        {
            allUpdates.Add(update);

            // Yield all non-text updates (tool calls, etc.)
            bool hasNonTextContent = update.Contents.Any(c => c is not TextContent);
            if (hasNonTextContent)
            {
                yield return update;
            }
        }

        var response = allUpdates.ToAgentResponse();

        // Try to deserialize the structured state response
        JsonElement stateSnapshot;
        try
        {
            stateSnapshot = JsonSerializer.Deserialize<JsonElement>(response.Text, this._jsonSerializerOptions);
        }
        catch (JsonException)
        {
            yield break;
        }

        // Serialize and emit as STATE_SNAPSHOT via DataContent
        byte[] stateBytes = JsonSerializer.SerializeToUtf8Bytes(
            stateSnapshot,
            this._jsonSerializerOptions.GetTypeInfo(typeof(JsonElement)));
        yield return new AgentResponseUpdate
        {
            Contents = [new DataContent(stateBytes, "application/json")]
        };

        // Second run: Generate user-friendly summary
        var secondRunMessages = messages.Concat(response.Messages).Append(
            new ChatMessage(
                ChatRole.System,
                [new TextContent("Please provide a concise summary of the state changes in at most two sentences.")]));

        await foreach (var update in this.InnerAgent.RunStreamingAsync(secondRunMessages, session, options, cancellationToken).ConfigureAwait(false))
        {
            yield return update;
        }
    }
}

상태 관리를 사용하여 에이전트 구성

using Microsoft.Agents.AI;
using Azure.AI.Projects;
using Azure.Identity;

AIAgent CreateRecipeAgent(JsonSerializerOptions jsonSerializerOptions)
{
    string endpoint = Environment.GetEnvironmentVariable("AZURE_OPENAI_ENDPOINT")
        ?? throw new InvalidOperationException("AZURE_OPENAI_ENDPOINT is not set.");
    string deploymentName = Environment.GetEnvironmentVariable("AZURE_OPENAI_DEPLOYMENT_NAME")
        ?? throw new InvalidOperationException("AZURE_OPENAI_DEPLOYMENT_NAME is not set.");

    // Create base agent
    AIAgent baseAgent = new AIProjectClient(
        new Uri(endpoint),
        new DefaultAzureCredential())
        .AsAIAgent(
            model: deploymentName,
            name: "RecipeAgent",
        instructions: """
            You are a helpful recipe assistant. When users ask you to create or suggest a recipe,
            respond with a complete RecipeResponse JSON object that includes:
            - recipe.title: The recipe name
            - recipe.cuisine: Type of cuisine (e.g., Italian, Mexican, Japanese)
            - recipe.ingredients: Array of ingredient strings with quantities
            - recipe.steps: Array of cooking instruction strings
            - recipe.prep_time_minutes: Preparation time in minutes
            - recipe.cook_time_minutes: Cooking time in minutes
            - recipe.skill_level: One of "beginner", "intermediate", or "advanced"

            Always include all fields in the response. Be creative and helpful.
            """);

    // Wrap with state management middleware
    return new SharedStateAgent(baseAgent, jsonSerializerOptions);
}

에이전트 엔드포인트 매핑

using Microsoft.Agents.AI.Hosting.AGUI.AspNetCore;

WebApplicationBuilder builder = WebApplication.CreateBuilder(args);
builder.Services.AddHttpClient().AddLogging();
builder.Services.ConfigureHttpJsonOptions(options =>
    options.SerializerOptions.TypeInfoResolverChain.Add(RecipeSerializerContext.Default));
builder.Services.AddAGUI();

WebApplication app = builder.Build();

var jsonOptions = app.Services.GetRequiredService<IOptions<Microsoft.AspNetCore.Http.Json.JsonOptions>>().Value;
AIAgent recipeAgent = CreateRecipeAgent(jsonOptions.SerializerOptions);
app.MapAGUI("/", recipeAgent);

await app.RunAsync();

주요 개념

• 상태 검색: 미들웨어가 클라이언트가 상태 관리를 요청하는 시기를 감지하기 위해 ag_ui_state 내에서 ChatOptions.AdditionalProperties을(를) 확인합니다.
• Two-Phase 응답: 먼저 JSON 스키마(구조적 상태)를 생성한 다음 사용자에게 친숙한 요약을 생성합니다.
• 구조적 상태 모델: JSON 속성 이름을 사용하여 상태 구조에 대한 C# 클래스 정의
• JSON 스키마 응답 형식: 구조화된 출력을 확인하는 데 사용 ChatResponseFormat.ForJsonSchema<T>()
• STATE_SNAPSHOT 이벤트: AG-UI 프레임워크가 자동으로 STATE_SNAPSHOT 이벤트로 변환하는 미디어 유형으로 내보내는 DataContentapplication/json 이벤트.
• 상태 컨텍스트: 현재 상태가 시스템 메시지로 삽입되어 에이전트에 컨텍스트를 제공합니다.

작동 방식

1. 클라이언트는 에서 상태를 사용하여 요청을 보냅니다. ChatOptions.AdditionalProperties["ag_ui_state"]
2. 미들웨어는 상태를 검색하고 JSON 스키마 응답 형식으로 첫 번째 실행을 수행합니다.
3. 미들웨어는 시스템 메시지에서 컨텍스트로 현재 상태를 추가합니다.
4. 에이전트가 상태 모델과 일치하는 구조적 상태 업데이트를 생성합니다.
5. 미들웨어는 상태를 직렬화하여 STATE_SNAPSHOT 이벤트로 방출한다 DataContent.
6. 미들웨어는 사용자에게 친숙한 요약을 생성하기 위해 두 번째 실행을 수행합니다.
7. 클라이언트는 상태 스냅샷과 자연어 요약을 모두 받습니다.

클라이언트 구현(C#)

전체 클라이언트 구현 예제는 전체 양방향 상태 흐름을 보여 주는 아래 Python 클라이언트 패턴을 참조하세요.

상태 모델 정의

먼저 상태 구조에 대한 Pydantic 모델을 정의합니다. 이렇게 하면 형식 안전성 및 유효성 검사가 보장됩니다.

from enum import Enum
from pydantic import BaseModel, Field


class SkillLevel(str, Enum):
    """The skill level required for the recipe."""
    BEGINNER = "Beginner"
    INTERMEDIATE = "Intermediate"
    ADVANCED = "Advanced"


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):
    """An ingredient with its details."""
    icon: str = Field(..., description="Emoji icon representing the ingredient (e.g., 🥕)")
    name: str = Field(..., description="Name of the ingredient")
    amount: str = Field(..., description="Amount or quantity of the ingredient")


class Recipe(BaseModel):
    """A complete recipe."""
    title: str = Field(..., description="The title of the recipe")
    skill_level: SkillLevel = Field(..., description="The skill level required")
    special_preferences: list[str] = Field(
        default_factory=list, description="Dietary preferences (e.g., Vegetarian, Gluten-free)"
    )
    cooking_time: CookingTime = Field(..., description="The estimated cooking time")
    ingredients: list[Ingredient] = Field(..., description="Complete list of ingredients")
    instructions: list[str] = Field(..., description="Step-by-step cooking instructions")

상태 스키마

상태 스키마를 정의하여 상태의 구조 및 형식을 지정합니다.

state_schema = {
    "recipe": {"type": "object", "description": "The current recipe"},
}

예측 상태 업데이트

예측 상태 업데이트는 LLM이 툴 인수를 생성할 때 이를 상태에 반영하여 낙관적 UI 업데이트를 지원합니다.

predict_state_config = {
    "recipe": {"tool": "update_recipe", "tool_argument": "recipe"},
}

이 구성은 recipe 상태 필드를 recipe 도구의 update_recipe 인수로 매핑합니다. 에이전트가 도구를 호출하면, LLM이 인수를 생성함과 동시에 인수들이 실시간으로 상태에 스트리밍됩니다.

상태 업데이트 도구 정의

Pydantic 모델을 허용하는 도구 함수를 만듭니다.

from agent_framework import tool


@tool
def update_recipe(recipe: Recipe) -> str:
    """Update the recipe with new or modified content.

    You MUST write the complete recipe with ALL fields, even when changing only a few items.
    When modifying an existing recipe, include ALL existing ingredients and instructions plus your changes.
    NEVER delete existing data - only add or modify.

    Args:
        recipe: The complete recipe object with all details

    Returns:
        Confirmation that the recipe was updated
    """
    return "Recipe updated."

상태 관리를 사용하여 에이전트 만들기

상태 관리를 사용하는 전체 서버 구현은 다음과 같습니다.

"""AG-UI server with state management."""

from agent_framework import Agent
from agent_framework.openai import OpenAIChatCompletionClient
from agent_framework_ag_ui import (
    AgentFrameworkAgent,
    RecipeConfirmationStrategy,
    add_agent_framework_fastapi_endpoint,
)
from azure.identity import AzureCliCredential
from fastapi import FastAPI

# Create the chat agent with tools
agent = Agent(
    name="recipe_agent",
    instructions="""You are a helpful recipe assistant that creates and modifies recipes.

    CRITICAL RULES:
    1. You will receive the current recipe state in the system context
    2. To update the recipe, you MUST use the update_recipe tool
    3. When modifying a recipe, ALWAYS include ALL existing data plus your changes in the tool call
    4. NEVER delete existing ingredients or instructions - only add or modify
    5. After calling the tool, provide a brief conversational message (1-2 sentences)

    When creating a NEW recipe:
    - Provide all required fields: title, skill_level, cooking_time, ingredients, instructions
    - Use actual emojis for ingredient icons (🥕 🧄 🧅 🍅 🌿 🍗 🥩 🧀)
    - Leave special_preferences empty unless specified
    - Message: "Here's your recipe!" or similar

    When MODIFYING or IMPROVING an existing recipe:
    - Include ALL existing ingredients + any new ones
    - Include ALL existing instructions + any new/modified ones
    - Update other fields as needed
    - Message: Explain what you improved (e.g., "I upgraded the ingredients to premium quality")
    - When asked to "improve", enhance with:
      * Better ingredients (upgrade quality, add complementary flavors)
      * More detailed instructions
      * Professional techniques
      * Adjust skill_level if complexity changes
      * Add relevant special_preferences

    Example improvements:
    - Upgrade "chicken" → "organic free-range chicken breast"
    - Add herbs: basil, oregano, thyme
    - Add aromatics: garlic, shallots
    - Add finishing touches: lemon zest, fresh parsley
    - Make instructions more detailed and professional
    """,
    client=OpenAIChatCompletionClient(
        model=deployment_name,
        azure_endpoint=endpoint,
        api_version=os.getenv("AZURE_OPENAI_API_VERSION"),
        credential=AzureCliCredential(),
    ),
    tools=[update_recipe],
)

# Wrap agent with state management
recipe_agent = AgentFrameworkAgent(
    agent=agent,
    name="RecipeAgent",
    description="Creates and modifies recipes with streaming state updates",
    state_schema={
        "recipe": {"type": "object", "description": "The current recipe"},
    },
    predict_state_config={
        "recipe": {"tool": "update_recipe", "tool_argument": "recipe"},
    },
    confirmation_strategy=RecipeConfirmationStrategy(),
)

# Create FastAPI app
app = FastAPI(title="AG-UI Recipe Assistant")
add_agent_framework_fastapi_endpoint(app, recipe_agent, "/")

if __name__ == "__main__":
    import uvicorn
    uvicorn.run(app, host="127.0.0.1", port=8888)

주요 개념

• Pydantic 모델: 형식 안전성 및 유효성 검사를 사용하여 구조화된 상태 정의
• 상태 스키마: 상태 필드 형식을 지정하는 단순 형식
• 예측 상태 구성: 스트리밍 업데이트를 위한 도구 인수에 상태 필드를 매핑합니다.
• 상태 주입: 컨텍스트를 제공하기 위해 현재 상태가 시스템 메시지로 자동으로 삽입됩니다.
• 전체 업데이트: 도구는 델타뿐만 아니라 전체 상태를 작성해야 합니다.
• 확인 전략: 도메인에 대한 승인 메시지 사용자 지정(레시피, 문서, 작업 계획 등)

상태 이벤트의 이해

상태 스냅샷 이벤트

도구가 완료될 때 내보내는 현재 상태의 전체 스냅샷:

{
    "type": "STATE_SNAPSHOT",
    "snapshot": {
        "recipe": {
            "title": "Classic Pasta Carbonara",
            "skill_level": "Intermediate",
            "special_preferences": ["Authentic Italian"],
            "cooking_time": "30 min",
            "ingredients": [
                {"icon": "🍝", "name": "Spaghetti", "amount": "400g"},
                {"icon": "🥓", "name": "Guanciale or bacon", "amount": "200g"},
                {"icon": "🥚", "name": "Egg yolks", "amount": "4"},
                {"icon": "🧀", "name": "Pecorino Romano", "amount": "100g grated"},
                {"icon": "🧂", "name": "Black pepper", "amount": "To taste"}
            ],
            "instructions": [
                "Bring a large pot of salted water to boil",
                "Cut guanciale into small strips and fry until crispy",
                "Beat egg yolks with grated Pecorino and black pepper",
                "Cook spaghetti until al dente",
                "Reserve 1 cup pasta water, then drain pasta",
                "Remove pan from heat, add hot pasta to guanciale",
                "Quickly stir in egg mixture, adding pasta water to create creamy sauce",
                "Serve immediately with extra Pecorino and black pepper"
            ]
        }
    }
}

State Delta 이벤트

JSON 패치 형식을 사용하여, LLM 스트림 도구 인수로 내보내는 증분 상태 업데이트:

{
    "type": "STATE_DELTA",
    "delta": [
        {
            "op": "replace",
            "path": "/recipe",
            "value": {
                "title": "Classic Pasta Carbonara",
                "skill_level": "Intermediate",
                "cooking_time": "30 min",
                "ingredients": [
                    {"icon": "🍝", "name": "Spaghetti", "amount": "400g"}
                ],
                "instructions": ["Bring a large pot of salted water to boil"]
            }
        }
    ]
}

클라이언트 구현

이 패키지는 AG-UI 서버와의 연결을 제공하여 Python 클라이언트 환경을 .NET과 동일한 수준으로 향상시킵니다.

"""AG-UI client with state management."""

import asyncio
import json
import os
from typing import Any

import jsonpatch
from agent_framework import Agent, Message, Role
from agent_framework_ag_ui import AGUIChatClient


async def main():
    """Example client with state tracking."""
    server_url = os.environ.get("AGUI_SERVER_URL", "http://127.0.0.1:8888/")
    print(f"Connecting to AG-UI server at: {server_url}\n")

    # Create AG-UI chat client
    chat_client = AGUIChatClient(server_url=server_url)

    # Wrap with Agent for convenient API
    agent = Agent(
        name="ClientAgent",
        client=chat_client,
        instructions="You are a helpful assistant.",
    )

    # Get a thread for conversation continuity
    thread = agent.create_session()

    # Track state locally
    state: dict[str, Any] = {}

    try:
        while True:
            message = input("\nUser (:q to quit, :state to show state): ")
            if not message.strip():
                continue

            if message.lower() in (":q", "quit"):
                break

            if message.lower() == ":state":
                print(f"\nCurrent state: {json.dumps(state, indent=2)}")
                continue

            print()
            # Stream the agent response with state
            async for update in agent.run(message, session=thread, stream=True):
                # Handle text content
                if update.text:
                    print(update.text, end="", flush=True)

                # Handle state updates
                for content in update.contents:
                    # STATE_SNAPSHOT events come as DataContent with application/json
                    if hasattr(content, 'media_type') and content.media_type == 'application/json':
                        # Parse state snapshot
                        state_data = json.loads(content.data.decode() if isinstance(content.data, bytes) else content.data)
                        state = state_data
                        print("\n[State Snapshot Received]")

                    # STATE_DELTA events are handled similarly
                    # Apply JSON Patch deltas to maintain state
                    if hasattr(content, 'delta') and content.delta:
                        patch = jsonpatch.JsonPatch(content.delta)
                        state = patch.apply(state)
                        print("\n[State Delta Applied]")

            print(f"\n\nCurrent state: {json.dumps(state, indent=2)}")
            print()

    except KeyboardInterrupt:
        print("\n\nExiting...")


if __name__ == "__main__":
    # Install dependencies: pip install agent-framework-ag-ui jsonpatch --pre
    asyncio.run(main())

주요 이점

다음을 AGUIChatClient 제공합니다.

• 간소화된 연결: HTTP/SSE 통신 자동 처리
• 스레드 관리: 대화 연속성을 위한 기본 제공 스레드 ID 추적
• 에이전트 통합: 친숙한 API와 Agent 원활하게 작동
• 상태 처리: 서버에서 상태 이벤트를 자동으로 해석합니다.
• .NET과의 패리티: 언어 간 일관된 경험

확인 전략 사용

매개 confirmation_strategy 변수를 사용하면 도메인에 대한 승인 메시지를 사용자 지정할 수 있습니다.

from agent_framework_ag_ui import RecipeConfirmationStrategy

recipe_agent = AgentFrameworkAgent(
    agent=agent,
    state_schema={"recipe": {"type": "object", "description": "The current recipe"}},
    predict_state_config={"recipe": {"tool": "update_recipe", "tool_argument": "recipe"}},
    confirmation_strategy=RecipeConfirmationStrategy(),
)

사용 가능한 전략:

• DefaultConfirmationStrategy() - 모든 에이전트에 대한 일반 메시지
• RecipeConfirmationStrategy() - 레시피별 메시지
• DocumentWriterConfirmationStrategy() - 문서 편집 메시지
• TaskPlannerConfirmationStrategy() - 작업 계획 메시지

필요한 메서드를 ConfirmationStrategy 상속하고 구현하여 사용자 지정 전략을 만들 수도 있습니다.

상호 작용 예제

서버 및 클라이언트가 실행 중인 경우:

User (:q to quit, :state to show state): I want to make a classic Italian pasta carbonara

[Run Started]
[Calling Tool: update_recipe]
[State Updated]
[State Updated]
[State Updated]
[Tool Result: Recipe updated.]
Here's your recipe!
[Run Finished]

============================================================
CURRENT STATE
============================================================

recipe:
  title: Classic Pasta Carbonara
  skill_level: Intermediate
  special_preferences: ['Authentic Italian']
  cooking_time: 30 min
  ingredients:
    - 🍝 Spaghetti: 400g
    - 🥓 Guanciale or bacon: 200g
    - 🥚 Egg yolks: 4
    - 🧀 Pecorino Romano: 100g grated
    - 🧂 Black pepper: To taste
  instructions:
    1. Bring a large pot of salted water to boil
    2. Cut guanciale into small strips and fry until crispy
    3. Beat egg yolks with grated Pecorino and black pepper
    4. Cook spaghetti until al dente
    5. Reserve 1 cup pasta water, then drain pasta
    6. Remove pan from heat, add hot pasta to guanciale
    7. Quickly stir in egg mixture, adding pasta water to create creamy sauce
    8. Serve immediately with extra Pecorino and black pepper

============================================================

적용 중인 예측 상태 업데이트

예측 상태 업데이트를 사용하는 경우 predict_state_config, 클라이언트가 실시간으로 도구 인수를 생성할 때 STATE_DELTA 이벤트를 수신합니다. 이는 LLM이 도구를 실행하기 전에 발생합니다.

// Agent starts generating tool call for update_recipe
// Client receives STATE_DELTA events as the recipe argument streams:

// First delta - partial recipe with title
{
  "type": "STATE_DELTA",
  "delta": [{"op": "replace", "path": "/recipe", "value": {"title": "Classic Pasta"}}]
}

// Second delta - title complete with more fields
{
  "type": "STATE_DELTA",
  "delta": [{"op": "replace", "path": "/recipe", "value": {
    "title": "Classic Pasta Carbonara",
    "skill_level": "Intermediate"
  }}]
}

// Third delta - ingredients starting to appear
{
  "type": "STATE_DELTA",
  "delta": [{"op": "replace", "path": "/recipe", "value": {
    "title": "Classic Pasta Carbonara",
    "skill_level": "Intermediate",
    "cooking_time": "30 min",
    "ingredients": [
      {"icon": "🍝", "name": "Spaghetti", "amount": "400g"}
    ]
  }}]
}

// ... more deltas as the LLM generates the complete recipe

이를 통해 클라이언트는 에이전트가 생각하는 대로 실시간으로 낙관적 UI 업데이트를 표시하여 사용자에게 즉각적인 피드백을 제공할 수 있습니다.

휴먼 인 더 루프가 있는 상태

다음을 설정 require_confirmation=True하여 상태 관리를 승인 워크플로와 결합할 수 있습니다.

recipe_agent = AgentFrameworkAgent(
    agent=agent,
    state_schema={"recipe": {"type": "object", "description": "The current recipe"}},
    predict_state_config={"recipe": {"tool": "update_recipe", "tool_argument": "recipe"}},
    require_confirmation=True,  # Require approval for state changes
    confirmation_strategy=RecipeConfirmationStrategy(),
)

사용하도록 설정된 경우:

1. 에이전트가 도구 인수를 생성할 때 상태 업데이트 스트림(이벤트를 통한 STATE_DELTA 예측 업데이트)
2. 에이전트가 도구를 실행하기 전에 승인을 요청합니다(이벤트를 통해 FUNCTION_APPROVAL_REQUEST ).
3. 승인되면 도구가 실행되고 최종 상태가 내보내집니다(이벤트를 통해 STATE_SNAPSHOT ).
4. 거부되면 예측 상태 변경 내용이 삭제됩니다.

고급 상태 패턴

여러 필드가 있는 복합 상태

다양한 도구를 사용하여 여러 상태 필드를 관리할 수 있습니다.

from pydantic import BaseModel


class TaskStep(BaseModel):
    """A single task step."""
    description: str
    status: str = "pending"
    estimated_duration: str = "5 min"


@tool
def generate_task_steps(steps: list[TaskStep]) -> str:
    """Generate task steps for a given task."""
    return f"Generated {len(steps)} steps."


@tool
def update_preferences(preferences: dict[str, Any]) -> str:
    """Update user preferences."""
    return "Preferences updated."


# Configure with multiple state fields
agent_with_multiple_state = AgentFrameworkAgent(
    agent=agent,
    state_schema={
        "steps": {"type": "array", "description": "List of task steps"},
        "preferences": {"type": "object", "description": "User preferences"},
    },
    predict_state_config={
        "steps": {"tool": "generate_task_steps", "tool_argument": "steps"},
        "preferences": {"tool": "update_preferences", "tool_argument": "preferences"},
    },
)

와일드카드 도구 인수 사용

도구가 복잡한 중첩 데이터를 반환하는 경우 모든 도구 인수를 상태에 매핑하는 데 사용합니다 "*" .

@tool
def create_document(title: str, content: str, metadata: dict[str, Any]) -> str:
    """Create a document with title, content, and metadata."""
    return "Document created."


# Map all tool arguments to document state
predict_state_config = {
    "document": {"tool": "create_document", "tool_argument": "*"}
}

그러면 전체 도구 호출(모든 인수)이 document 상태 필드에 매핑됩니다.

모범 사례

Pydantic 모델 사용

형식 안전을 위해 구조화된 모델을 정의합니다.

class Recipe(BaseModel):
    """Use Pydantic models for structured, validated state."""
    title: str
    skill_level: SkillLevel
    ingredients: list[Ingredient]
    instructions: list[str]

이점:

• 형식 안전성: 데이터 형식의 자동 유효성 검사
• 설명서: 필드 설명은 설명서로 사용됩니다.
• IDE 지원: 자동 완성 및 형식 검사
• Serialization: 자동 JSON 변환

전체 상태 업데이트

항상 델타뿐만 아니라 전체 상태를 작성합니다.

@tool
def update_recipe(recipe: Recipe) -> str:
    """
    You MUST write the complete recipe with ALL fields.
    When modifying a recipe, include ALL existing ingredients and
    instructions plus your changes. NEVER delete existing data.
    """
    return "Recipe updated."

이렇게 하면 상태 일관성과 적절한 예측 업데이트가 보장됩니다.

매개 변수 이름 일치

도구 매개 변수 이름이 구성과 일치하는지 tool_argument 확인합니다.

# Tool parameter name
def update_recipe(recipe: Recipe) -> str:  # Parameter name: 'recipe'
    ...

# Must match in predict_state_config
predict_state_config = {
    "recipe": {"tool": "update_recipe", "tool_argument": "recipe"}  # Same name
}

지침에 컨텍스트 제공

상태 관리에 대한 명확한 지침을 포함합니다.

agent = Agent(
    instructions="""
    CRITICAL RULES:
    1. You will receive the current recipe state in the system context
    2. To update the recipe, you MUST use the update_recipe tool
    3. When modifying a recipe, ALWAYS include ALL existing data plus your changes
    4. NEVER delete existing ingredients or instructions - only add or modify
    """,
    ...
)

확인 전략 사용

도메인에 대한 승인 메시지를 사용자 지정합니다.

from agent_framework_ag_ui import RecipeConfirmationStrategy

recipe_agent = AgentFrameworkAgent(
    agent=agent,
    confirmation_strategy=RecipeConfirmationStrategy(),  # Domain-specific messages
)

다음 단계

이제 모든 핵심 AG-UI 기능을 배웠습니다. 다음으로 다음을 수행할 수 있습니다.

• 에이전트 프레임워크 설명서 살펴보기
• 모든 AG-UI 기능을 결합한 완전한 애플리케이션 빌드
• 프로덕션에 AG-UI 서비스 배포

추가 리소스

• AG-UI 개요
• 시작하기
• 백 엔드 도구 렌더링