Jegyzet
Az oldalhoz való hozzáférés engedélyezést igényel. Próbálhatod be jelentkezni vagy könyvtárat váltani.
Az oldalhoz való hozzáférés engedélyezést igényel. Megpróbálhatod a könyvtár váltását.
Ez az oktatóanyag lépés bemutatja, hogyan használhat olyan függvényeszközöket, amelyek emberi jóváhagyást igényelnek egy ügynök segítségével, ahol az ügynök az Azure OpenAI Chat Completion szolgáltatásra épül.
Ha az ügynökök bármilyen felhasználói bemenetet igényelnek, például egy függvényhívás jóváhagyásához, akkor ezt ember által vezérelt mintának nevezzük. A felhasználói bemenetet igénylő ügynökfuttatás egy válaszsal fejeződik be, amely jelzi, hogy milyen bemenetre van szükség a felhasználótól, ahelyett, hogy végleges választ adna. Ezután az ügynök hívója felelős a szükséges bemenetek begyűjtéséért a felhasználótól, majd ezeket az ügynöknek adja vissza egy új ügynökfolyamat részeként.
Előfeltételek
Az előfeltételekről és a NuGet-csomagok telepítéséről lásd az egyszerű ügynök létrehozása és futtatása lépést ebben az oktatóanyagban.
Az ügynök létrehozása funkcióeszközökkel
Függvények használatakor minden egyes függvény esetében jelezhető, hogy a végrehajtás előtt emberi jóváhagyásra van-e szükség.
Ez úgy történik, hogy a AIFunction példányt egy ApprovalRequiredAIFunction példányba ágyazzuk.
Íme egy példa egy egyszerű függvényeszközre, amely meghamisít egy adott hely időjárását.
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.";
Ha létre szeretne hozni egy AIFunction elemet, majd be szeretné tördelni egybe ApprovalRequiredAIFunction, az alábbiakat teheti:
AIFunction weatherFunction = AIFunctionFactory.Create(GetWeather);
AIFunction approvalRequiredWeatherFunction = new ApprovalRequiredAIFunction(weatherFunction);
Az ügynök létrehozásakor most már biztosíthatja az ügynök számára a jóváhagyást igénylő függvényeszközt, ha átad egy eszközlistát a AsAIAgent metódusnak.
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]);
Mivel most már rendelkezik egy jóváhagyást igénylő függvénnyel, az ügynök ahelyett, hogy közvetlenül hajtja végre a függvényt, és visszaadja az eredményt, egy jóváhagyási kéréssel válaszolhat.
A válasz tartalmát bármely FunctionApprovalRequestContent példány esetében ellenőrizheti, ami azt jelzi, hogy az ügynöknek felhasználói jóváhagyásra van szüksége egy függvényhez.
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();
Ha vannak függvény-jóváhagyási kérelmek, a függvényhívás részletei, beleértve a nevet és az argumentumokat, megtalálhatók a FunctionCall példány tulajdonságában FunctionApprovalRequestContent .
Ez megjeleníthető a felhasználó számára, hogy eldönthesse, jóváhagyja vagy elutasítja-e a függvényhívást.
Ebben a példában tegyük fel, hogy van egy kérés.
FunctionApprovalRequestContent requestContent = functionApprovalRequests.First();
Console.WriteLine($"We require approval to execute '{requestContent.FunctionCall.Name}'");
Miután a felhasználó megadta a bemenetét, létrehozhat egy FunctionApprovalResponseContent példányt a CreateResponse metódus használatával a FunctionApprovalRequestContent-on.
A függvényhívás jóváhagyásához adja meg a true értéket, vagy az elutasításhoz a false értéket.
A válasz tartalma ezután átadható az ügynöknek egy új UserChatMessage részeként, ugyanazzal a munkamenet objektummal együtt, hogy az eredmény visszakerüljön az ügynöktől.
var approvalMessage = new ChatMessage(ChatRole.User, [requestContent.CreateResponse(true)]);
Console.WriteLine(await agent.RunAsync(approvalMessage, session));
Amikor függvényeszközöket használ emberi felülvizsgálattal történő jóváhagyásokkal, ne felejtse el ellenőrizni a válaszban szereplő FunctionApprovalRequestContent példányokat minden ügynöki futtatást követően, amíg az összes függvényhívást jóvá nem hagyják vagy el nem utasítják.
Jótanács
A teljes futtatható példákért tekintse meg a .NET-mintákat .
Ez az oktatóanyag lépés bemutatja, hogyan használhat olyan funkció eszközöket, amelyek emberi jóváhagyást igényelnek egy ügynök segítségével.
Ha az ügynökök bármilyen felhasználói bemenetet igényelnek, például egy függvényhívás jóváhagyásához, akkor ezt ember által vezérelt mintának nevezzük. A felhasználói bemenetet igénylő ügynökfuttatás egy válaszsal fejeződik be, amely jelzi, hogy milyen bemenetre van szükség a felhasználótól, ahelyett, hogy végleges választ adna. Ezután az ügynök hívója felelős a szükséges bemenetek begyűjtéséért a felhasználótól, majd ezeket az ügynöknek adja vissza egy új ügynökfolyamat részeként.
Előfeltételek
Az előfeltételekről és a Python-csomagok telepítéséről lásd a Hozzon létre és futtasson egy egyszerű ügynököt lépést ebben az oktatóanyagban.
Az ügynök létrehozása jóváhagyást igénylő funkcionális eszközökkel.
Függvények használatakor minden egyes függvény esetében jelezhető, hogy a végrehajtás előtt emberi jóváhagyásra van-e szükség.
Ez úgy történik, hogy a approval_mode paramétert "always_require" állítja be, amikor a @tool dekorátort használja.
Íme egy példa egy egyszerű függvényeszközre, amely meghamisít egy adott hely időjárását.
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."
Jóváhagyást igénylő függvény létrehozásához használja a következő paramétert 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%."
Az ügynök létrehozásakor most már megadhatja az ügynöknek a jóváhagyást igénylő funkció eszközt, egy lista átadásával a Agent konstruktornak.
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
Mivel most már rendelkezik egy jóváhagyást igénylő függvénnyel, az ügynök ahelyett, hogy közvetlenül hajtja végre a függvényt, és visszaadja az eredményt, egy jóváhagyási kéréssel válaszolhat. Minden felhasználói bemeneti kérés esetében ellenőrizheti a választ, ami azt jelzi, hogy az ügynöknek felhasználói jóváhagyásra van szüksége egy függvényhez.
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}")
Ha vannak függvény-jóváhagyási kérések, a függvényhívás részletei, beleértve a nevet és az argumentumokat, megtalálhatók a function_call felhasználói beviteli kérelem tulajdonságában.
Ez megjeleníthető a felhasználó számára, hogy eldönthesse, jóváhagyja vagy elutasítja-e a függvényhívást.
Miután a felhasználó megadta a bemenetét, a felhasználói beviteli create_response kérelem metódusával hozhat létre választ.
A függvényhívás jóváhagyásához adja meg a True értéket, vagy az elutasításhoz a False értéket.
A válasz ezután átadható az ügynöknek egy új Messagefájlban, hogy az eredmény visszakerüljön az ügynöktől.
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)
Jóváhagyások kezelése ciklusban
Ha több, jóváhagyást igénylő függvényhívással dolgozik, előfordulhat, hogy a jóváhagyásokat egy ciklusban kell kezelnie, amíg az összes függvényt nem hagyja jóvá vagy nem utasítja el:
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)
Amikor olyan függvényeszközöket használ, amelyek emberi felügyeletet igényelnek a jóváhagyások során, ne felejtse el ellenőrizni a felhasználói bemeneti kéréseket a válaszban, minden modul futtatása után, amíg az összes függvényhívást jóvá nem hagyják vagy el nem utasítják.
Teljes példa
# 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())