Używanie narzędzi funkcjonalnych z zatwierdzeniami przy udziale człowieka

W tym kroku samouczka pokazano, jak używać narzędzi funkcji wymagających zatwierdzenia przez człowieka przy użyciu agenta opartego na usłudze Azure OpenAI Chat Completion.

Gdy agenci wymagają danych wejściowych od użytkownika, na przykład w celu zatwierdzenia wywołania funkcji, nazywa się to wzorcem z udziałem człowieka. Uruchomienie agenta, które wymaga danych wejściowych użytkownika, zostanie ukończone z odpowiedzią wskazującą, jakie dane wejściowe są wymagane od użytkownika, zamiast kończyć się ostateczną odpowiedzią. Obiekt wywołujący agenta jest następnie odpowiedzialny za uzyskanie wymaganych danych wejściowych od użytkownika i przekazanie ich z powrotem do agenta w ramach nowego uruchomienia agenta.

Wymagania wstępne

Aby uzyskać wymagania wstępne i zainstalować pakiety NuGet, zobacz krok Tworzenie i uruchamianie prostego agenta w tym samouczku.

Tworzenie agenta za pomocą narzędzi funkcji

W przypadku korzystania z funkcji można wskazać dla każdej funkcji, czy wymaga zatwierdzenia przez człowieka przed wykonaniem. Odbywa się to przez opakowanie wystąpienia AIFunction w wystąpienie ApprovalRequiredAIFunction.

Oto przykład prostego narzędzia funkcjonalnego, które symuluje pobieranie danych pogodowych dla danej lokalizacji.

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.";

Aby utworzyć element AIFunction , a następnie opakować go w obiekcie ApprovalRequiredAIFunction, możesz wykonać następujące czynności:

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

Podczas tworzenia agenta można teraz dostarczyć agentowi funkcję narzędzia wymagającego zatwierdzenia, przekazując listę narzędzi do metody 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]);

Ponieważ masz teraz funkcję, która wymaga zatwierdzenia, agent może odpowiedzieć z żądaniem zatwierdzenia, zamiast wykonywać funkcję bezpośrednio i zwracać wynik. Możesz sprawdzić zawartość odpowiedzi dla dowolnego FunctionApprovalRequestContent wystąpienia, co wskazuje, że agent wymaga zatwierdzenia przez użytkownika dla funkcji.

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

Jeśli istnieją jakiekolwiek żądania zatwierdzenia funkcji, szczegóły wywołania funkcji, w tym nazwę i argumenty, można znaleźć we FunctionCall właściwości w wystąpieniu FunctionApprovalRequestContent. Można to wyświetlić użytkownikowi, aby zdecydować, czy zatwierdzić, czy odrzucić wywołanie funkcji. W tym przykładzie załóżmy, że istnieje jedno żądanie.

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

Po podaniu danych wejściowych przez użytkownika, możesz utworzyć FunctionApprovalResponseContent wystąpienie przy użyciu metody CreateResponse na FunctionApprovalRequestContent. Przekaż true, aby zatwierdzić wywołanie funkcji, lub false, aby ją odrzucić.

Zawartość odpowiedzi można następnie przekazać do agenta w nowym obiekcie UserChatMessage, wraz z tym samym obiektem sesji, aby ponownie uzyskać wynik od agenta.

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

Za każdym razem, gdy używasz narzędzi funkcyjnych z zatwierdzeniami typu human-in-the-loop, pamiętaj, aby sprawdzić wystąpienia FunctionApprovalRequestContent w odpowiedzi, po każdym uruchomieniu agenta, aż wszystkie wywołania funkcji zostaną zatwierdzone lub odrzucone.

Tip

Zobacz przykłady dla platformy .NET , aby uzyskać pełne przykłady możliwych do uruchomienia.

W tym kroku samouczka pokazano, jak używać narzędzi funkcjonalnych wymagających zatwierdzenia przez agenta.

Gdy agenci wymagają danych wejściowych od użytkownika, na przykład w celu zatwierdzenia wywołania funkcji, nazywa się to wzorcem z udziałem człowieka. Uruchomienie agenta, które wymaga danych wejściowych użytkownika, zostanie ukończone z odpowiedzią wskazującą, jakie dane wejściowe są wymagane od użytkownika, zamiast kończyć się ostateczną odpowiedzią. Obiekt wywołujący agenta jest następnie odpowiedzialny za uzyskanie wymaganych danych wejściowych od użytkownika i przekazanie ich z powrotem do agenta w ramach nowego uruchomienia agenta.

Wymagania wstępne

Aby uzyskać wymagania wstępne i zainstalować pakiety języka Python, zobacz krok Tworzenie i uruchamianie prostego agenta w tym samouczku.

Utwórz agenta z użyciem narzędzi funkcji wymagających zatwierdzenia

W przypadku korzystania z funkcji można wskazać dla każdej funkcji, czy wymaga zatwierdzenia przez człowieka przed wykonaniem. W tym celu należy ustawić approval_mode parametr na "always_require" w przypadku korzystania z dekoratora @tool .

Oto przykład prostego narzędzia funkcjonalnego, które symuluje pobieranie danych pogodowych dla danej lokalizacji.

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

Aby utworzyć funkcję, która wymaga zatwierdzenia, możesz użyć parametru approval_mode :

@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%."

Podczas tworzenia agenta można teraz podać narzędzie do funkcji wymagającej zatwierdzenia agentowi, przekazując listę narzędzi do konstruktora Agent.

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

Ponieważ masz teraz funkcję, która wymaga zatwierdzenia, agent może odpowiedzieć z żądaniem zatwierdzenia, zamiast wykonywać funkcję bezpośrednio i zwracać wynik. Możesz sprawdzić odpowiedź na wszystkie żądania danych wejściowych użytkownika, które wskazują, że agent wymaga zatwierdzenia przez użytkownika dla funkcji.

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

Jeśli istnieją jakiekolwiek żądania zatwierdzenia funkcji, szczegóły wywołania funkcji, w tym nazwy i argumentów, można znaleźć we function_call właściwości w żądaniu wejściowym użytkownika. Można to wyświetlić użytkownikowi, aby zdecydować, czy zatwierdzić, czy odrzucić wywołanie funkcji.

Po tym jak użytkownik poda swoje dane wejściowe, możesz utworzyć odpowiedź przy użyciu metody create_response w ramach żądania dotyczącego danych wejściowych użytkownika. Przekaż True, aby zatwierdzić wywołanie funkcji, lub False, aby ją odrzucić.

Następnie można przekazać odpowiedź do agenta w nowym pliku Message, aby uzyskać wynik z powrotem od agenta.

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)

Obsługa zatwierdzeń w pętli

Podczas pracy z wieloma wywołaniami funkcji, które wymagają zatwierdzenia, może być konieczne obsłużenie zatwierdzeń w pętli, dopóki wszystkie funkcje nie zostaną zatwierdzone lub odrzucone:

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)

Za każdym razem, gdy używasz narzędzi funkcji z aprobatą typu "człowiek w pętli", pamiętaj, aby sprawdzić, czy w odpowiedzi po każdego uruchomienia agenta znajdują się żądania danych wejściowych użytkownika, aż wszystkie wywołania funkcji zostaną zaakceptowane lub odrzucone.

Kompletny przykład

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

Dalsze kroki

Tworzenie danych wyjściowych ze strukturą przy użyciu agentów