Human-in-the-Loop mit AG-UI

Die MAF-Toolgenehmigung bleibt für die Entscheidung verantwortlich, ob ein Tool eine Genehmigung erfordert. AG-UI die Genehmigungsanforderung an den Client und die Entscheidung des Clients zurück zum Server transportiert.

Informationen zu Genehmigungsrichtlinien, Bedingungsregeln und allgemeinen Sicherheitsrichtlinien finden Sie unter Funktionstools mit Genehmigungen mit menschlicher Prüfung verwenden.

Genehmigung fordern

Schließen Sie die MAF-Funktion mit ApprovalRequiredAIFunction und machen Sie den Agent normal verfügbar:

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

Wenn das Modell das Tool aufruft, beendet der AG-UI Adapter die Ausführung mit einem Toolaufrufunterbrechung, anstatt die Funktion auszuführen.

Behandeln eines Interrupts in einem .NET-Client

AGUIChatClient zeigt den Interrupt als ToolApprovalRequestContent. Erstellen und Senden einer Antwort mithilfe der normalen MAF-Genehmigungstypen:

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

Verwenden Sie beim Senden der Antwort dieselbe AgentSession, damit der Client die unterbrochene Ausführung fortsetzen kann. Verwenden Sie approved: false, um den Anruf abzulehnen. Der Adapter wandelt die MAF-Antwort in die kanonische AG-UI-Wiederaufnahme-Payload um.

Nächste Schritte

In dieser Anleitung erfahren Sie, wie Sie prozessbegleitete Workflows mit AG-UI implementieren, in denen Benutzer Werkzeugausführungen genehmigen müssen, bevor sie ausgeführt werden. Dies ist für sensible Vorgänge wie Finanztransaktionen, Datenänderungen oder Aktionen von wesentlicher Bedeutung.

Voraussetzungen

Bevor Sie beginnen, stellen Sie sicher, dass Sie das Lernprogramm zum Rendern von Back-End-Tools abgeschlossen haben, und verstehen Sie Folgendes:

• Wie man Funktionstools erstellt
• So streamen AG-UI Toolereignisse
• Grundlegendes Server- und Clientsetup

Was ist Human-in-the-Loop?

Human-in-the-Loop (HITL) ist ein Muster, bei dem der Agent die Benutzergenehmigung anfordert, bevor bestimmte Vorgänge ausgeführt werden. Mit AG-UI:

• Der Agent generiert Toolaufrufe wie gewohnt.
• Anstatt sofort auszuführen, sendet der Server Genehmigungsanforderungen an den Client.
• Der Client zeigt die Anforderung an und fordert den Benutzer auf.
• Der Benutzer genehmigt oder lehnt die Aktion ab.
• Der Server empfängt die Antwort und fährt entsprechend fort.

Benefits

• Sicherheit: Verhindern, dass unbeabsichtigte Aktionen ausgeführt werden
• Transparenz: Benutzer sehen genau, was der Agent tun möchte
• Kontrolle: Benutzer haben letzte Entscheidung über sensible Vorgänge
• Compliance: Einhaltung gesetzlicher Vorschriften für die menschliche Aufsicht

Markierungstools für die Genehmigung

Um eine Genehmigung für ein Tool zu verlangen, verwenden Sie den approval_mode Parameter im @tool Dekorateur:

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"

Genehmigungsmodi

• always_require: Fordern Sie vor der Ausführung immer die Genehmigung an
• never_require: Keine Genehmigung anfordern (Standardverhalten)
• conditional: Anfordern der Genehmigung basierend auf bestimmten Bedingungen (benutzerdefinierte Logik)

Erstellen eines Servers mit "Human-in-the-Loop"

Hier ist eine vollständige Serverimplementierung mit genehmigungsrelevanten Tools:

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

Wichtige Konzepte

• AgentFrameworkAgent Wrapper: Ermöglicht AG-UI Protokollfeatures wie human-in-the-loop
• require_confirmation=True: Aktiviert den Genehmigungsworkflow für markierte Tools.
• Kontrolle auf Werkzeugebene: Nur Tools, die mit approval_mode="always_require" gekennzeichnet sind, fordern eine Genehmigung an.

Genehmigungsunterbrechungen verstehen

Wenn ein Tool eine Genehmigung erfordert, endet die Ausführung mit einem kanonischen AG-UI Interrupt.

Unterbrechung der Genehmigung

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

Die Toolgenehmigung unterbricht die Verwendung reason: "tool_call" und schließt ein toolCallId. Die endgültige ChatResponseUpdate aus AGUIChatClient behält die Werte für outcome und interrupts in additional_properties bei. Interrupt und ResumeEntry sind Protokolltypen aus ag_ui.core, nicht von Agent Framework-spezifischen Modellen.

Lebenslaufformat

Setzen Sie denselben Thread mit einem kanonischen resume Array fort. Verwenden Sie accepted: false, um den Vorgang abzulehnen, während der Agent fortfahren kann. Verwenden Sie status: "cancelled" ohne Nutzlast, um die unterbrochene Ausführung abzubrechen.

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

Client-Software mit Unterstützung für Genehmigungen

Hier ist ein Client, der Genehmigungsanforderungen mithilfe von AGUIChatClient verarbeitet.

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

Beispielinteraktion

Wenn der Server und der Client ausgeführt werden:

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]

Wenn der Benutzer Folgendes ablehnt:

Approve this action? (yes/no): no

[Sending approval response: False]

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

Benutzerdefinierte Bestätigungsmeldungen

Passen Sie Freigabe- und Bestätigungsmeldungen in der Benutzeroberfläche Ihres AG-UI-Clients an, wenn Freigabeunterbrechungen vom Server gerendert werden. Die Python AgentFrameworkAgent macht Genehmigungsanforderungen verfügbar und unterbricht Metadaten. Es wird kein serverseitiges Bestätigungsstrategieobjekt verwendet.

Bewährte Methoden

Löschen von Toolbeschreibungen

Geben Sie detaillierte Beschreibungen an, damit Benutzer verstehen, was sie genehmigen:

@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

Granulare Genehmigung

Anfordern der Genehmigung einzelner vertraulicher Aktionen anstelle von Batchverarbeitung:

# 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

Informative Argumente

Verwenden Sie beschreibende Parameternamen, und stellen Sie Kontext bereit:

@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

Timeoutbehandlung

Legen Sie geeignete Timeouts für Genehmigungsanforderungen fest:

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

Selektive Genehmigung

Sie können Tools kombinieren, für die eine Genehmigung erforderlich ist, und solche, die keine Genehmigung erfordern.

# 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

Stapelgenehmigungen und Stornierungen

Eine Modellantwort kann sowohl genehmigungsrelevante Tools als auch Tools enthalten, die keine Genehmigung erfordern. Durch das Auflösen des sichtbaren Interrupts werden auch die übrigen Tool-Aufrufe aus diesem Batch entsprechend den jeweiligen Genehmigungsentscheidungen abgeschlossen. Beispielsweise wird ein never_require-Sibling-Element ausgeführt, und dessen TOOL_CALL_RESULT wird im fortgesetzten Durchlauf gestreamt, selbst wenn das genehmigungspflichtige Sibling-Element abgelehnt wird.

Das Abbrechen mit status: "cancelled" bricht die Wiederaufnahme der Genehmigung ab und löscht den in der Warteschlange befindlichen Genehmigungsstatus für den Thread. Spätere Anforderungen können keine veralteten Toolaufrufe aus dem abgebrochenen Batch wieder herstellen oder ausführen.

Nächste Schritte

Zusätzliche Ressourcen

• AG-UI Übersicht
• Backend-Werkzeug-Rendering
• Werkzeuge für Funktionen mit Freigaben

Go unterstützt AG-UI-Human-in-the-Loop-Abläufe mit genehmigungspflichtigen Tools. Umschließen Sie ein Funktionstool mit tool.ApprovalRequiredFunc, und hosten Sie dann den Agent durch 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)},
    },
})