Utilisation d’outils fonctionnels avec intégration humaine dans le processus d'approbation

Cette étape du tutoriel vous montre comment utiliser des outils de fonction qui nécessitent une approbation humaine avec un agent, où l’agent est basé sur le service Azure OpenAI Chat Completion.

Lorsque les agents nécessitent une entrée utilisateur, par exemple pour approuver un appel de fonction, il s’agit d’un schéma avec intervention humaine. Une exécution de l’agent qui nécessite une entrée utilisateur est terminée avec une réponse qui indique quelle entrée est requise de l’utilisateur, au lieu de terminer avec une réponse finale. L’appelant de l’agent est ensuite chargé d’obtenir l’entrée requise de l’utilisateur et de la renvoyer à l’agent dans le cadre d’une nouvelle exécution de l’agent.

Prerequisites

Pour connaître les prérequis et l’installation des packages NuGet, consultez l’étape Créer et exécuter un agent simple dans ce tutoriel.

Créer l’agent avec des outils de fonction

Lorsque vous utilisez des fonctions, il est possible d’indiquer pour chaque fonction s’il nécessite une approbation humaine avant d’être exécutée. Pour ce faire, envelopper l’instance AIFunction dans une instance ApprovalRequiredAIFunction.

Voici un exemple d’outil de fonction simple qui simule l'obtention de la météo pour un emplacement donné.

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

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

Pour créer un AIFunction et l’encapsuler ensuite dans un ApprovalRequiredAIFunction, vous pouvez effectuer les opérations suivantes :

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

Lors de la création de l'agent, vous pouvez désormais fournir à l'agent la fonction nécessitant une approbation, en passant une liste d’outils à la méthode AsAIAgent.

AIAgent agent = new AIProjectClient(
    new Uri("<your-foundry-project-endpoint>"),
    new DefaultAzureCredential())
     .AsAIAgent(
        model: "gpt-4o-mini",
        instructions: "You are a helpful assistant",
        tools: [approvalRequiredWeatherFunction]);

Étant donné que vous disposez maintenant d’une fonction qui nécessite une approbation, l’agent peut répondre avec une demande d’approbation au lieu d’exécuter la fonction directement et de retourner le résultat. Vous pouvez vérifier le contenu de la réponse pour toutes les ToolApprovalRequestContent instances, ce qui indique que l’agent requiert l’approbation de l’utilisateur pour une fonction.

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

var toolApprovalRequests = response.Messages
    .SelectMany(x => x.Contents)
    .OfType<ToolApprovalRequestContent>()
    .ToList();

S’il existe des demandes d’approbation de fonction, l’appel de fonction incluant son nom et ses arguments est disponible à partir de la ToolCall propriété sur l’instance ToolApprovalRequestContent . Cela peut être montré à l’utilisateur afin qu’il puisse décider s’il faut approuver ou rejeter l’appel de fonction. Pour cet exemple, supposons qu’il existe une seule requête.

ToolApprovalRequestContent requestContent = toolApprovalRequests.First();
var functionCall = (FunctionCallContent)requestContent.ToolCall;
Console.WriteLine($"We require approval to execute '{functionCall.Name}'");

Une fois que l’utilisateur a fourni son entrée, utilisez la CreateResponse méthode pour ToolApprovalRequestContent créer la réponse d’approbation. Utilisez true pour approuver l’appel de fonction, ou false pour le rejeter.

Le contenu de la réponse peut ensuite être transmis à l’agent dans un nouveau UserChatMessage, ainsi que le même objet de session pour récupérer le résultat de l’agent.

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

Chaque fois que vous utilisez des outils de fonction avec des approbations par intervention humaine, n’oubliez pas de vérifier la présence d’instances ToolApprovalRequestContent dans la réponse, après l’exécution de chaque agent, jusqu’à ce que tous les appels de fonction aient été approuvés ou rejetés.

Cette étape du tutoriel vous montre comment utiliser des outils de fonction qui nécessitent une approbation humaine avec un agent.

Lorsque les agents nécessitent une entrée utilisateur, par exemple pour approuver un appel de fonction, il s’agit d’un schéma avec intervention humaine. Une exécution de l’agent qui nécessite une entrée utilisateur est terminée avec une réponse qui indique quelle entrée est requise de l’utilisateur, au lieu de terminer avec une réponse finale. L’appelant de l’agent est ensuite chargé d’obtenir l’entrée requise de l’utilisateur et de la renvoyer à l’agent dans le cadre d’une nouvelle exécution de l’agent.

Prerequisites

Pour connaître les prérequis et l’installation des packages Python, consultez l’étape Créer et exécuter un agent simple dans ce tutoriel.

Créer l’agent avec des outils de fonction nécessitant une approbation

Lorsque vous utilisez des fonctions, il est possible d’indiquer pour chaque fonction s’il nécessite une approbation humaine avant d’être exécutée. Pour ce faire, définissez le paramètre approval_mode sur "always_require" en utilisant le décorateur @tool.

Voici un exemple d’outil de fonction simple qui simule l'obtention de la météo pour un emplacement donné.

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

Pour créer une fonction qui nécessite une approbation, vous pouvez utiliser le approval_mode paramètre :

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

Lors de la création de l’agent, vous pouvez maintenant fournir à l'agent l'outil de fonction nécessitant une approbation, en transmettant une liste d’outils au Agent constructeur.

from agent_framework import Agent
from agent_framework.openai import OpenAIChatClient

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

Étant donné que vous disposez maintenant d’une fonction qui nécessite une approbation, l’agent peut répondre avec une demande d’approbation au lieu d’exécuter la fonction directement et de retourner le résultat. Vous pouvez vérifier la réponse pour toutes les demandes d’entrée utilisateur, ce qui indique que l’agent requiert l’approbation de l’utilisateur pour une fonction.

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:
        if user_input_needed.function_call is None:
            continue
        print(f"Function: {user_input_needed.function_call.name}")
        print(f"Arguments: {user_input_needed.function_call.arguments}")

S’il existe des demandes d’approbation de fonction, le détail de l’appel de fonction, y compris le nom et les arguments, se trouve dans la function_call propriété sur la demande d’entrée utilisateur. Cela peut être montré à l’utilisateur afin qu’il puisse décider s’il faut approuver ou rejeter l’appel de fonction.

Une fois que l’utilisateur a fourni son entrée, vous pouvez créer une réponse à l’aide de la to_function_approval_response méthode sur la demande d’entrée utilisateur. Utilisez True pour approuver l’appel de fonction, ou False pour le rejeter.

La réponse peut ensuite être transmise à l’agent dans un nouveau Message, afin de récupérer le résultat de l’agent.

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.to_function_approval_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)

Gestion des approbations dans une boucle

Lorsque vous utilisez plusieurs appels de fonction qui nécessitent une approbation, vous devrez peut-être gérer les approbations dans une boucle jusqu’à ce que toutes les fonctions soient approuvées ou rejetées :

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:
            if user_input_needed.function_call is None:
                continue
            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.to_function_approval_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)

Chaque fois que vous utilisez des outils de fonction avec des approbations par intervention humaine, n’oubliez pas de vérifier des demandes d'entrée utilisateur dans la réponse, après l’exécution de chaque agent, jusqu’à ce que tous les appels de fonction aient été approuvés ou rejetés.

Exemple complet

# 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 OpenAIChatClient

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=OpenAIChatClient(),
        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())

Approbation de l’outil

Vous pouvez exiger l’approbation humaine avant l’exécution d’un outil en l’encapsulant avec tool.ApprovalRequiredFunc:

import "github.com/microsoft/agent-framework-go/tool"

approvedWeatherTool := tool.ApprovalRequiredFunc(weatherTool)

a := foundryprovider.NewAgent(endpoint, token, foundryprovider.ModelDeployment(model), foundryprovider.AgentConfig{
    Instructions: "You are a helpful assistant.",
    Config: agent.Config{
        Tools: []tool.Tool{approvedWeatherTool},
    },
})

Lorsque le modèle demande un appel d’outil, l’infrastructure l’intercepte et attend l’approbation avant l’exécution. Le flux d’approbation est géré par le biais d’un intergiciel.

Encapsulez les fonctions qui nécessitent une approbation, ApprovalRequiredAIFunctionpuis ajoutez-les par le biais HarnessAgentOptions.ChatOptions.Toolsde :

using Microsoft.Agents.AI;
using Microsoft.Extensions.AI;

var weatherTool = new ApprovalRequiredAIFunction(
    AIFunctionFactory.Create(GetWeather));

AIAgent agent = chatClient.AsHarnessAgent(new HarnessAgentOptions
{
    ChatOptions = new ChatOptions
    {
        Instructions = "You are a helpful assistant.",
        Tools = [weatherTool],
    },
});

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

DisableToolAutoApproval falseest défini par défaut sur , de sorte que le harnais ajoute ToolApprovalAgent. Avec la valeur par défaut ToolApprovalAgentOptions, aucune règle heuristique n’est configurée ; les éléments sans correspondance ToolApprovalRequestContent retournent toujours à l’appelant pour approbation. Pour ajouter des rappels d’approbation automatique approuvés, définissez ToolApprovalAgentOptions.AutoApprovalRules.

Le paramètre DisableToolAutoApproval = true supprime uniquement la règle debout, la mise en file d’attente et l’intergiciel d’approbation automatique heuristique. Elle ne supprime pas l’exigence d’approbation d’un ApprovalRequiredAIFunction. La liaison de réponse d’approbation et le contournement des outils qui ne nécessitent pas d’approbation restent également activés par défaut ; leurs retraits distincts sont DisableApprovalResponseBinding et DisableApprovalNotRequiredFunctionBypassing.

Marquez l’outil avec approval_mode="always_require" et passez-le à create_harness_agent:

from agent_framework import create_harness_agent, tool

@tool(approval_mode="always_require")
def get_weather_detail(location: str) -> str:
    """Get detailed weather information for a location."""
    return f"The weather in {location} is cloudy with a high of 15°C."

agent = create_harness_agent(
    client=client,
    agent_instructions="You are a helpful weather assistant.",
    tools=get_weather_detail,
)

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

disable_tool_auto_approval=False ajoute ToolApprovalMiddleware par défaut. L’intergiciel nécessite la même AgentSession opération dans les allers-retours d’approbation, met en file d’attente plusieurs demandes, applique les approbations permanentes des réponses utilisateur antérieures et évalue auto_approval_rules avant de renvoyer une demande à l’appelant. Avec auto_approval_rules=None, aucun rappel heuristique approuve automatiquement un appel.

La définition disable_tool_auto_approval=True supprime ce middleware, mais il ne modifie pas les données de approval_model’outil ; le flux d’approbation normal result.user_input_requests s’applique toujours.

Un harnais Go empaqueté n’est actuellement pas disponible. Habillage des outils requis pour approbation avec tool.ApprovalRequiredFunc et composez directement le middleware d’approbation.