Menggunakan alat fungsi dengan persetujuan manusia dalam proses yang melibatkan interaksi manusia

Langkah tutorial ini menunjukkan kepada Anda cara menggunakan alat fungsi yang memerlukan persetujuan manusia dengan agen, di mana agen dibangun di layanan Penyelesaian Obrolan Azure OpenAI.

Ketika agen memerlukan input pengguna apa pun, misalnya untuk menyetujui panggilan fungsi, ini disebut sebagai pola human-in-the-loop. Eksekusi agen yang memerlukan input pengguna, akan selesai dengan respons yang menunjukkan input apa yang diperlukan dari pengguna, alih-alih menyelesaikan jawaban akhir. Pemanggil agen kemudian bertanggung jawab untuk mendapatkan input yang diperlukan dari pengguna, dan meneruskannya kembali ke agen sebagai bagian dari eksekusi agen yang baru.

Prasyarat

Untuk prasyarat dan menginstal paket NuGet, lihat langkah Membuat dan menjalankan agen sederhana dalam tutorial ini.

Membuat agen dengan alat fungsi

Saat menggunakan fungsi, dimungkinkan untuk menunjukkan untuk setiap fungsi, apakah memerlukan persetujuan manusia sebelum dijalankan. Ini dilakukan dengan membungkus AIFunction instans dalam instans ApprovalRequiredAIFunction .

Berikut adalah contoh perangkat fungsi sederhana yang mensimulasikan pengambilan cuaca untuk lokasi tertentu.

using System;
using System.ComponentModel;
using System.Linq;
using Azure.AI.OpenAI;
using Azure.Identity;
using Microsoft.Agents.AI;
using Microsoft.Extensions.AI;
using OpenAI;

[Description("Get the weather for a given location.")]
static string GetWeather([Description("The location to get the weather for.")] string location)
    => $"The weather in {location} is cloudy with a high of 15°C.";

Untuk membuat AIFunction lalu membungkusnya dalam ApprovalRequiredAIFunction, Anda bisa melakukan hal berikut:

AIFunction weatherFunction = AIFunctionFactory.Create(GetWeather);
AIFunction approvalRequiredWeatherFunction = new ApprovalRequiredAIFunction(weatherFunction);

Saat membuat agen, Anda sekarang dapat memberikan alat fungsi yang membutuhkan persetujuan kepada agen dengan meneruskan daftar alat ke metode AsAIAgent.

AIAgent agent = new AzureOpenAIClient(
    new Uri("https://<myresource>.openai.azure.com"),
    new AzureCliCredential())
     .GetChatClient("gpt-4o-mini")
     .AsAIAgent(instructions: "You are a helpful assistant", tools: [approvalRequiredWeatherFunction]);

Karena Anda sekarang memiliki fungsi yang memerlukan persetujuan, agen mungkin merespons dengan permintaan persetujuan, alih-alih menjalankan fungsi secara langsung dan mengembalikan hasilnya. Anda dapat memeriksa konten respons untuk instans apa pun FunctionApprovalRequestContent , yang menunjukkan bahwa agen memerlukan persetujuan pengguna untuk fungsi.

AgentSession session = await agent.CreateSessionAsync();
AgentResponse response = await agent.RunAsync("What is the weather like in Amsterdam?", session);

var functionApprovalRequests = response.Messages
    .SelectMany(x => x.Contents)
    .OfType<FunctionApprovalRequestContent>()
    .ToList();

Jika ada permintaan persetujuan fungsi, detail panggilan fungsi termasuk nama dan argumen dapat ditemukan di FunctionCall properti pada FunctionApprovalRequestContent instans. Ini dapat ditampilkan kepada pengguna, sehingga mereka dapat memutuskan apakah akan menyetujui atau menolak panggilan fungsi. Untuk contoh ini, asumsikan ada satu permintaan.

FunctionApprovalRequestContent requestContent = functionApprovalRequests.First();
Console.WriteLine($"We require approval to execute '{requestContent.FunctionCall.Name}'");

Setelah pengguna memberikan input mereka, Anda dapat membuat instance FunctionApprovalResponseContent menggunakan metode CreateResponse pada FunctionApprovalRequestContent. Teruskan true untuk menyetujui panggilan fungsi, atau false untuk menolaknya.

Konten respons kemudian dapat diteruskan ke agen dalam konteks baru UserChatMessage, bersama dengan objek sesi yang sama untuk mendapatkan hasil dari agen.

var approvalMessage = new ChatMessage(ChatRole.User, [requestContent.CreateResponse(true)]);
Console.WriteLine(await agent.RunAsync(approvalMessage, session));

Setiap kali Anda menggunakan alat fungsi dengan persetujuan yang melibatkan manusia, ingatlah untuk memeriksa instance FunctionApprovalRequestContent dalam respons, setiap kali agen dijalankan, hingga semua panggilan fungsi disetujui atau ditolak.

Petunjuk / Saran

Lihat sampel .NET untuk contoh lengkap yang dapat dijalankan.

Langkah tutorial ini menunjukkan kepada Anda cara menggunakan alat fungsi yang memerlukan persetujuan manusia dengan agen.

Prasyarat

Untuk prasyarat dan menginstal paket Python, lihat langkah Membuat dan menjalankan agen sederhana dalam tutorial ini.

Membuat agen dengan alat fungsi yang memerlukan persetujuan

Saat menggunakan fungsi, dimungkinkan untuk menunjukkan untuk setiap fungsi, apakah memerlukan persetujuan manusia sebelum dijalankan. Ini dilakukan dengan mengatur parameter approval_mode ke "always_require" ketika menggunakan dekorator @tool.

Berikut adalah contoh perangkat fungsi sederhana yang mensimulasikan pengambilan cuaca untuk lokasi tertentu.

from typing import Annotated
from agent_framework import tool

@tool
def get_weather(location: Annotated[str, "The city and state, e.g. San Francisco, CA"]) -> str:
    """Get the current weather for a given location."""
    return f"The weather in {location} is cloudy with a high of 15°C."

Untuk membuat fungsi yang memerlukan persetujuan, Anda dapat menggunakan approval_mode parameter :

@tool(approval_mode="always_require")
def get_weather_detail(location: Annotated[str, "The city and state, e.g. San Francisco, CA"]) -> str:
    """Get detailed weather information for a given location."""
    return f"The weather in {location} is cloudy with a high of 15°C, humidity 88%."

Saat membuat agen, Anda sekarang dapat memberikan alat fungsi yang memerlukan persetujuan kepada agen, dengan meneruskan daftar alat fungsi ke Agent konstruktor.

from agent_framework import Agent
from agent_framework.openai import OpenAIResponsesClient

async with Agent(
    chat_client=OpenAIResponsesClient(),
    name="WeatherAgent",
    instructions="You are a helpful weather assistant.",
    tools=[get_weather, get_weather_detail],
) as agent:
    # Agent is ready to use

Karena Anda sekarang memiliki fungsi yang memerlukan persetujuan, agen mungkin merespons dengan permintaan persetujuan, alih-alih menjalankan fungsi secara langsung dan mengembalikan hasilnya. Anda dapat memeriksa respons untuk setiap permintaan input pengguna, yang menunjukkan bahwa agen memerlukan persetujuan pengguna untuk fungsi.

result = await agent.run("What is the detailed weather like in Amsterdam?")

if result.user_input_requests:
    for user_input_needed in result.user_input_requests:
        print(f"Function: {user_input_needed.function_call.name}")
        print(f"Arguments: {user_input_needed.function_call.arguments}")

Jika ada permintaan persetujuan fungsi, detail panggilan fungsi termasuk nama dan argumen dapat ditemukan di function_call properti pada permintaan input pengguna. Ini dapat ditampilkan kepada pengguna, sehingga mereka dapat memutuskan apakah akan menyetujui atau menolak panggilan fungsi.

Setelah pengguna memberikan input mereka, Anda dapat membuat respons menggunakan create_response metode pada permintaan input pengguna. Teruskan True untuk menyetujui panggilan fungsi, atau False untuk menolaknya.

Respons kemudian dapat diteruskan ke agen di Message yang baru, untuk gunanya mendapatkan hasilnya kembali dari agen.

from agent_framework import Message

# Get user approval (in a real application, this would be interactive)
user_approval = True  # or False to reject

# Create the approval response
approval_message = Message(
    role="user", 
    contents=[user_input_needed.create_response(user_approval)]
)

# Continue the conversation with the approval
final_result = await agent.run([
    "What is the detailed weather like in Amsterdam?",
    Message(role="assistant", contents=[user_input_needed]),
    approval_message
])
print(final_result.text)

Menangani persetujuan dalam perulangan

Saat bekerja dengan beberapa panggilan fungsi yang memerlukan persetujuan, Anda mungkin perlu menangani persetujuan dalam sebuah loop hingga semua fungsi dikerjakan atau ditolak.

async def handle_approvals(query: str, agent) -> str:
    """Handle function call approvals in a loop."""
    current_input = query

    while True:
        result = await agent.run(current_input)

        if not result.user_input_requests:
            # No more approvals needed, return the final result
            return result.text

        # Build new input with all context
        new_inputs = [query]

        for user_input_needed in result.user_input_requests:
            print(f"Approval needed for: {user_input_needed.function_call.name}")
            print(f"Arguments: {user_input_needed.function_call.arguments}")

            # Add the assistant message with the approval request
            new_inputs.append(Message(role="assistant", contents=[user_input_needed]))

            # Get user approval (in practice, this would be interactive)
            user_approval = True  # Replace with actual user input

            # Add the user's approval response
            new_inputs.append(
                Message(role="user", contents=[user_input_needed.create_response(user_approval)])
            )

        # Continue with all the context
        current_input = new_inputs

# Usage
result_text = await handle_approvals("Get detailed weather for Seattle and Portland", agent)
print(result_text)

Setiap kali Anda menggunakan alat fungsi dengan persetujuan keterlibatan manusia, ingatlah untuk memeriksa permintaan input pengguna dalam respons, setelah setiap kali agen dijalankan, sampai semua panggilan fungsi telah disetujui atau ditolak.

Contoh lengkap

# Copyright (c) Microsoft. All rights reserved.

import asyncio
from random import randrange
from typing import TYPE_CHECKING, Annotated, Any

from agent_framework import Agent, AgentResponse, Message, tool
from agent_framework.openai import OpenAIResponsesClient

if TYPE_CHECKING:
    from agent_framework import SupportsAgentRun

"""
Demonstration of a tool with approvals.

This sample demonstrates using AI functions with user approval workflows.
It shows how to handle function call approvals without using threads.
"""

conditions = ["sunny", "cloudy", "raining", "snowing", "clear"]


# NOTE: approval_mode="never_require" is for sample brevity. Use "always_require" in production; see samples/02-agents/tools/function_tool_with_approval.py and samples/02-agents/tools/function_tool_with_approval_and_sessions.py.
@tool(approval_mode="never_require")
def get_weather(location: Annotated[str, "The city and state, e.g. San Francisco, CA"]) -> str:
    """Get the current weather for a given location."""
    # Simulate weather data
    return f"The weather in {location} is {conditions[randrange(0, len(conditions))]} and {randrange(-10, 30)}°C."


# Define a simple weather tool that requires approval
@tool(approval_mode="always_require")
def get_weather_detail(location: Annotated[str, "The city and state, e.g. San Francisco, CA"]) -> str:
    """Get the current weather for a given location."""
    # Simulate weather data
    return (
        f"The weather in {location} is {conditions[randrange(0, len(conditions))]} and {randrange(-10, 30)}°C, "
        "with a humidity of 88%. "
        f"Tomorrow will be {conditions[randrange(0, len(conditions))]} with a high of {randrange(-10, 30)}°C."
    )


async def handle_approvals(query: str, agent: "SupportsAgentRun") -> AgentResponse:
    """Handle function call approvals.

    When we don't have a thread, we need to ensure we include the original query,
    the approval request, and the approval response in each iteration.
    """
    result = await agent.run(query)
    while len(result.user_input_requests) > 0:
        # Start with the original query
        new_inputs: list[Any] = [query]

        for user_input_needed in result.user_input_requests:
            print(
                f"\nUser Input Request for function from {agent.name}:"
                f"\n  Function: {user_input_needed.function_call.name}"
                f"\n  Arguments: {user_input_needed.function_call.arguments}"
            )

            # Add the assistant message with the approval request
            new_inputs.append(Message("assistant", [user_input_needed]))

            # Get user approval
            user_approval = await asyncio.to_thread(input, "\nApprove function call? (y/n): ")

            # Add the user's approval response
            new_inputs.append(
                Message("user", [user_input_needed.to_function_approval_response(user_approval.lower() == "y")])
            )

        # Run again with all the context
        result = await agent.run(new_inputs)

    return result


async def handle_approvals_streaming(query: str, agent: "SupportsAgentRun") -> None:
    """Handle function call approvals with streaming responses.

    When we don't have a thread, we need to ensure we include the original query,
    the approval request, and the approval response in each iteration.
    """
    current_input: str | list[Any] = query
    has_user_input_requests = True
    while has_user_input_requests:
        has_user_input_requests = False
        user_input_requests: list[Any] = []

        # Stream the response
        async for chunk in agent.run(current_input, stream=True):
            if chunk.text:
                print(chunk.text, end="", flush=True)

            # Collect user input requests from the stream
            if chunk.user_input_requests:
                user_input_requests.extend(chunk.user_input_requests)

        if user_input_requests:
            has_user_input_requests = True
            # Start with the original query
            new_inputs: list[Any] = [query]

            for user_input_needed in user_input_requests:
                print(
                    f"\n\nUser Input Request for function from {agent.name}:"
                    f"\n  Function: {user_input_needed.function_call.name}"
                    f"\n  Arguments: {user_input_needed.function_call.arguments}"
                )

                # Add the assistant message with the approval request
                new_inputs.append(Message("assistant", [user_input_needed]))

                # Get user approval
                user_approval = await asyncio.to_thread(input, "\nApprove function call? (y/n): ")

                # Add the user's approval response
                new_inputs.append(
                    Message("user", [user_input_needed.to_function_approval_response(user_approval.lower() == "y")])
                )

            # Update input with all the context for next iteration
            current_input = new_inputs


async def run_weather_agent_with_approval(stream: bool) -> None:
    """Example showing AI function with approval requirement."""
    print(f"\n=== Weather Agent with Approval Required ({'Streaming' if stream else 'Non-Streaming'}) ===\n")

    async with Agent(
        client=OpenAIResponsesClient(),
        name="WeatherAgent",
        instructions=("You are a helpful weather assistant. Use the get_weather tool to provide weather information."),
        tools=[get_weather, get_weather_detail],
    ) as agent:
        query = "Can you give me an update of the weather in LA and Portland and detailed weather for Seattle?"
        print(f"User: {query}")

        if stream:
            print(f"\n{agent.name}: ", end="", flush=True)
            await handle_approvals_streaming(query, agent)
            print()
        else:
            result = await handle_approvals(query, agent)
            print(f"\n{agent.name}: {result}\n")


async def main() -> None:
    print("=== Demonstration of a tool with approvals ===\n")

    await run_weather_agent_with_approval(stream=False)
    await run_weather_agent_with_approval(stream=True)


if __name__ == "__main__":
    asyncio.run(main())

# Copyright (c) Microsoft. All rights reserved.

import asyncio
from random import randrange
from typing import TYPE_CHECKING, Annotated, Any

from agent_framework import Agent, AgentResponse, Message, tool
from agent_framework.openai import OpenAIResponsesClient

if TYPE_CHECKING:
    from agent_framework import SupportsAgentRun

"""
Demonstration of a tool with approvals.

This sample demonstrates using AI functions with user approval workflows.
It shows how to handle function call approvals without using threads.
"""

conditions = ["sunny", "cloudy", "raining", "snowing", "clear"]


# NOTE: approval_mode="never_require" is for sample brevity. Use "always_require" in production; see samples/02-agents/tools/function_tool_with_approval.py and samples/02-agents/tools/function_tool_with_approval_and_sessions.py.
@tool(approval_mode="never_require")
def get_weather(location: Annotated[str, "The city and state, e.g. San Francisco, CA"]) -> str:
    """Get the current weather for a given location."""
    # Simulate weather data
    return f"The weather in {location} is {conditions[randrange(0, len(conditions))]} and {randrange(-10, 30)}°C."


# Define a simple weather tool that requires approval
@tool(approval_mode="always_require")
def get_weather_detail(location: Annotated[str, "The city and state, e.g. San Francisco, CA"]) -> str:
    """Get the current weather for a given location."""
    # Simulate weather data
    return (
        f"The weather in {location} is {conditions[randrange(0, len(conditions))]} and {randrange(-10, 30)}°C, "
        "with a humidity of 88%. "
        f"Tomorrow will be {conditions[randrange(0, len(conditions))]} with a high of {randrange(-10, 30)}°C."
    )


async def handle_approvals(query: str, agent: "SupportsAgentRun") -> AgentResponse:
    """Handle function call approvals.

    When we don't have a thread, we need to ensure we include the original query,
    the approval request, and the approval response in each iteration.
    """
    result = await agent.run(query)
    while len(result.user_input_requests) > 0:
        # Start with the original query
        new_inputs: list[Any] = [query]

        for user_input_needed in result.user_input_requests:
            print(
                f"\nUser Input Request for function from {agent.name}:"
                f"\n  Function: {user_input_needed.function_call.name}"
                f"\n  Arguments: {user_input_needed.function_call.arguments}"
            )

            # Add the assistant message with the approval request
            new_inputs.append(Message("assistant", [user_input_needed]))

            # Get user approval
            user_approval = await asyncio.to_thread(input, "\nApprove function call? (y/n): ")

            # Add the user's approval response
            new_inputs.append(
                Message("user", [user_input_needed.to_function_approval_response(user_approval.lower() == "y")])
            )

        # Run again with all the context
        result = await agent.run(new_inputs)

    return result


async def handle_approvals_streaming(query: str, agent: "SupportsAgentRun") -> None:
    """Handle function call approvals with streaming responses.

    When we don't have a thread, we need to ensure we include the original query,
    the approval request, and the approval response in each iteration.
    """
    current_input: str | list[Any] = query
    has_user_input_requests = True
    while has_user_input_requests:
        has_user_input_requests = False
        user_input_requests: list[Any] = []

        # Stream the response
        async for chunk in agent.run(current_input, stream=True):
            if chunk.text:
                print(chunk.text, end="", flush=True)

            # Collect user input requests from the stream
            if chunk.user_input_requests:
                user_input_requests.extend(chunk.user_input_requests)

        if user_input_requests:
            has_user_input_requests = True
            # Start with the original query
            new_inputs: list[Any] = [query]

            for user_input_needed in user_input_requests:
                print(
                    f"\n\nUser Input Request for function from {agent.name}:"
                    f"\n  Function: {user_input_needed.function_call.name}"
                    f"\n  Arguments: {user_input_needed.function_call.arguments}"
                )

                # Add the assistant message with the approval request
                new_inputs.append(Message("assistant", [user_input_needed]))

                # Get user approval
                user_approval = await asyncio.to_thread(input, "\nApprove function call? (y/n): ")

                # Add the user's approval response
                new_inputs.append(
                    Message("user", [user_input_needed.to_function_approval_response(user_approval.lower() == "y")])
                )

            # Update input with all the context for next iteration
            current_input = new_inputs


async def run_weather_agent_with_approval(stream: bool) -> None:
    """Example showing AI function with approval requirement."""
    print(f"\n=== Weather Agent with Approval Required ({'Streaming' if stream else 'Non-Streaming'}) ===\n")

    async with Agent(
        client=OpenAIResponsesClient(),
        name="WeatherAgent",
        instructions=("You are a helpful weather assistant. Use the get_weather tool to provide weather information."),
        tools=[get_weather, get_weather_detail],
    ) as agent:
        query = "Can you give me an update of the weather in LA and Portland and detailed weather for Seattle?"
        print(f"User: {query}")

        if stream:
            print(f"\n{agent.name}: ", end="", flush=True)
            await handle_approvals_streaming(query, agent)
            print()
        else:
            result = await handle_approvals(query, agent)
            print(f"\n{agent.name}: {result}\n")


async def main() -> None:
    print("=== Demonstration of a tool with approvals ===\n")

    await run_weather_agent_with_approval(stream=False)
    await run_weather_agent_with_approval(stream=True)


if __name__ == "__main__":
    asyncio.run(main())

Langkah selanjutnya

Menghasilkan Output Terstruktur dengan agen

Saran dan Komentar

Apakah halaman ini membantu?

Last updated on 2026-02-13

Bagikan melalui

Menggunakan alat fungsi dengan persetujuan manusia dalam proses yang melibatkan interaksi manusia

Prasyarat

Membuat agen dengan alat fungsi

Prasyarat

Membuat agen dengan alat fungsi yang memerlukan persetujuan

Menangani persetujuan dalam perulangan

Contoh lengkap

Langkah selanjutnya

Saran dan Komentar

Sumber Daya Tambahan: