透過 AG-UI 進行人機互動

MAF 工具核准流程仍負責判定工具是否需要核准。 AG-UI 將核准請求傳送給客戶端，並將客戶端的決策回傳給伺服器。

有關核准原則、條件式規則和一般安全指引，請參閱使用具有人機互動核准的函式工具。

需要核准

用 ApprovalRequiredAIFunction 包裝 MAF 函式，並照常公開代理程式：

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

當模型呼叫工具時，AG-UI 介面卡會以工具呼叫中斷結束執行，而非執行該函式。

處理來自 .NET 用戶端的中斷

AGUIChatClient 將中斷呈現為 ToolApprovalRequestContent。 使用一般的 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.
    }
}

傳送回應時重用相同的 AgentSession，讓用戶端能繼續已中斷的執行作業。 用 approved: false 來拒絕接聽電話。 轉接器會將 MAF 回應轉換為標準的 AG-UI 恢復承載資料。

下一步

本教學將示範如何使用 AG-UI 實作需要人為介入的工作流程，其中使用者需先核准工具的執行操作。 這對於金融交易、資料修改或具有重大後果的操作等敏感操作至關重要。

Prerequisites

在開始之前，請確保您已完成 後端工具渲染 教學課程並了解：

• 如何建立函數工具
• AG-UI 如何串流工具事件資料
• 基本伺服器和用戶端設定

什麼是人機互動？

人機交互 （HITL） 是一種模式，代理在執行某些操作之前請求用戶批准。 使用 AG-UI：

• 代理程式會照常產生工具呼叫
• 伺服器不會立即執行，而是將核准請求傳送給用戶端
• 用戶端會顯示要求並提示使用者
• 使用者核准或拒絕動作
• 伺服器收到回應並相應地進行

Benefits

• 安全：防止執行意外操作
• 透明度：使用者準確地看到客服專員想要做什麼
• 控制：使用者對敏感操作擁有最終決定權
• 合規性：滿足人工監督的監管要求

核准標記工具

若要核准工具，請在 @tool 裝飾項目中使用 approval_mode 參數：

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"

審批模式

• always_require：在執行前始終請求批准
• never_require：從不要求核准 （預設行為）
• conditional：根據特定條件請求批准（自訂邏輯）

建立具有人工介入迴圈的伺服器

以下是包含需要核准工具的完整伺服器實作：

"""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,
    "/",
    single_user=True,
)

if __name__ == "__main__":
    import uvicorn

    uvicorn.run(app, host="127.0.0.1", port=8888)

關鍵概念

• AgentFrameworkAgent 包裝函式：啟用 AG-UI 通訊協定功能，例如人機互動
• require_confirmation=True：啟動標記工具的核准工作流程
• 工具層級控制：只有標記為 的工具 approval_mode="always_require" 才會要求核准

了解核准中斷

當工具需要核准時，執行會以標準的 AG-UI 中斷結束。

批准中斷

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

工具核准中斷會使用 reason: "tool_call" ，並包含一個 toolCallId。 來自 ChatResponseUpdate 的最後一個 AGUIChatClient 會保留 outcome 中的 interrupts 和 additional_properties 值。 Interrupt 以及 ResumeEntry 是來自 ag_ui.core的協定類型，而非 Agent Framework 專屬模型。

履歷格式

用標準 resume 陣列繼續執行同一個執行緒。 用 accepted: false 來拒絕操作，同時讓代理人繼續。 使用不含承載資料的 status: "cancelled" 來取消已中斷的執行。

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

提供審批支持的客戶

以下是處理 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())

互動範例

當伺服器和用戶端執行時：

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]

如果使用者拒絕：

Approve this action? (yes/no): no

[Sending approval response: False]

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

自訂確認訊息

當從伺服器轉譯核准中斷時，在您的 AG-UI 用戶端 UI 中自訂核准與確認訊息。 Python AgentFrameworkAgent 會公開核准要求與中斷中繼資料；它不接受伺服器端確認策略物件。

最佳做法

清晰的工具描述

提供詳細的描述，讓使用者瞭解他們正在核准的內容：

@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

精細核准

要求核准個別敏感動作，而不是批次處理：

# 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

資訊豐富的論點

使用描述性參數名稱並提供上下文：

@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

逾時處理

為核准請求設定適當的逾時：

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

選擇性批准

您可以將需要核准的工具與不需要核准的工具混合使用：

# 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

批次核准與取消

一個模型回應可以包含需要核准的工具，以及不需要核准的工具。 解決可見的中斷同時也會根據各自的核准決策，完成該批次中的其他工具呼叫。 例如，某個 never_require 同層級項目會執行，而且其 TOOL_CALL_RESULT 會在繼續執行時進行串流傳送，即使需要核准的同層級項目遭到拒絕亦然。

經核准的呼叫會繼續經由正常的 Agent 原則與中介軟體管線處理；核准不會繞過准入或排程。 如果在恢復授權時函式叫用已停用，則該授權會維持擱置狀態，直到重新啟用函式叫用後明確重試為止。 完成的終端機同級進程可以重播以恢復用戶端狀態，而不會重新執行其副作用。

使用 status: "cancelled" 取消會中止核准繼續，並清除該執行緒的佇列核准狀態。 後續的請求無法重新浮現或執行來自已取消批次的過時工具呼叫。

下一步

其他資源

• AG-UI 概覽
• 後端工具渲染
• 經核可的功能性工具

使用 tool.ApprovalRequiredFunc 將 Go 函式工具標示為需要核准。 這會在代理層設定批准，但並未實作端到端的 AG-UI 審核流程。

目前的 Go NewJSONHTTPHandler 不會轉換 ToolApprovalRequestContent 成核准事件，也不會在 HTTP 請求間保留代理會話。 應用程式需要自訂的核准傳輸機制和經過授權的工作階段保存，才能完成整個流程。

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)},
    },
})