Human-in-the-Loop met AG-UI

De goedkeuring van MAF-tools blijft verantwoordelijk voor de beslissing of een tool goedkeuring vereist. AG-UI transporteert de goedkeuringsaanvraag naar de client en de beslissing van de client terug naar de server.

Zie Functiehulpmiddelen gebruiken met human-in-the-loop-goedkeuringen voor goedkeuringsbeleidsregels, conditionele regels en algemene veiligheidsrichtlijnen.

Goedkeuring vereist

Omsluit de MAF-functie met ApprovalRequiredAIFunction en stel de agent op de normale manier beschikbaar:

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

Wanneer het model de tool aanroept, rondt de AG-UI-adapter de uitvoering af met een tool-call-interrupt in plaats van de functie uit te voeren.

De onderbreking van een .NET-client oplossen

AGUIChatClient geeft de interrupt weer als ToolApprovalRequestContent. Een antwoord maken en verzenden met de normale MAF-goedkeuringstypen:

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

Gebruik hetzelfde AgentSession wanneer u het antwoord verzendt, zodat de client de onderbroken uitvoering kan voortzetten. Gebruik approved: false dit om de aanroep te weigeren. De adapter zet het MAF-antwoord om naar de standaard AG-UI resume-payload.

Volgende stappen 

In deze zelfstudie leert u hoe u human-in-the-loop-werkstromen implementeert met AG-UI, waar gebruikers hulpprogramma-uitvoeringen moeten goedkeuren voordat ze worden uitgevoerd. Dit is essentieel voor gevoelige bewerkingen, zoals financiële transacties, gegevenswijzigingen of acties die aanzienlijke gevolgen hebben.

Prerequisites

Voordat u begint, moet u ervoor zorgen dat u de handleiding Backend Tool Rendering hebt voltooid en het volgende begrijpt:

• Functiehulpprogramma's maken
• Hoe AG-UI tool-gebeurtenissen streamt
• Eenvoudige server- en clientinstallatie

Wat is Human-in-the-Loop?

HITL (Human-In-The-Loop) is een patroon waarbij de agent goedkeuring van de gebruiker aanvraagt voordat bepaalde bewerkingen worden uitgevoerd. Met AG-UI:

• De agent genereert hulpprogramma-aanroepen zoals gebruikelijk
• In plaats van onmiddellijk uit te voeren, verzendt de server goedkeuringsaanvragen naar de client
• De cliënt geeft het verzoek weer en verzoekt de gebruiker
• De gebruiker keurt de actie goed of weigert
• De server ontvangt de reactie en handelt overeenkomstig.

Benefits

• Veiligheid: Voorkomen dat onbedoelde acties worden uitgevoerd
• Transparantie: Gebruikers zien precies wat de agent wil doen
• Controle: gebruikers hebben laatste uitspraak over gevoelige bewerkingen
• Naleving: Voldoen aan wettelijke vereisten voor menselijk toezicht

Hulpmiddelen voor het markeren ten behoeve van goedkeuring

Als u goedkeuring voor een hulpprogramma wilt vereisen, gebruikt u de approval_mode parameter in de @tool decorator:

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"

Goedkeuringsmodi

• always_require: Altijd goedkeuring aanvragen vóór uitvoering
• never_require: Nooit goedkeuring aanvragen (standaardgedrag)
• conditional: Goedkeuring aanvragen op basis van bepaalde voorwaarden (aangepaste logica)

Een server maken met Human-in-the-Loop

Hier volgt een volledige server-implementatie met door goedkeuring vereiste hulpprogramma's:

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

Belangrijkste concepten

• AgentFrameworkAgent wrapper: maakt AG-UI protocolfuncties mogelijk, zoals human-in-the-loop
• require_confirmation=True: Activeert goedkeuringswerkstroom voor gemarkeerde hulpprogramma's
• Besturingselement op hulpprogrammaniveau: alleen hulpprogramma's die zijn gemarkeerd met approval_mode="always_require" , vragen goedkeuring aan

Inzicht in goedkeuringsonderbrekingen

Wanneer een hulpprogramma goedkeuring vereist, wordt de uitvoering voltooid met een canonieke AG-UI interrupt.

Goedkeuringsonderbreking

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

De goedkeuring van hulpprogramma's onderbreekt het gebruik reason: "tool_call" en omvat een toolCallId. De laatste ChatResponseUpdate van AGUIChatClient behoudt de outcome en interrupts waarden in additional_properties. Interrupt en ResumeEntry zijn protocoltypen van ag_ui.core, niet agentframeworkspecifieke modellen.

CV-indeling

Dezelfde thread hervatten met een canonieke resume matrix. Gebruik accepted: false om de bewerking af te wijzen terwijl de agent kan doorgaan. Gebruik status: "cancelled" zonder payload om de onderbroken run te annuleren.

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

Client met goedkeuringsondersteuning

Hier volgt een client die goedkeuringsaanvragen AGUIChatClient verwerkt:

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

Voorbeeldinteractie

Met de server en client in werking:

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]

Als de gebruiker weigert:

Approve this action? (yes/no): no

[Sending approval response: False]

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

Aangepaste bevestigingsberichten

Pas goedkeurings- en bevestigingsberichten aan in de gebruikersinterface van je AG-UI-client bij het weergeven van goedkeuringsinterrupts van de server. De Python AgentFrameworkAgent stelt goedkeuringsaanvragen en metadata over onderbrekingen beschikbaar; deze accepteert geen bevestigingsstrategieobject aan serverzijde.

Beste praktijken

Duidelijke beschrijvingen van tools

Geef gedetailleerde beschrijvingen zodat gebruikers begrijpen wat ze goedkeuren:

@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

Fijnmazige goedkeuring

Goedkeuring aanvragen voor afzonderlijke gevoelige acties in plaats van batchverwerking:

# 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

Informatieve argumenten

Gebruik beschrijvende parameternamen en geef context op:

@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

Afhandeling van time-outs

Stel de juiste time-outs in voor goedkeuringsaanvragen:

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

Selectieve goedkeuring

U kunt hulpprogramma's combineren waarvoor goedkeuring is vereist met die waarvoor dat niet het geval is.

# 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

Batchgewijze goedkeuringen en annulering

Eén modelantwoord kan zowel hulpprogramma's bevatten die vereist zijn voor goedkeuring als hulpprogramma's waarvoor geen goedkeuring is vereist. Door de zichtbare onderbreking op te lossen, worden ook de andere toolaanroepen uit die batch afgerond volgens de bijbehorende goedkeuringsbeslissingen. Bijvoorbeeld: een never_require naastliggende tak wordt uitgevoerd en de TOOL_CALL_RESULT ervan wordt in de hervatte uitvoering gestreamd, zelfs wanneer de naastliggende tak waarvoor goedkeuring is vereist, wordt afgewezen.

Als u annuleert met status: "cancelled" het afbreken van het goedkeurings hervat en wordt de goedkeuringsstatus in de wachtrij voor de thread gewist. Latere verzoeken kunnen verouderde toolaanroepen uit de geannuleerde batch niet opnieuw oproepen of uitvoeren.

Volgende stappen 

Aanvullende bronnen

• overzicht vanAG-UI
• Rendering van back-endhulpprogramma
• Functiehulpprogramma's met goedkeuringen

Go ondersteunt AG-UI-human-in-the-loop-workflows met tools waarvoor goedkeuring is vereist. Verpak een functiehulpmiddel met tool.ApprovalRequiredFunc, en host de agent vervolgens 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)},
    },
})