Hinweis
Für den Zugriff auf diese Seite ist eine Autorisierung erforderlich. Sie können versuchen, sich anzumelden oder das Verzeichnis zu wechseln.
Für den Zugriff auf diese Seite ist eine Autorisierung erforderlich. Sie können versuchen, das Verzeichnis zu wechseln.
In diesem Schritt des Tutorials wird gezeigt, wie Sie Funktionstools verwenden, die eine menschliche Genehmigung durch einen Agenten erfordern, bei dem der Agent auf dem Azure OpenAI Chat Completion Service basiert.
Wenn Agents eine Benutzereingabe erfordern, z. B. zum Genehmigen eines Funktionsaufrufs, wird dies als "Human-in-the-Loop"-Muster bezeichnet. Ein Agentenlauf, der eine Benutzereingabe erfordert, liefert eine Antwort, die angibt, welche Eingabe vom Benutzer erforderlich ist, anstatt mit einer endgültigen Antwort abzuschließen. Der Anrufer des Agents ist dann dafür verantwortlich, die erforderlichen Eingaben vom Benutzer zu erhalten und sie im Rahmen eines neuen Agentendurchlaufs an den Agenten zurückzugeben.
Voraussetzungen
Für Voraussetzungen und die Installation von NuGet-Paketen siehe den Schritt Erstellen und Ausführen eines einfachen Agenten in diesem Tutorial.
Erstellen Sie den Agenten mit Funktionstools
Bei der Verwendung von Funktionen ist es möglich, für jede Funktion anzugeben, ob sie eine menschliche Genehmigung erfordert, bevor sie ausgeführt wird.
Dazu wird die AIFunction Instanz in eine ApprovalRequiredAIFunction Instanz umschlossen.
Hier ist ein Beispiel für ein einfaches Funktionstool, das das Wetter für einen bestimmten Ort fälscht.
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.";
Gehen Sie wie folgt vor, um eine Datei AIFunction zu erstellen und dann in eine ApprovalRequiredAIFunctionumzuschließen:
AIFunction weatherFunction = AIFunctionFactory.Create(GetWeather);
AIFunction approvalRequiredWeatherFunction = new ApprovalRequiredAIFunction(weatherFunction);
Beim Erstellen des Agents können Sie jetzt dem Agenten ein Funktionstool bereitstellen, das eine Genehmigung erfordert, indem Sie eine Liste von Tools an die AsAIAgent-Methode übergeben.
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]);
Da Sie jetzt über eine Funktion verfügen, die eine Genehmigung erfordert, antwortet der Agent möglicherweise mit einer Genehmigungsanforderung, anstatt die Funktion direkt auszuführen und das Ergebnis zurückzugeben.
Sie können den Antwortinhalt für alle FunctionApprovalRequestContent Instanzen überprüfen, was angibt, dass der Agent eine Benutzergenehmigung für eine Funktion erfordert.
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();
Wenn Funktionsgenehmigungsanforderungen vorhanden sind, finden Sie die Details des Funktionsaufrufs einschließlich Name und Argumente in der FunctionCall Eigenschaft der FunctionApprovalRequestContent Instanz.
Dies kann dem Benutzer angezeigt werden, damit er entscheiden kann, ob der Funktionsaufruf genehmigt oder abgelehnt werden soll.
Gehen Sie in diesem Beispiel davon aus, dass eine Anforderung vorhanden ist.
FunctionApprovalRequestContent requestContent = functionApprovalRequests.First();
Console.WriteLine($"We require approval to execute '{requestContent.FunctionCall.Name}'");
Nachdem der Benutzer seine Eingabe bereitgestellt hat, können Sie eine FunctionApprovalResponseContent Instanz mithilfe der CreateResponse Methode auf dem/der FunctionApprovalRequestContent erstellen.
Übergeben Sie true, um den Funktionsaufruf zu genehmigen, oder false, um ihn abzulehnen.
Der Antwortinhalt kann dann zusammen mit demselben Sitzungsobjekt an den Agenten UserChatMessage übergeben werden, um das Ergebnis vom Agenten zu erhalten.
var approvalMessage = new ChatMessage(ChatRole.User, [requestContent.CreateResponse(true)]);
Console.WriteLine(await agent.RunAsync(approvalMessage, session));
Wenn Sie Funktionstools mit menschlichen Genehmigungen im Prozess verwenden, denken Sie daran, nach jedem Agentenlauf nach FunctionApprovalRequestContent-Instanzen in der Antwort zu suchen, bis alle Funktionsaufrufe genehmigt oder abgelehnt wurden.
Tipp
Vollständige Runnable-Beispiele finden Sie in den .NET-Beispielen .
In diesem Schritt des Tutorials erfahren Sie, wie Sie Funktionen verwenden, die eine menschliche Genehmigung durch einen Agenten erfordern.
Wenn Agents eine Benutzereingabe erfordern, z. B. zum Genehmigen eines Funktionsaufrufs, wird dies als "Human-in-the-Loop"-Muster bezeichnet. Ein Agentenlauf, der eine Benutzereingabe erfordert, liefert eine Antwort, die angibt, welche Eingabe vom Benutzer erforderlich ist, anstatt mit einer endgültigen Antwort abzuschließen. Der Anrufer des Agents ist dann dafür verantwortlich, die erforderlichen Eingaben vom Benutzer zu erhalten und sie im Rahmen eines neuen Agentendurchlaufs an den Agenten zurückzugeben.
Voraussetzungen
Die erforderlichen Komponenten und die Installation von Python-Paketen finden Sie im Abschnitt zum Erstellen und Ausführen eines einfachen Agent-Schritts in diesem Lernprogramm.
Erstellen des Agents mit Funktionstools, die eine Genehmigung erfordern
Bei der Verwendung von Funktionen ist es möglich, für jede Funktion anzugeben, ob sie eine menschliche Genehmigung erfordert, bevor sie ausgeführt wird.
Dies geschieht durch Setzen des approval_mode Parameters auf "always_require", wenn der @tool Dekorator verwendet wird.
Hier ist ein Beispiel für ein einfaches Funktionstool, das das Wetter für einen bestimmten Ort fälscht.
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."
Um eine Funktion zu erstellen, die eine Genehmigung erfordert, können Sie den approval_mode Parameter verwenden:
@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%."
Beim Erstellen des Agents können Sie nun das Genehmigung erfordernde Funktionstool bereitstellen, indem Sie dem Agent-Konstruktor eine Liste von Tools übergeben.
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
Da Sie jetzt über eine Funktion verfügen, die eine Genehmigung erfordert, antwortet der Agent möglicherweise mit einer Genehmigungsanforderung, anstatt die Funktion direkt auszuführen und das Ergebnis zurückzugeben. Sie können die Antwort auf alle Benutzereingabeanforderungen überprüfen, was angibt, dass der Agent eine Benutzergenehmigung für eine Funktion erfordert.
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}")
Wenn Funktionsgenehmigungsanforderungen vorhanden sind, finden Sie die Details des Funktionsaufrufs einschließlich Name und Argumente in der function_call Eigenschaft für die Benutzereingabeanforderung.
Dies kann dem Benutzer angezeigt werden, damit er entscheiden kann, ob der Funktionsaufruf genehmigt oder abgelehnt werden soll.
Nachdem der Benutzer seine Eingabe bereitgestellt hat, können Sie eine Antwort mithilfe der create_response Methode für die Benutzereingabeanforderung erstellen.
Übergeben Sie True, um den Funktionsaufruf zu genehmigen, oder False, um ihn abzulehnen.
Die Antwort kann dann an den Agent in einem neuen Messageübergeben werden, um das Ergebnis vom Agent zurück zu erhalten.
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)
Genehmigungen in einer Schleife bearbeiten
Wenn Sie mit mehreren Funktionsaufrufen arbeiten, die eine Genehmigung erfordern, müssen Sie möglicherweise Genehmigungen in einer Schleife verarbeiten, bis alle Funktionen genehmigt oder abgelehnt werden:
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)
Wenn Sie Funktionstools mit menschlichen Genehmigungen im Genehmigungsprozess verwenden, denken Sie daran, nach jedem Agentendurchlauf im Antwortfeld nach Benutzereingaben zu suchen, bis alle Funktionsaufrufe genehmigt oder abgelehnt wurden.
Vollständiges Beispiel
# 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())