Uso di strumenti funzionali con approvazioni umane integrate

Questo passaggio dell'esercitazione illustra come utilizzare strumenti funzionali che richiedono l'approvazione umana con un agente, basato sul servizio di Completamento della chat OpenAI di Azure.

Quando gli agenti richiedono un input dell'utente, ad esempio per approvare una chiamata di funzione, questo è definito come un modello di interazione umano-in-the-loop. Un'esecuzione dell'agente che richiede l'input dell'utente, verrà completata con una risposta che indica l'input richiesto dall'utente, invece di completare con una risposta finale. Il chiamante dell'agente è quindi responsabile di ottenere l'input richiesto dall'utente e consegnarlo all'agente come parte di una nuova esecuzione dell'agente.

Prerequisiti

Per i prerequisiti e l'installazione dei pacchetti NuGet, vedere il passaggio Creare ed eseguire un agente semplice in questa esercitazione.

Creare l'agente con gli strumenti per le funzioni

Quando si usano le funzioni, è possibile indicare per ogni funzione, se è necessaria l'approvazione umana prima dell'esecuzione. Questa operazione viene eseguita incapsulando l'istanza AIFunction all'interno dell'istanza ApprovalRequiredAIFunction.

Ecco un esempio di uno strumento di funzione semplice che fa finta di ottenere il meteo per una determinata posizione.

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

Per creare un oggetto AIFunction e quindi eseguirne il wrapping in un ApprovalRequiredAIFunctionoggetto , è possibile eseguire le operazioni seguenti:

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

Quando si crea l'agente, ora è possibile fornire lo strumento di funzione di approvazione all'agente passando un elenco di strumenti al metodo 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]);

Poiché ora si dispone di una funzione che richiede l'approvazione, l'agente potrebbe rispondere con una richiesta di approvazione, anziché eseguire direttamente la funzione e restituire il risultato. È possibile controllare il contenuto della risposta per tutte FunctionApprovalRequestContent le istanze, che indica che l'agente richiede l'approvazione dell'utente per una funzione.

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

Se vi sono richieste di approvazione delle funzioni, i dettagli della chiamata di funzione, inclusi il nome e gli argomenti, sono disponibili nella proprietà FunctionCall dell'istanza FunctionApprovalRequestContent. Può essere visualizzato all'utente, in modo che possa decidere se approvare o rifiutare la chiamata di funzione. Per questo esempio, si supponga che sia presente una richiesta.

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

Dopo che l'utente ha fornito il proprio input, è possibile creare un'istanza FunctionApprovalResponseContent usando il CreateResponse metodo in FunctionApprovalRequestContent. Passare true per approvare la chiamata di funzione o false rifiutarla.

Il contenuto della risposta può quindi essere passato all'agente in un nuovo UserChatMessageoggetto , insieme allo stesso oggetto sessione per ottenere il risultato dall'agente.

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

Ogni volta che si usano strumenti di funzione con approvazioni con l'intervento umano, ricordarsi di verificare la presenza di istanze FunctionApprovalRequestContent nella risposta, dopo l'esecuzione di ogni agente, finché tutte le chiamate di funzione siano approvate o rifiutate.

Suggerimento

Per esempi eseguibili completi, vedere gli esempi .NET .

Questo passaggio dell'esercitazione illustra come usare le funzioni che richiedono l'approvazione da parte di un essere umano con un agente.

Quando gli agenti richiedono un input dell'utente, ad esempio per approvare una chiamata di funzione, questo è definito come un modello di interazione umano-in-the-loop. Un'esecuzione dell'agente che richiede l'input dell'utente, verrà completata con una risposta che indica l'input richiesto dall'utente, invece di completare con una risposta finale. Il chiamante dell'agente è quindi responsabile di ottenere l'input richiesto dall'utente e consegnarlo all'agente come parte di una nuova esecuzione dell'agente.

Prerequisiti

Per i prerequisiti e l'installazione dei pacchetti Python, vedere il passaggio Creare ed eseguire un agente semplice in questa esercitazione.

Creare l'agente con gli strumenti per le funzioni che richiedono l'approvazione

Quando si usano le funzioni, è possibile indicare per ogni funzione, se è necessaria l'approvazione umana prima dell'esecuzione. Questa operazione viene eseguita impostando il parametro approval_mode su "always_require" quando si usa il decoratore @tool.

Ecco un esempio di uno strumento di funzione semplice che fa finta di ottenere il meteo per una determinata posizione.

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

Per creare una funzione che richiede l'approvazione, è possibile usare il approval_mode parametro :

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

Quando si crea l'agente, è ora possibile fornire all'agente uno strumento di funzione che richiede approvazione, passando un elenco di strumenti al costruttore 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

Poiché ora si dispone di una funzione che richiede l'approvazione, l'agente potrebbe rispondere con una richiesta di approvazione, anziché eseguire direttamente la funzione e restituire il risultato. È possibile controllare la risposta per qualsiasi richiesta di input dell'utente, che indica che l'agente richiede l'approvazione dell'utente per una funzione.

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

Se sono presenti richieste di approvazione delle funzioni, i dettagli della chiamata di funzione, inclusi il nome e gli argomenti, sono disponibili nella proprietà nella function_call richiesta di input dell'utente. Può essere visualizzato all'utente, in modo che possa decidere se approvare o rifiutare la chiamata di funzione.

Dopo che l'utente ha fornito il proprio input, è possibile creare una risposta usando il metodo create_response sulla richiesta di input dell'utente. Passare True per approvare la chiamata di funzione o False rifiutarla.

La risposta può quindi essere passata all'agente in un nuovo Message, per ottenere indietro il risultato dall'agente.

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)

Gestione delle approvazioni in un ciclo

Quando si usano più chiamate di funzione che richiedono l'approvazione, potrebbe essere necessario gestire le approvazioni in un ciclo fino a quando tutte le funzioni non vengono approvate o rifiutate:

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)

Ogni volta che si utilizzano strumenti di funzione con approvazione umana nel processo, ricordarsi di verificare la presenza di richieste di input da parte dell'utente nella risposta, dopo ogni esecuzione dell'agente, fino a quando tutte le chiamate di funzione non siano state approvate o rifiutate.

Esempio completo

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

Passaggi successivi

Produzione di output strutturato con agenti