Humain-dans-la-boucle avec AG-UI

L’approbation des outils MAF reste chargée de déterminer si un outil nécessite une approbation. AG-UI transporte la demande d’approbation au client et la décision du client vers le serveur.

Pour les stratégies d’approbation, les règles conditionnelles et les consignes générales de sécurité, consultez Utiliser les outils de fonction avec approbations via une intervention humaine.

Exiger une approbation

Encapsulez la fonction MAF avec ApprovalRequiredAIFunction et exposez l’agent normalement :

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

Lorsque le modèle appelle l’outil, l’adaptateur AG-UI termine l’exécution avec une interruption d’appel d’outil au lieu d’exécuter la fonction.

Résoudre l’interruption d’un client .NET

AGUIChatClient expose l’interruption sous la forme de ToolApprovalRequestContent. Créez et envoyez une réponse à l’aide des types d’approbation MAF normaux :

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

Réutilisez le même AgentSession lors de l’envoi de la réponse afin que le client puisse reprendre l’exécution interrompue. Utilisez approved: false pour rejeter l’appel. L’adaptateur convertit la réponse MAF en charge utile de reprise AG-UI canonique.

Étapes suivantes

Ce tutoriel vous montre comment implémenter des workflows avec intervention humaine avec AG-UI, où les utilisateurs doivent approuver la mise en œuvre des outils avant leur mise en œuvre. Cela est essentiel pour les opérations sensibles telles que les transactions financières, les modifications de données ou les actions qui ont des conséquences significatives.

Prerequisites

Avant de commencer, vérifiez que vous avez terminé le didacticiel De rendu de l’outil principal et comprenez :

• Guide pratique pour créer des outils de fonction
• Comment AG-UI transmet les événements liés aux outils
• Configuration du serveur et du client de base

Qu’est-ce que Human-in-the-Loop ?

Human-in-the-Loop (HITL) est un modèle où l’agent demande l’approbation de l’utilisateur avant d’exécuter certaines opérations. Avec AG-UI :

• L’assistant génère des appels de fonction comme d’habitude
• Au lieu d’exécuter immédiatement, le serveur envoie des demandes d’approbation au client
• Le client affiche la demande et invite l’utilisateur
• L’utilisateur approuve ou rejette l’action
• Le serveur reçoit la réponse et se poursuit en conséquence

Benefits

• Sécurité : empêcher l’exécution d’actions involontaires
• Transparence : les utilisateurs voient exactement ce que l’agent veut faire
• Contrôle : les utilisateurs ont le dernier mot sur les opérations sensibles
• Conformité : Répondre aux exigences réglementaires en matière de surveillance humaine

Outils de marquage pour approbation

Pour exiger l’approbation d’un outil, utilisez le approval_mode paramètre dans le @tool décorateur :

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"

Modes d’approbation

• always_require: Toujours demander l’approbation avant l’exécution
• never_require: Ne jamais demander d’approbation (comportement par défaut)
• conditional: Demander une approbation basée sur certaines conditions (logique personnalisée)

Création d’un serveur avec Human-in-the-Loop

Voici une implémentation complète du serveur avec les outils nécessaires à l’approbation :

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

Concepts clés

• AgentFrameworkAgent wrapper : active les fonctionnalités de protocole AG-UI comme human-in-the-loop
• require_confirmation=True: active le flux de travail d’approbation pour les outils marqués
• Contrôle au niveau de l’outil : seuls les outils marqués avec approval_mode="always_require" devront demander l’approbation

Comprendre les interruptions d’approbation

Lorsqu’un outil nécessite une approbation, l’exécution se termine par une interruption de AG-UI canonique.

Interruption de validation

{
  "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’approbation de l’outil interrompt l’utilisation reason: "tool_call" et inclut un toolCallId. Le ChatResponseUpdate final de AGUIChatClient conserve les valeurs de outcome et de interrupts dans additional_properties. Interrupt et ResumeEntry sont des types de protocole issus de ag_ui.core, et non des modèles spécifiques à Agent Framework.

Format de CV

Reprendre le même thread avec un tableau canonique resume . Permet accepted: false de rejeter l’opération tout en permettant à l’agent de continuer. Utilisez status: "cancelled" sans charge utile pour annuler l’exécution interrompue.

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

Client avec fonctionnalité d'approbation

Voici un client AGUIChatClient qui gère les demandes d’approbation :

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

Exemple d’interaction

Avec le serveur et le client en cours d’exécution :

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]

Si l’utilisateur rejette :

Approve this action? (yes/no): no

[Sending approval response: False]

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

Messages de confirmation personnalisés

Personnalisez les messages d’approbation et de confirmation dans votre interface utilisateur du client AG-UI lors du rendu des interruptions d’approbation du serveur. Le SDK Python AgentFrameworkAgent expose les demandes d’approbation et les métadonnées d’interruption ; il n’accepte pas d’objet de stratégie de confirmation côté serveur.

Meilleures pratiques

Effacer les descriptions des outils

Fournissez des descriptions détaillées afin que les utilisateurs comprennent ce qu’ils approuvent :

@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

Approbation granulaire

Demandez l’approbation pour des actions sensibles individuelles plutôt que par lot :

# 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

Arguments informatifs

Utilisez des noms de paramètres descriptifs et fournissez un contexte :

@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

Gestion du délai d’expiration

Définissez les délais d’expiration appropriés pour les demandes d’approbation :

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

Approbation sélective

Vous pouvez combiner des outils qui nécessitent une approbation avec ceux qui ne le font pas :

# 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

Approbations par lots et annulation

Une réponse de modèle peut contenir à la fois des outils et des outils requis par approbation qui ne nécessitent pas d’approbation. Le traitement de l’interruption affichée finalise également les autres appels d’outils de ce lot, selon leurs décisions d’approbation. Par exemple, un élément pair never_require s’exécute et son TOOL_CALL_RESULT est transmis en continu lors de l’exécution reprise, même lorsque l’élément pair nécessitant une approbation est rejeté.

L’annulation avec status: "cancelled" interrompt la reprise de l’approbation et efface l’état d’approbation en file d’attente pour le fil. Les requêtes ultérieures ne peuvent pas faire réapparaître ni exécuter des appels d’outils obsolètes provenant du lot annulé.

Étapes suivantes

Ressources additionnelles

• Vue d’ensemble deAG-UI
• Rendu de l'outil côté serveur
• Outils de fonction avec approbations

Go prend en charge les flux avec intervention humaine d’AG-UI, avec des outils nécessitant une approbation. Encapsulez un outil de fonction avec tool.ApprovalRequiredFunc, puis hébergez l’agent via 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)},
    },
})