Human-in-the-Loop con AG-UI

La aprobación de la herramienta MAF sigue siendo responsable de decidir si una herramienta requiere aprobación. AG-UI transporta la solicitud de aprobación al cliente y la decisión del cliente de vuelta al servidor.

Para consultar las políticas de aprobación, las reglas condicionales y la orientación general de seguridad, consulte Uso de herramientas de función con aprobaciones con intervención humana.

Requerimiento de aprobación

Envuelva la función MAF con ApprovalRequiredAIFunction y exponga el 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);

Cuando el modelo llama a la herramienta, el adaptador de AG-UI finaliza la ejecución con una interrupción de llamada a herramientas en lugar de ejecutar la función.

Resolución de la interrupción desde un cliente de .NET

AGUIChatClient muestra la interrupción como ToolApprovalRequestContent. Cree y envíe una respuesta mediante los tipos de aprobación normales de 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.
    }
}

Vuelva a usar lo mismo AgentSession al enviar la respuesta para que el cliente pueda continuar la ejecución interrumpida. Use approved: false para rechazar la llamada. El adaptador convierte la respuesta de MAF en la carga útil canónica de reanudación de AG-UI.

Pasos siguientes

En este tutorial se muestra cómo implementar flujos de trabajo humanos en bucle con AG-UI, donde los usuarios deben aprobar las ejecuciones de herramientas antes de que se realicen. Esto es esencial para las operaciones confidenciales, como transacciones financieras, modificaciones de datos o acciones que tienen consecuencias significativas.

Prerequisites

Antes de comenzar, asegúrese de que ha completado el tutorial de representación de la herramienta de back-end y comprenda lo siguiente:

• Cómo crear herramientas de función
• Cómo AG-UI transmite eventos de la herramienta
• Configuración básica del servidor y del cliente

¿Qué es Human-in-the-Loop?

Human-in-the-Loop (HITL) es un patrón en el que el agente solicita la aprobación del usuario antes de ejecutar determinadas operaciones. Con AG-UI:

• El agente genera invocaciones a herramientas como de costumbre
• En lugar de ejecutarse inmediatamente, el servidor envía solicitudes de aprobación al cliente.
• El cliente muestra la solicitud y solicita al usuario
• El usuario aprueba o rechaza la acción.
• El servidor recibe la respuesta y continúa en consecuencia.

Benefits

• Seguridad: impedir que se ejecuten acciones no deseadas
• Transparencia: los usuarios ven exactamente lo que el agente quiere hacer
• Control: los usuarios tienen la decisión final sobre las operaciones sensibles
• Cumplimiento: cumplir los requisitos normativos para la supervisión humana

Herramientas de Marcado para Aprobación

Para solicitar la aprobación de una herramienta, use el parámetro approval_mode en el 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 aprobación

• always_require: solicitar siempre la aprobación antes de la ejecución
• never_require: nunca se solicita aprobación (comportamiento predeterminado)
• conditional: solicitud de aprobación basada en determinadas condiciones (lógica personalizada)

Creación de un servidor con human-in-the-loop

Esta es una implementación completa del servidor con herramientas necesarias para aprobación:

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

Conceptos clave

• AgentFrameworkAgent contenedor: habilita características del protocolo AG-UI como human-in-the-loop
• require_confirmation=True: activa el flujo de trabajo de aprobación para las herramientas marcadas.
• Control de nivel de herramienta: solo las herramientas marcadas con approval_mode="always_require" solicitarán aprobación

Comprender las interrupciones de aprobación

Cuando una herramienta requiere aprobación, la ejecución finaliza con una interrupción de AG-UI canónica.

Interrupción de aprobación

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

Las interrupciones de aprobación de herramientas utilizan reason: "tool_call" e incluyen un toolCallId. La versión final de ChatResponseUpdate de AGUIChatClient conserva los valores de outcome y interrupts en additional_properties. Interrupt y ResumeEntry son tipos de protocolo de ag_ui.core, no modelos específicos de Agent Framework.

Formato del currículum

Retome el mismo hilo con una matriz resume canónica. Use accepted: false para rechazar la operación al permitir que el agente continúe. Usa status: "cancelled" sin carga para cancelar la ejecución interrumpida.

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

Cliente con soporte para aprobación

Este es un cliente que usa AGUIChatClient que controla las solicitudes de aprobación:

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

Interacción de ejemplo

Con el servidor y el cliente en ejecución:

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 el usuario rechaza:

Approve this action? (yes/no): no

[Sending approval response: False]

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

Mensajes de confirmación personalizados

Personalice los mensajes de aprobación y confirmación en la interfaz de usuario de su cliente AG-UI al mostrar las interrupciones de aprobación procedentes del servidor. El AgentFrameworkAgent de Python expone las solicitudes de aprobación y los metadatos de la interrupción; no admite un objeto de estrategia de confirmación del lado del servidor.

Prácticas recomendadas

Borrar descripciones de herramientas

Proporcione descripciones detalladas para que los usuarios comprendan lo que aprueban:

@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

Aprobación detallada

Solicite la aprobación de acciones confidenciales individuales en lugar de procesar por lotes:

# 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 nombres de parámetro descriptivos y proporcione 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

Gestión del tiempo de espera

Establezca los tiempos de espera adecuados para las solicitudes de aprobación:

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

Aprobación selectiva

Puede combinar herramientas que requieran aprobación con las que no:

# 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

Aprobaciones y cancelación por lotes

Una respuesta de un modelo puede contener tanto herramientas que requieren aprobación como herramientas que no requieren aprobación. La resolución de la interrupción visible también completa las demás llamadas a herramientas de ese lote de acuerdo con sus decisiones de aprobación. Por ejemplo, un never_require hermano se ejecuta y su TOOL_CALL_RESULT se transmite en la ejecución reanudada incluso cuando el hermano que requiere aprobación es rechazado.

Las llamadas aprobadas se reanudan a través de la directiva normal del Agente y la canalización de middleware; la aprobación no omite la admisión ni la programación. Si la invocación de función está deshabilitada cuando se reanuda una concesión, la concesión permanece pendiente para un reintento explícito después de volver a habilitar la invocación. Los nodos terminales hermanos ya finalizados pueden reproducirse para restaurar el estado del cliente sin volver a ejecutar sus efectos secundarios.

La cancelación con status: "cancelled" interrumpe la reanudación de la aprobación y borra el estado de aprobación en cola para el subproceso. Las solicitudes posteriores no pueden reaparecer ni ejecutar llamadas a herramientas obsoletas desde el lote cancelado.

Pasos siguientes

Recursos adicionales

• Información general deAG-UI
• Representación de herramientas de back-end
• Herramientas de funciones con aprobaciones

Go admite flujos AG-UI con intervención humana (human-in-the-loop) con herramientas que requieren aprobación. Envuelva una herramienta de funciones con tool.ApprovalRequiredFunc, y a continuación aloje el agente a través 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)},
    },
})