Gestão do Estado com AG-UI

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

• Leia o estado fornecido pelo cliente a partir do original RunAgentInput.
• Mapear chamadas ou resultados de ferramentas selecionadas para eventos de estado da AG-UI com AGUIStreamOptions.

O mapeamento de estados é opcional. Resultados arbitrários de ferramentas não se tornam automaticamente estado partilhado.

Leia o estado do cliente

MapAGUIServer armazena a origem RunAgentInput em ChatOptions. Um agente de delegação ou middleware 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 é entrada do pedido. Valide a sua forma e valores antes de o usar em prompts, encaminhamentos ou operações privilegiadas.

Emitir um instantâneo de estado

Mapeie um resultado da 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 na JsonElement ferramenta antes de o devolver. O resultado de generate_recipe torna-se então o snapshot e substitui o estado partilhado atual do cliente.

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

Deltas do estado de emissão

Associe o resultado de uma ferramenta a STATE_DELTA quando esta devolve um JSON Patch RFC 6902:

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

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

Use um snapshot para inicializar ou substituir estados e deltas para alterações incrementais.

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

Chamadas de ferramenta de mapas para o estado

AGUIStreamOptions.MapCall mapeia um FunctionCallContent selecionado para eventos adicionais de AG-UI emitidos após os eventos normais de chamada da ferramenta. Use-o quando o estado deriva dos argumentos da 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);

A aplicação detém o mapeamento e a forma do estado. MapCall não infere estado a partir de argumentos arbitrários de ferramentas nem suprime a execução normal da ferramenta. As atualizações incrementais exigem que o cliente do modelo subjacente exponha argumentos de chamada de ferramenta transmitidos e que a aplicação configure a extração correspondente de argumentos.

Estado de receção num cliente .NET

O cliente AG-UI .NET expõe eventos do protocolo de estado através 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 reter e aplicar o estado partilhado, enviando depois o estado atual em pedidos posteriores quando a aplicação assim o exigir.

Passos seguintes

Modelos de Estados

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

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 do seu estado:

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

Atualizações de estado preditivas

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

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

Esta 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 o modelo de linguagem de máquina os gera.

Definir ferramenta de atualização de estado

Crie uma função de ferramenta que aceite o 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 agente com gestão de estado

Aqui está uma implementação completa do servidor com 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

• Modelos Pydantic: Definir 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: Personalize as mensagens de aprovação para o seu domínio (receita, documento, planeamento 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 é finalizada:

{
    "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 incrementais usando o formato JSON Patch, emitidas como os argumentos da ferramenta LLM streams:

{
    "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 fornece AGUIChatClient a conexão com servidores AG-UI, trazendo a experiência do cliente Python para a paridade com o .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 dispõe:

• Conexão simplificada: manipulação automática de comunicação HTTP/SSE
• Gerenciamento de threads: rastreamento de ID de thread integrado para continuidade de conversa
• Integração de agentes: funciona perfeitamente com Agent uma API familiar
• State Handling: Análise automática de eventos de estado do servidor
• Paridade com .NET: experiência consistente entre idiomas

Confirmação do estado previsto

Defina require_confirmation=True em AgentFrameworkAgent quando as mudanças de estado previstas tiverem de 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,
)

Personaliza a cópia de confirmação na interface do cliente AG-UI ao renderizar o evento de confirmação.

Exemplo de interação

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 de estado preditivas em ação

Ao usar atualizações de estado preditivas com predict_state_config, o cliente recebe STATE_DELTA eventos à medida que o LLM gera argumentos da ferramenta em tempo real, antes da execução da ferramenta.

// 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 conforme o agente está pensando, fornecendo feedback imediato 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. Fluxo de atualizações de estado à medida que o agente gera argumentos de ferramenta (atualizações preditivas por meio de STATE_DELTA eventos)
2. O agente faz uma pausa antes de executar a ferramenta com uma 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 rejeitadas, as alterações de estado preditivo são descartadas

Padrões de estado avançados

Estado complexo com vários campos

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

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 da ferramenta curinga

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

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

Melhores práticas

Usar modelos pidânticos

Definir 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]

Benefícios:

• Segurança do tipo: Validação automática de tipos de dados
• Documentação: As descrições dos campos servem como documentação
• Suporte IDE: Preenchimento automático e verificação de tipo
• Serialização: conversão automática de JSON

Atualizações de estado completas

Escreva sempre 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 atualizações preditivas adequadas.

Corresponder nomes de parâmetros

Verifique se os nomes dos parâmetros 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 nas instruções

Inclua instruções claras sobre a gestão do 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 interface de confirmação

Personalize mensagens de aprovação e confirmação de estado no seu cliente AG-UI ao apresentar eventos de confirmação a partir do servidor.

Próximas Etapas

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

• Explore a documentação do Agent Framework
• Crie um aplicativo completo combinando todos os recursos AG-UI
• Implante seu serviço de AG-UI na produção

Recursos adicionais

• Visão geral do AG-UI
• Começar
• Renderização de ferramentas de back-end

No Go AG-UI, a gestão do estado pode ser implementada com middleware que emite atualizações estruturadas de message.DataContent a par das atualizações normais de texto.

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