Human-in-the-Loop z AG-UI

Zatwierdzenie narzędzia MAF pozostaje odpowiedzialne za podjęcie decyzji, czy narzędzie wymaga zatwierdzenia. AG-UI transportuje żądanie zatwierdzenia do klienta i decyzję klienta z powrotem do serwera.

Aby zapoznać się z zasadami zatwierdzania, regułami warunkowymi i ogólnymi wskazówkami dotyczącymi bezpieczeństwa, zobacz Korzystanie z narzędzi funkcji z zatwierdzeniami z udziałem człowieka.

Wymagaj zatwierdzenia

Opakuj funkcję MAF znacznikiem ApprovalRequiredAIFunction i udostępnij agenta w zwykły sposób:

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

Gdy model wywołuje narzędzie, adapter AG-UI kończy uruchomienie przerwaniem wywołania narzędzia zamiast wykonać funkcję.

Rozwiązywanie przerwania działania klienta .NET

AGUIChatClient przedstawia przerwanie jako ToolApprovalRequestContent. Utwórz i wyślij odpowiedź przy użyciu normalnych typów zatwierdzeń 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.
    }
}

Użyj ponownie tego samego AgentSession podczas wysyłania odpowiedzi, aby klient mógł kontynuować przerwane uruchomienie. Użyj polecenia approved: false , aby odrzucić połączenie. Adapter przekształca odpowiedź MAF w kanoniczny ładunek danych wznowienia AG-UI.

Następne kroki

W tym samouczku pokazano, jak zaimplementować procesy robocze z udziałem człowieka za pomocą AG-UI, gdzie użytkownicy muszą zatwierdzać uruchamianie narzędzi przed ich wykonaniem. Jest to niezbędne w przypadku operacji poufnych, takich jak transakcje finansowe, modyfikacje danych lub działania, które mają poważne konsekwencje.

Wymagania wstępne

Przed rozpoczęciem upewnij się, że ukończono samouczek renderowania narzędzi backendowych i zapoznaj się z tematem:

• Jak tworzyć narzędzia funkcji
• Jak AG-UI przesyła strumieniowo wydarzenia narzędzi
• Konfiguracja serwera podstawowego i klienta

Co to jest human-in-the-loop?

Human-in-the-Loop (HITL) to wzorzec, w którym agent żąda zatwierdzenia przez użytkownika przed wykonaniem określonych operacji. Z AG-UI:

• Agent generuje wywołania narzędzi w zwykły sposób
• Zamiast wykonywać natychmiast, serwer wysyła żądania zatwierdzenia do klienta
• Klient wyświetla żądanie i monituje użytkownika
• Użytkownik zatwierdza lub odrzuca akcję
• Serwer otrzymuje odpowiedź i postępuje odpowiednio

Benefits

• Bezpieczeństwo: uniemożliwia wykonywanie niezamierzonych akcji
• Przezroczystość: użytkownicy widzą dokładnie to, co agent chce zrobić
• Kontrola: Użytkownicy mają ostateczne informacje o operacjach poufnych
• Zgodność: spełnianie wymagań regulacyjnych dotyczących nadzoru ludzkiego

Wyznaczanie narzędzi zatwierdzania

Aby wymagać zatwierdzenia dla narzędzia, użyj parametru approval_mode w dekoratorze @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"

Tryby zatwierdzania

• always_require: Zawsze żądaj zatwierdzenia przed wykonaniem
• never_require: Nigdy nie żądaj zatwierdzenia (zachowanie domyślne)
• conditional: Żądanie zatwierdzenia na podstawie określonych warunków (logika niestandardowa)

Konfiguracja serwera z udziałem człowieka

Oto kompletna implementacja serwera z narzędziami wymaganymi do zatwierdzenia:

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

Kluczowe pojęcia

• AgentFrameworkAgent wrapper: włącza funkcje protokołu AG-UI, takie jak człowiek-w-pętli
• require_confirmation=True: Aktywuje przepływ pracy zatwierdzania dla oznaczonych narzędzi
• Kontrola na poziomie narzędzia: tylko narzędzia oznaczone za pomocą approval_mode="always_require" będą żądać zatwierdzenia

Zrozumienie przerwań zatwierdzania

Gdy narzędzie wymaga zatwierdzenia, uruchomienie kończy się standardowym przerwaniem AG-UI.

Przerwanie zatwierdzania

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

Akceptacja narzędzia przerywa korzystanie z reason: "tool_call" i użycie toolCallId. Końcowy ChatResponseUpdate z AGUIChatClient zachowuje wartości outcome i interrupts w additional_properties. Interrupt i ResumeEntry to typy protokołów pochodzące z ag_ui.core, a nie modele specyficzne dla Agent Framework.

Wznów formatowanie

Wznów ten sam wątek z kanoniczną tablicą resume. Użyj polecenia accepted: false , aby odrzucić operację, zezwalając agentowi na kontynuowanie. Użyj status: "cancelled" bez ładunku danych, aby anulować przerwane uruchomienie.

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

Klient ze wsparciem zatwierdzania

Oto klient, który przetwarza żądania zatwierdzenia przy użyciu AGUIChatClient:

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

Przykładowa interakcja

Po uruchomieniu serwera i klienta:

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]

Jeśli użytkownik odrzuci:

Approve this action? (yes/no): no

[Sending approval response: False]

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

Niestandardowe komunikaty potwierdzenia

Dostosuj komunikaty zatwierdzenia i potwierdzenia w interfejsie użytkownika klienta AG-UI podczas renderowania przerwań zatwierdzania z serwera. Python AgentFrameworkAgent udostępnia żądania zatwierdzenia i metadane przerwań; nie przyjmuje obiektu strategii potwierdzenia po stronie serwera.

Najlepsze praktyki

Wyczyść opisy narzędzi

Podaj szczegółowe opisy, aby użytkownicy zrozumieli, co zatwierdzają:

@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

Granularne zatwierdzanie

Zażądaj zatwierdzenia poszczególnych akcji poufnych zamiast dzielenia na partie:

# 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

Argumenty informacyjne

Użyj opisowych nazw parametrów i podaj kontekst:

@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

Zarządzanie timeoutem

Ustaw odpowiednie limity czasu dla żądań zatwierdzenia:

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

Zatwierdzanie selektywne

Możesz mieszać narzędzia, które wymagają zatwierdzenia z tymi, które nie:

# 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

Zbiorcze zatwierdzanie i anulowanie

Jedna odpowiedź modelu może zawierać zarówno narzędzia wymagane do zatwierdzenia, jak i narzędzia, które nie wymagają zatwierdzenia. Rozwiązanie widocznego przerwania wykonuje również inne wywołania narzędzi z tej partii zgodnie z ich decyzjami o zatwierdzeniu. Na przykład równorzędny element never_require jest wykonywany, a jego TOOL_CALL_RESULT jest przesyłane strumieniowo we wznowionym przebiegu, nawet gdy równorzędny element wymagający zatwierdzenia zostanie odrzucony.

Zatwierdzone wywołania są ponownie kierowane przez standardową politykę Agenta i potok middleware; zatwierdzenie nie omija kontroli dopuszczenia ani harmonogramowania. Jeśli wywoływanie funkcji jest wyłączone w chwili wznowienia grantu, grant pozostaje w stanie oczekiwania do czasu jawnego ponowienia próby po ponownym włączeniu wywoływania funkcji. Ukończone równorzędne terminale można odtworzyć w celu przywrócenia stanu klienta bez ponownego wykonywania ich skutków ubocznych.

Anulowanie przy użyciu status: "cancelled" przerywa wznawianie zatwierdzania i czyści stan oczekującego zatwierdzenia dla wątku. Późniejsze żądania nie mogą przywracać ani wykonywać nieaktualnych wywołań narzędzia z anulowanej partii.

Następne kroki

Dodatkowe zasoby

• Omówienie AG-UI
• Renderowanie narzędzi zaplecza
• Narzędzia funkcji z zatwierdzeniami

Go obsługuje przepływy AG-UI z udziałem człowieka, z narzędziami wymagającymi zatwierdzenia. Opakuj narzędzie funkcji za pomocą tool.ApprovalRequiredFunc, a następnie udostępnij agenta przez 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)},
    },
})