Remarque
L’accès à cette page nécessite une autorisation. Vous pouvez essayer de vous connecter ou de modifier des répertoires.
L’accès à cette page nécessite une autorisation. Vous pouvez essayer de modifier des répertoires.
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.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.";
Pour créer un AIFunction wrapper et l’encapsuler 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 AzureOpenAIClient(
new Uri("https://<myresource>.openai.azure.com"),
new AzureCliCredential())
.GetChatClient("gpt-4o-mini")
.AsAIAgent(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 FunctionApprovalRequestContent 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 functionApprovalRequests = response.Messages
.SelectMany(x => x.Contents)
.OfType<FunctionApprovalRequestContent>()
.ToList();
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 FunctionCall propriété sur l’instance FunctionApprovalRequestContent .
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.
FunctionApprovalRequestContent requestContent = functionApprovalRequests.First();
Console.WriteLine($"We require approval to execute '{requestContent.FunctionCall.Name}'");
Une fois que l’utilisateur a fourni son entrée, vous pouvez créer une FunctionApprovalResponseContent instance à l’aide de la CreateResponse méthode sur le FunctionApprovalRequestContent.
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 FunctionApprovalRequestContent 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.
Conseil / Astuce
Consultez les exemples .NET pour obtenir des exemples exécutables complets.
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 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
É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:
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 create_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 envoyée à 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.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)
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:
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)
Chaque fois que vous utilisez des outils de fonction avec des approbations avec intervention humaine dans la boucle, assurez-vous de vérifier les demandes d'entrée utilisateur dans la réponse, après chaque exécution d'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 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())