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 usan reason: "tool_call" e incluyen una 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. Python AgentFrameworkAgent pone a disposición las solicitudes de aprobación y los metadatos de interrupción; no acepta un objeto de estrategia de confirmación en el 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. Resolver la interrupción visible también completa las demás llamadas a herramientas de ese mismo lote según sus decisiones de aprobación. Por ejemplo, se ejecuta un elemento hermano never_require y su TOOL_CALL_RESULT se transmite en flujo en la ejecución reanudada incluso cuando se rechaza el elemento hermano que requiere aprobación.

Cancelar con status: "cancelled" interrumpe la reanudación de la aprobación y elimina el estado de aprobación en cola del hilo. Las solicitudes posteriores no pueden volver a aparecer 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 de AG-UI 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)},
    },
})