Gerenciamento de Estado com AG-UI

O estado AG-UI é o JSON visível ao cliente associado a uma execução. No .NET, a integração fornece dois mecanismos explícitos:

• Estado de leitura fornecido pelo cliente da RunAgentInput de origem.
• Mapeie chamadas de ferramenta ou resultados selecionados para eventos de estado do AG-UI com AGUIStreamOptions.

O mapeamento de estado é opcional. Os resultados arbitrários da ferramenta não se tornam automaticamente um estado compartilhado.

Ler o estado do cliente

MapAGUIServer armazena a origem RunAgentInput em ChatOptions. Um middleware de agente de delegação ou de cliente de chat pode recuperá-lo com TryGetRunAgentInput:

using System.Text.Json;
using AGUI.Abstractions;
using AGUI.Server;
using Microsoft.Extensions.AI;

static bool TryGetClientState(ChatOptions options, out JsonElement state)
{
    if (options.TryGetRunAgentInput(out RunAgentInput? input) &&
        input.State is { ValueKind: not JsonValueKind.Undefined } value)
    {
        state = value;
        return true;
    }

    state = default;
    return false;
}

O estado do cliente é a entrada da solicitação. Valide sua estrutura e valores antes de usá-la em prompts, roteamento ou operações privilegiadas.

Gerar um instantâneo do estado

Mapeie um resultado de ferramenta para STATE_SNAPSHOT quando a ferramenta retorna o estado completo:

using AGUI.Server;

AGUIStreamOptions streamOptions = new AGUIStreamOptions()
    .MapResultAsStateSnapshot("generate_recipe");

app.MapAGUIServer("/", agent).WithMetadata(streamOptions);

MapResultAsStateSnapshot requer que o FunctionResultContent.Result valor seja um JsonElement. Serialize um POCO, dicionário ou coleção para JsonElement na ferramenta antes de retorná-la. O resultado de generate_recipe passa a ser o instantâneo e substitui o estado compartilhado atual do cliente.

Para outros tipos de resultados, use MapResult com um mapeador personalizado que constrói o StateSnapshotEvent.

Emitir deltas de estado

Mapeie um resultado da ferramenta para STATE_DELTA quando ele retornar um Patch JSON RFC 6902:

AGUIStreamOptions streamOptions = new AGUIStreamOptions()
    .MapResultAsStateSnapshot("create_plan")
    .MapResultAsStateDelta("update_plan_step");

app.MapAGUIServer("/", agent).WithMetadata(streamOptions);

Use um instantâneo para inicializar ou substituir o estado e deltas para alterações incrementais.

MapResultAsStateDelta também requer um JsonElement resultado. O elemento deve conter um array de JSON Patch RFC 6902. Use MapResult com um mapeador personalizado se a ferramenta retornar outra representação.

Mapear chamadas de ferramenta para estado

AGUIStreamOptions.MapCall mapeia um FunctionCallContent selecionado para eventos adicionais da AG-UI emitidos após os eventos normais de chamada de ferramenta. Use-o quando o estado deriva de argumentos de ferramenta em vez do resultado da ferramenta:

AGUIStreamOptions streamOptions = new AGUIStreamOptions()
    .MapCall("write_document", call =>
    {
        if (call.Arguments?.TryGetValue("document", out object? document) is not true)
        {
            return [];
        }

        JsonElement snapshot = JsonSerializer.SerializeToElement(new { document });
        return [new StateSnapshotEvent { Snapshot = snapshot }];
    });

app.MapAGUIServer("/", agent).WithMetadata(streamOptions);

O aplicativo é proprietário do mapeamento e da forma de estado. MapCall não infere o estado de argumentos arbitrários da ferramenta ou suprime a execução normal da ferramenta. As atualizações incrementais exigem que o cliente do modelo subjacente exponha os argumentos de chamada de ferramenta em fluxo e que o aplicativo configure a extração correspondente desses argumentos.

Receber estado em um cliente .NET

O cliente .NET do AG-UI expõe eventos do protocolo de estado por meio de ChatResponseUpdate.RawRepresentation:

await foreach (AgentResponseUpdate update in agent.RunStreamingAsync(messages, session))
{
    if (update.AsChatResponseUpdate().RawRepresentation is StateSnapshotEvent snapshot)
    {
        JsonElement state = snapshot.Snapshot;
    }
    else if (update.AsChatResponseUpdate().RawRepresentation is StateDeltaEvent delta)
    {
        JsonElement changes = delta.Delta;
    }
}

O cliente é responsável por armazenar e aplicar o estado compartilhado e, em seguida, enviar o estado atual em solicitações subsequentes quando o aplicativo exigir isso.

Próximas Etapas 

Definir modelos de estado

Primeiro, defina modelos Pydantic para sua estrutura de estado. Isso garante a segurança e a validação do tipo:

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

Esquema de estado

Defina um esquema de estado para especificar a estrutura e os tipos de seu estado:

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

Atualizações de estado preditivo

O estado preditivo atualiza os argumentos da ferramenta de fluxo para o estado à medida que a LLM os gera, permitindo atualizações de interface do usuário otimistas:

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

Essa configuração mapeia o recipe campo de estado para o recipe argumento da update_recipe ferramenta. Quando o agente chama a ferramenta, os argumentos são transmitidos para o estado em tempo real à medida que a LLM os gera.

Definir Ferramenta de Atualização de Estado

Crie uma função utilitária que aceite seu modelo 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."

Criar o agente com o Gerenciamento de Estado

Aqui está uma implementação completa do servidor com o gerenciamento de estado:

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

from agent_framework import Agent
from agent_framework.openai import OpenAIChatCompletionClient
from agent_framework_ag_ui import (
    AgentFrameworkAgent,
    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"},
    },
)

# 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)

Conceitos-chave

• Pydantic Models: Defina o estado estruturado com segurança de tipo e validação
• Esquema de Estado: formato simples especificando tipos de campo de estado
• Configuração de Estado Preditivo: mapeia campos de estado para argumentos de ferramenta para atualizações de streaming
• Injeção de Estado: o estado atual é injetado automaticamente como mensagens do sistema para fornecer contexto
• Atualizações completas: as ferramentas devem gravar o estado completo, não apenas deltas
• Estratégia de confirmação: personalizar mensagens de aprovação para seu domínio (receita, documento, planejamento de tarefas etc.)

Noções básicas sobre eventos de estado

Evento de instantâneo de estado

Um instantâneo completo do estado atual, emitido quando a ferramenta conclui a operação:

{
    "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"
            ]
        }
    }
}

Evento Delta do Estado

Atualizações de estado incremental usando o formato de patch JSON, emitido como argumentos da ferramenta de fluxos 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"]
            }
        }
    ]
}

Implementação do cliente

O agent_framework_ag_ui pacote oferece AGUIChatClient para conexão com servidores AG-UI, equiparando a experiência do cliente Python à do .NET.

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

import asyncio
import json
import os
from typing import Any

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(endpoint=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 surfaced through AG-UI events.
                for content in update.contents:
                    if content.type == "data" and getattr(content, "media_type", None) == "application/json":
                        print("\n[JSON state payload received]")

            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 --pre
    asyncio.run(main())

Principais benefícios

O AGUIChatClient fornece:

• Conexão simplificada: tratamento automático da comunicação HTTP/SSE
• Gerenciamento de threads: rastreamento interno de ID de thread para continuidade da conversa
• Integração de agente: funciona perfeitamente com Agent com a API familiar
• Tratamento de Estado: análise automática de eventos de estado do servidor
• Paridade com .NET: experiência consistente entre idiomas

Confirmando o estado previsto

Defina require_confirmation=True em AgentFrameworkAgent quando as alterações de estado previstas precisarem aguardar a confirmação do cliente antes de serem aplicadas:

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

Personalize o texto de confirmação na interface do cliente AG-UI ao renderizar o evento de confirmação.

Interação de exemplo

Com o servidor e o cliente em execução:

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

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

Atualizações Preditivas de Estado em Ação

Ao usar atualizações de estado preditivo com predict_state_config, o cliente recebe STATE_DELTA eventos à medida que o LLM gera argumentos de ferramenta em tempo real, antes que a ferramenta seja executada:

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

Isso permite que o cliente mostre atualizações otimistas da interface do usuário em tempo real à medida que o agente está pensando, fornecendo comentários imediatos aos usuários.

Estado com Human-in-the-Loop

Você pode combinar o gerenciamento de estado com fluxos de trabalho de aprovação definindo 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
)

Quando ativado:

1. Transmissão de atualizações de estado à medida que o agente gera argumentos de ferramenta (atualizações preditivas via eventos STATE_DELTA)
2. O agente faz uma pausa antes de executar a ferramenta com interrupção tool_call em RUN_FINISHED.outcome.interrupts
3. Se aprovada, a ferramenta é executada e o estado final é emitido (via STATE_SNAPSHOT evento)
4. Se rejeitado, as alterações de estado preditivo serão descartadas

Padrões de estado avançados

Estado complexo com vários campos

Você pode gerenciar vários campos de estado com ferramentas diferentes:

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

Usando argumentos de ferramenta curinga

Quando uma ferramenta retorna dados aninhados complexos, use "*" para mapear todos os argumentos da ferramenta para o estado:

@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": "*"}
}

Isso mapeia toda a chamada de ferramenta (todos os argumentos) para o document campo de estado.

Práticas recomendadas

Usar modelos pedantes

Defina modelos estruturados para segurança de tipo:

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

Benefits:

• Segurança do tipo: validação automática de tipos de dados
• Documentação: descrições de campo servem como documentação
• Suporte ao IDE: preenchimento automático e verificação de tipos
• Serialização: conversão JSON automática

Finalizar Atualizações de Estado

Sempre escreva o estado completo, não apenas deltas:

@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."

Isso garante a consistência do estado e as atualizações preditivas adequadas.

Igualar Nomes de Parâmetros

Verifique se os nomes de parâmetro da ferramenta correspondem à tool_argument configuração:

# 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
}

Fornecer contexto em instruções

Inclua instruções claras sobre o gerenciamento de estado:

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

Personalizar a interface de confirmação

Personalize mensagens de aprovação e confirmação de estado em seu cliente AG-UI ao renderizar eventos de confirmação do servidor.

Próximas etapas

Agora você aprendeu todos os principais recursos de AG-UI! Em seguida, você pode:

• Explorar a documentação do Agent Framework
• Criar um aplicativo completo combinando todos os recursos de AG-UI
• Implantar seu serviço de AG-UI em produção

Recursos adicionais

• Visão geral doAG-UI
• Introdução
• Renderização da ferramenta de back-end

O gerenciamento de estado do AG-UI em Go pode ser implementado com middleware que emite atualizações estruturadas message.DataContent junto com atualizações de texto normais.

stateSnapshotMiddleware := agent.MiddlewareFunc(func(next agent.RunFunc, ctx context.Context, messages []*message.Message, opts ...agent.Option) iter.Seq2[*agent.ResponseUpdate, error] {
    return func(yield func(*agent.ResponseUpdate, error) bool) {
        for update, err := range next(ctx, messages, opts...) {
            if err != nil {
                yield(nil, err)
                return
            }
            if update != nil {
                // Inspect update contents and yield DataContent snapshots as needed.
            }
            if !yield(update, nil) {
                return
            }
        }
    }
})

a := foundryprovider.NewAgent(endpoint, token, foundryprovider.ModelDeployment(model), foundryprovider.AgentConfig{
    Config: agent.Config{
        Middlewares: []agent.Middleware{stateSnapshotMiddleware},
    },
})