Human-in-the-Loop com AG-UI

A aprovação da ferramenta MAF permanece responsável por decidir se uma ferramenta requer aprovação. AG-UI transporta a solicitação de aprovação para o cliente e a decisão do cliente de volta para o servidor.

Para políticas de aprovação, regras condicionais e diretrizes gerais de segurança, consulte Usar ferramentas de função com aprovações humanas no loop.

Exigir aprovação

Envolva a função MAF com ApprovalRequiredAIFunction e exponha o agente normalmente:

using Microsoft.Agents.AI;
using Microsoft.Extensions.AI;

AIFunction deleteFile = AIFunctionFactory.Create(
    (string path) => $"Deleted {path}",
    name: "delete_file",
    description: "Delete a file.");

AITool approvalRequiredTool = new ApprovalRequiredAIFunction(deleteFile);
AIAgent agent = chatClient.AsAIAgent(tools: [approvalRequiredTool]);

app.MapAGUIServer("/", agent);

Quando o modelo chama a ferramenta, o adaptador AG-UI termina a execução com uma interrupção de chamada de ferramenta em vez de executar a função.

Resolver a interrupção de um cliente .NET

AGUIChatClient exibe a interrupção como ToolApprovalRequestContent. Crie e envie uma resposta usando os tipos normais de aprovação do MAF:

ToolApprovalRequestContent? request = null;

await foreach (AgentResponseUpdate update in
    remoteAgent.RunStreamingAsync(messages, session))
{
    request ??= update.Contents
        .OfType<ToolApprovalRequestContent>()
        .FirstOrDefault();
}

if (request is not null)
{
    ToolApprovalResponseContent response = request.CreateResponse(approved: true);
    ChatMessage resume = new(ChatRole.User, [response]);

    await foreach (AgentResponseUpdate update in
        remoteAgent.RunStreamingAsync([resume], session))
    {
        // Process the resumed response.
    }
}

Reutilize o mesmo AgentSession ao enviar a resposta para que o cliente possa continuar a execução interrompida. Use approved: false para rejeitar a chamada. O adaptador converte a resposta MAF para o payload canônico de retomada do AG-UI.

Próximas Etapas 

Este tutorial mostra como implementar fluxos de trabalho humanos no loop com a interface do usuário do AG, em que os usuários devem aprovar execuções de ferramentas antes de serem executadas. Isso é essencial para operações confidenciais, como transações financeiras, modificações de dados ou ações que têm consequências significativas.

Prerequisites

Antes de começar, verifique se você concluiu o tutorial de Renderização da Ferramenta de Back-end e entenda:

• Como criar ferramentas de função
• Como AG-UI transmite eventos da ferramenta
• Configuração básica do servidor e do cliente

O que é Human-in-the-Loop?

HITL (Human-in-the-Loop) é um padrão em que o agente solicita a aprovação do usuário antes de executar determinadas operações. Com AG-UI:

• O agente gera chamadas de ferramenta como de costume
• Em vez de executar imediatamente, o servidor envia solicitações de aprovação para o cliente
• O cliente exibe a solicitação e solicita uma ação do usuário
• O usuário aprova ou rejeita a ação
• O servidor recebe a resposta e continua de acordo

Benefits

• Segurança: impedir que ações não intencionais sejam executadas
• Transparência: os usuários veem exatamente o que o agente quer fazer
• Controle: os usuários têm a palavra final sobre operações confidenciais
• Conformidade: atender aos requisitos regulatórios para supervisão humana

Ferramentas de marcação para aprovação

Para requerer aprovação de uma ferramenta, use o parâmetro approval_mode no decorador @tool:

from agent_framework import tool
from typing import Annotated
from pydantic import Field


@tool(approval_mode="always_require")
def send_email(
    to: Annotated[str, Field(description="Email recipient address")],
    subject: Annotated[str, Field(description="Email subject line")],
    body: Annotated[str, Field(description="Email body content")],
) -> str:
    """Send an email to the specified recipient."""
    # Send email logic here
    return f"Email sent to {to} with subject '{subject}'"


@tool(approval_mode="always_require")
def delete_file(
    filepath: Annotated[str, Field(description="Path to the file to delete")],
) -> str:
    """Delete a file from the filesystem."""
    # Delete file logic here
    return f"File {filepath} has been deleted"

Modos de aprovação

• always_require: sempre solicite aprovação antes da execução
• never_require: nunca solicite aprovação (comportamento padrão)
• conditional: solicitar aprovação com base em determinadas condições (lógica personalizada)

Criando um servidor com Human-in-the-Loop

Aqui está uma implementação completa do servidor com ferramentas necessárias para aprovação:

"""AG-UI server with human-in-the-loop."""

import os
from typing import Annotated

from agent_framework import Agent, tool
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
from pydantic import Field


# Tools that require approval
@tool(approval_mode="always_require")
def transfer_money(
    from_account: Annotated[str, Field(description="Source account number")],
    to_account: Annotated[str, Field(description="Destination account number")],
    amount: Annotated[float, Field(description="Amount to transfer")],
    currency: Annotated[str, Field(description="Currency code")] = "USD",
) -> str:
    """Transfer money between accounts."""
    return f"Transferred {amount} {currency} from {from_account} to {to_account}"


@tool(approval_mode="always_require")
def cancel_subscription(
    subscription_id: Annotated[str, Field(description="Subscription identifier")],
) -> str:
    """Cancel a subscription."""
    return f"Subscription {subscription_id} has been cancelled"


# Regular tools (no approval required)
@tool
def check_balance(
    account: Annotated[str, Field(description="Account number")],
) -> str:
    """Check account balance."""
    # Simulated balance check
    return f"Account {account} balance: $5,432.10 USD"


# Read required configuration
endpoint = os.environ.get("AZURE_OPENAI_ENDPOINT")
deployment_name = os.environ.get("AZURE_OPENAI_CHAT_COMPLETION_MODEL")

if not endpoint:
    raise ValueError("AZURE_OPENAI_ENDPOINT environment variable is required")
if not deployment_name:
    raise ValueError("AZURE_OPENAI_CHAT_COMPLETION_MODEL environment variable is required")

chat_client = OpenAIChatCompletionClient(
    model=deployment_name,
    azure_endpoint=endpoint,
    api_version=os.getenv("AZURE_OPENAI_API_VERSION"),
    credential=AzureCliCredential(),
)

# Create agent with tools
agent = Agent(
    name="BankingAssistant",
    instructions="You are a banking assistant. Help users with their banking needs. Always confirm details before performing transfers.",
    client=chat_client,
    tools=[transfer_money, cancel_subscription, check_balance],
)

# Wrap agent to enable human-in-the-loop
wrapped_agent = AgentFrameworkAgent(
    agent=agent,
    require_confirmation=True,  # Enable human-in-the-loop
)

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

if __name__ == "__main__":
    import uvicorn

    uvicorn.run(app, host="127.0.0.1", port=8888)

Conceitos-chave

• AgentFrameworkAgent wrapper: habilita recursos de protocolo AG-UI, como human-in-the-loop
• require_confirmation=True: ativa o fluxo de trabalho de aprovação para ferramentas marcadas
• Controle no nível da ferramenta: Somente as ferramentas marcadas com approval_mode="always_require" solicitarão a aprovação

Noções básicas sobre interrupções de aprovação

Quando uma ferramenta requer aprovação, a execução termina com uma interrupção canônica AG-UI.

Interrupção de aprovação

{
  "type": "RUN_FINISHED",
  "threadId": "thread-1",
  "runId": "run-1",
  "outcome": {
    "type": "interrupt",
    "interrupts": [
      {
        "id": "approval-1",
        "reason": "tool_call",
        "message": "Approve tool call transfer_money?",
        "toolCallId": "call-1",
        "responseSchema": {
          "type": "object",
          "properties": {
            "accepted": { "type": "boolean" },
            "arguments": { "type": "object" }
          },
          "required": ["accepted"]
        },
        "metadata": {
          "agent_framework": {
            "type": "function_approval_request",
            "function_call": {
              "call_id": "call-1",
              "name": "transfer_money",
              "arguments": {
                "from_account": "1234567890",
                "to_account": "0987654321",
                "amount": 500.00,
                "currency": "USD"
              }
            }
          }
        }
      }
    ]
  }
}

As interrupções de aprovação da ferramenta usam reason: "tool_call" e incluem um toolCallId. O ChatResponseUpdate final de AGUIChatClient preserva os valores de outcome e interrupts em additional_properties. Interrupt e ResumeEntry são tipos de protocolo de ag_ui.core, não modelos específicos do Agent Framework.

Formato do currículo

Retome o mesmo thread com uma matriz canônica resume . Use accepted: false para rejeitar a operação, permitindo que o agente continue. Use status: "cancelled" sem uma carga para cancelar a execução interrompida.

{
  "threadId": "thread-1",
  "messages": [],
  "resume": [
    {
      "interruptId": "approval-1",
      "status": "resolved",
      "payload": {
        "accepted": true
      }
    }
  ]
}

Cliente com suporte de aprovação

Aqui está um cliente usando AGUIChatClient que lida com solicitações de aprovação:

"""AG-UI client with human-in-the-loop support."""

import asyncio
import os

from agent_framework import Agent
from agent_framework_ag_ui import AGUIChatClient


def display_approval_request(update) -> None:
    """Display approval request details to the user."""
    print("\n\033[93m" + "=" * 60 + "\033[0m")
    print("\033[93mAPPROVAL REQUIRED\033[0m")
    print("\033[93m" + "=" * 60 + "\033[0m")

    # Display tool call details from update contents
    for i, content in enumerate(update.contents, 1):
        if content.type == "function_approval_request":
            function_call = content.function_call
            print(f"\nAction {i}:")
            print(f"  Tool: \033[95m{function_call.name}\033[0m")
            print(f"  Arguments:")
            for key, value in (function_call.arguments or {}).items():
                print(f"    {key}: {value}")

    print("\n\033[93m" + "=" * 60 + "\033[0m")


async def main():
    """Main client loop with approval handling."""
    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)

    # Create agent with the chat client
    agent = Agent(
        name="ClientAgent",
        client=chat_client,
        instructions="You are a helpful assistant.",
    )

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

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

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

            print("\nAssistant: ", end="", flush=True)
            pending_interrupts = []

            async for update in agent.run(message, session=thread, stream=True):
                # Check if this update carries an approval request.
                if any(content.type == "function_approval_request" for content in update.contents):
                    display_approval_request(update)

                if update.text:
                    print(f"\033[96m{update.text}\033[0m", end="", flush=True)

                properties = update.additional_properties or {}
                outcome = properties.get("outcome")
                if isinstance(outcome, dict) and outcome.get("type") == "interrupt":
                    pending_interrupts = outcome.get("interrupts", [])

            if pending_interrupts:
                resume_entries = []
                for interrupt in pending_interrupts:
                    prompt = interrupt.get("message", "Approve this action?")
                    user_choice = input(f"\n{prompt} (yes/no): ").strip().lower()
                    resume_entries.append({
                        "interruptId": interrupt["id"],
                        "status": "resolved",
                        "payload": {"accepted": user_choice in ("yes", "y")},
                    })

                print("\nAssistant: ", end="", flush=True)
                async for update in agent.run(
                    [],
                    session=thread,
                    stream=True,
                    options={
                        "available_interrupts": pending_interrupts,
                        "resume": resume_entries,
                    },
                ):
                    if update.text:
                        print(f"\033[96m{update.text}\033[0m", end="", flush=True)

            print()

    except KeyboardInterrupt:
        print("\n\nExiting...")
    except Exception as e:
        print(f"\n\033[91mError: {e}\033[0m")


if __name__ == "__main__":
    asyncio.run(main())

Interação de exemplo

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

User (:q or quit to exit): Transfer $500 from account 1234567890 to account 0987654321

[Run Started]
============================================================
APPROVAL REQUIRED
============================================================

Action 1:
  Tool: transfer_money
  Arguments:
    from_account: 1234567890
    to_account: 0987654321
    amount: 500.0
    currency: USD

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

Approve this action? (yes/no): yes

[Sending approval response: True]

[Tool Result: Transferred 500.0 USD from 1234567890 to 0987654321]
The transfer of $500 from account 1234567890 to account 0987654321 has been completed successfully.
[Run Finished]

Se o usuário rejeitar:

Approve this action? (yes/no): no

[Sending approval response: False]

I understand. The transfer has been cancelled and no money was moved.
[Run Finished]

Mensagens de confirmação personalizadas

Personalize as mensagens de aprovação e confirmação na interface do usuário do seu cliente AG-UI ao exibir interrupções de aprovação enviadas pelo servidor. A biblioteca Python AgentFrameworkAgent expõe solicitações de aprovação e metadados de interrupção; ela não aceita um objeto de estratégia de confirmação no servidor.

Práticas recomendadas

Limpar descrições de ferramentas

Forneça descrições detalhadas para que os usuários entendam o que estão aprovando:

@tool(approval_mode="always_require")
def delete_database(
    database_name: Annotated[str, Field(description="Name of the database to permanently delete")],
) -> str:
    """
    Permanently delete a database and all its contents.

    WARNING: This action cannot be undone. All data in the database will be lost.
    Use with extreme caution.
    """
    # Implementation
    pass

Aprovação granular

Solicite aprovação para ações confidenciais individuais em vez de envio em lote:

# Good: Individual approval per transfer
@tool(approval_mode="always_require")
def transfer_money(...): pass

# Avoid: Batching multiple sensitive operations
# Users should approve each operation separately

Argumentos informativos

Use nomes de parâmetro descritivos e forneça contexto:

@tool(approval_mode="always_require")
def purchase_item(
    item_name: Annotated[str, Field(description="Name of the item to purchase")],
    quantity: Annotated[int, Field(description="Number of items to purchase")],
    price_per_item: Annotated[float, Field(description="Price per item in USD")],
    total_cost: Annotated[float, Field(description="Total cost including tax and shipping")],
) -> str:
    """Purchase items from the store."""
    pass

Gerenciamento de tempo limite

Defina os tempos limite apropriados para solicitações de aprovação:

# Client side
async with httpx.AsyncClient(timeout=120.0) as client:  # 2 minutes for user to respond
    # Handle approval
    pass

Aprovação seletiva

Você pode misturar ferramentas que exigem aprovação com aquelas que não:

# No approval needed for read-only operations
@tool
def get_account_balance(...): pass

@tool
def list_transactions(...): pass

# Approval required for write operations
@tool(approval_mode="always_require")
def transfer_funds(...): pass

@tool(approval_mode="always_require")
def close_account(...): pass

Aprovações em lote e cancelamento

Uma resposta do modelo pode conter tanto ferramentas que exigem aprovação quanto ferramentas que não exigem aprovação. Ao solucionar a interrupção visível, as outras chamadas de ferramenta desse lote também são concluídas de acordo com as respectivas decisões de aprovação. Por exemplo, um ramo irmão never_require é executado, e seu TOOL_CALL_RESULT é transmitido na execução retomada mesmo quando o ramo irmão que exige aprovação é rejeitado.

Cancelar com status: "cancelled" interrompe a retomada da aprovação e limpa o estado de aprovação enfileirado da thread. Solicitações posteriores não podem ressurgir ou executar chamadas de ferramenta obsoletas do lote cancelado.

Próximas Etapas 

Recursos adicionais

• Visão geral doAG-UI
• Renderização da ferramenta de back-end
• Ferramentas de Funcionalidade com Aprovações

O Go oferece suporte a fluxos AG-UI com intervenção humana, com ferramentas que exigem aprovação. Encapsule uma ferramenta de função com tool.ApprovalRequiredFunc e, em seguida, hospede o agente por meio de aguiprovider.

approveExpense := functool.MustNew(functool.Config{
    Name:        "approve_expense_report",
    Description: "Approve the expense report.",
}, func(ctx context.Context, expenseReportID string) (string, error) {
    return fmt.Sprintf("Expense report %s approved", expenseReportID), nil
})

a := foundryprovider.NewAgent(endpoint, token, foundryprovider.ModelDeployment(model), foundryprovider.AgentConfig{
    Config: agent.Config{
        Tools: []tool.Tool{tool.ApprovalRequiredFunc(approveExpense)},
    },
})