Human-in-the-Loop con AG-UI

Il processo di approvazione degli strumenti MAF continua a essere responsabile di stabilire se uno strumento necessiti di approvazione. AG-UI trasmette la richiesta di approvazione al client e riporta al server la decisione del client.

Per i criteri di approvazione, le regole condizionali e le linee guida generali sulla sicurezza, vedere Usare gli strumenti funzione con approvazioni con supervisione umana.

Richiedi approvazione

Racchiudere la funzione MAF con ApprovalRequiredAIFunction ed esporre l'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 il modello invoca lo strumento, l'adattatore AG-UI termina l'esecuzione con un'interruzione di chiamata dello strumento anziché eseguire la funzione.

Risolvere l'interrupt da un client .NET

AGUIChatClient espone l'interruzione come ToolApprovalRequestContent. Creare e inviare una risposta usando i normali tipi di approvazione 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.
    }
}

Riutilizzare lo stesso AgentSession quando si invia la risposta, in modo che il client possa continuare l'esecuzione interrotta. Usare approved: false per rifiutare la chiamata. L'adattatore converte la risposta MAF nel payload canonico AG-UI resume.

Passaggi successivi

Questa esercitazione illustra come implementare flussi di lavoro con l'intervento umano utilizzando AG-UI, in cui gli utenti devono approvare le esecuzioni degli strumenti prima che possano essere eseguite. Ciò è essenziale per operazioni sensibili come transazioni finanziarie, modifiche ai dati o azioni che hanno conseguenze significative.

Prerequisites

Prima di iniziare, assicurarsi di aver completato l'esercitazione sul rendering dello strumento back-end e di comprendere:

• Come creare strumenti per le funzioni
• Come AG-UI gestisce gli eventi degli strumenti
• Configurazione di base del server e del client

Che cos'è Human-in-the-Loop?

Human-in-the-Loop (HITL) è un modello in cui l'agente richiede l'approvazione dell'utente prima di eseguire determinate operazioni. Con AG-UI:

• L'agente genera chiamate allo strumento come di consueto
• Anziché eseguire immediatamente, il server invia richieste di approvazione al client
• Il client visualizza la richiesta e avvisa l'utente.
• L'utente approva o rifiuta l'azione
• Il server riceve la risposta e procede di conseguenza

Benefits

• Sicurezza: impedire l'esecuzione di azioni indesiderate
• Trasparenza: gli utenti vedono esattamente cosa vuole fare l'agente
• Controllo: gli utenti hanno la parola finale sulle operazioni sensibili
• Conformità: soddisfare i requisiti normativi per la supervisione umana

Strumenti di Contrassegno per Approvazione

Per richiedere l'approvazione per uno strumento, usare il approval_mode parametro nel decoratore @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"

Modalità di approvazione

• always_require: richiedere sempre l'approvazione prima dell'esecuzione
• never_require: non richiedere mai l'approvazione (comportamento predefinito)
• conditional: richiedere l'approvazione in base a determinate condizioni (logica personalizzata)

Creazione di un server con human-in-the-loop

Ecco un'implementazione completa del server con gli strumenti necessari per l'approvazione:

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

Concetti chiave

• AgentFrameworkAgent wrapper: abilita le funzionalità del protocollo AG-UI come human-in-the-loop
• require_confirmation=True: attiva il flusso di lavoro di approvazione per gli strumenti contrassegnati
• Controllo a livello di strumento: solo gli strumenti contrassegnati con approval_mode="always_require" richiederanno l'approvazione

Comprendere le interruzioni dell'approvazione

Quando uno strumento richiede l'approvazione, l'esecuzione termina con un'interruzione AG-UI canonica.

Interruzione dell'approvazione

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

L’approvazione dello strumento interrompe l’uso di reason: "tool_call" e include un toolCallId. L'elemento finale ChatResponseUpdate di AGUIChatClient conserva i valori outcome e interrupts in additional_properties. Interrupt e ResumeEntry sono tipi di protocollo di ag_ui.core, non modelli specifici di Agent Framework.

Formato del curriculum

Riprendi lo stesso thread con un array canonico resume. Usare accepted: false per rifiutare l'operazione, consentendo all'agente di continuare. Usare status: "cancelled" senza un payload per annullare l'esecuzione interrotta.

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

Client con supporto per l'approvazione

Ecco un client che usa AGUIChatClient che gestisce le richieste di approvazione:

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

Interazione di esempio

Con il server e il client in esecuzione:

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 l'utente rifiuta:

Approve this action? (yes/no): no

[Sending approval response: False]

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

Messaggi di conferma personalizzati

Personalizza i messaggi di approvazione e conferma nell'interfaccia utente del client AG-UI durante il rendering delle interruzioni di approvazione provenienti dal server. Python AgentFrameworkAgent espone le richieste di approvazione e i metadati dell’interruzione; non accetta un oggetto per la strategia di conferma sul lato server.

Migliori pratiche

Cancella descrizioni degli strumenti

Fornire descrizioni dettagliate in modo che gli utenti comprendano cosa stanno approvando:

@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

Approvazione granulare

Richiedere l'approvazione per le singole azioni sensibili anziché l'invio in batch:

# 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

Argomenti informativi

Usare nomi di parametri descrittivi e fornire contesto:

@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

Gestione del timeout

Impostare i timeout appropriati per le richieste di approvazione:

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

Approvazione selettiva

È possibile combinare strumenti che richiedono l'approvazione con quelli che non:

# 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

Approvazioni e annullamenti in batch

Una risposta del modello può contenere sia strumenti che richiedono approvazione sia strumenti che non la richiedono. La risoluzione dell'interruzione visibile completa anche le altre invocazioni degli strumenti di quel batch, in base alle relative decisioni di approvazione. Ad esempio, un elemento never_require dello stesso livello viene eseguito e il relativo TOOL_CALL_RESULT viene trasmesso in streaming durante l'esecuzione ripresa anche quando l'elemento dello stesso livello che richiede approvazione viene rifiutato.

L'annullamento con status: "cancelled" interrompe la ripresa dell'approvazione e cancella lo stato di approvazione in coda per il thread. Le richieste successive non possono far riemergere né eseguire chiamate agli strumenti obsolete dal batch annullato.

Passaggi successivi

Risorse aggiuntive

• Panoramica di AG-UI
• Rendering degli strumenti back-end
• Strumenti per le funzioni con approvazioni

Go supporta workflow AG-UI human-in-the-loop con strumenti che richiedono l'approvazione. Racchiudi uno strumento funzione con tool.ApprovalRequiredFunc, quindi ospita l'agente tramite 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)},
    },
})