Verwendung von Funktionstools mit Mensch-in-der-Schleife-Genehmigungen

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

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 AIProjectClient(
    new Uri("<your-foundry-project-endpoint>"),
    new DefaultAzureCredential())
     .AsAIAgent(
        model: "gpt-4o-mini",
        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 ToolApprovalRequestContent 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 toolApprovalRequests = response.Messages
    .SelectMany(x => x.Contents)
    .OfType<ToolApprovalRequestContent>()
    .ToList();

Wenn Funktionsgenehmigungsanforderungen vorhanden sind, steht der Funktionsaufruf einschließlich name und Argumente in der ToolCall Eigenschaft der ToolApprovalRequestContent Instanz zur Verfügung. 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.

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

Nachdem der Benutzer seine Eingabe übermittelt hat, verwenden Sie die Methode CreateResponse auf ToolApprovalRequestContent, um die Genehmigungsantwort zu 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 ToolApprovalRequestContent-Instanzen in der Antwort zu suchen, bis alle Funktionsaufrufe genehmigt oder abgelehnt wurden.

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

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

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

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:
            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)

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

Toolgenehmigung

Sie können eine menschliche Bestätigung verlangen, bevor ein Tool ausgeführt wird, indem Sie es in tool.ApprovalRequiredFunc einschließen:

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

Wenn das Modell den Aufruf eines Tools anfordert, fängt das Framework diesen ab und wartet vor der Ausführung auf die Freigabe. Der Genehmigungsfluss wird über Middleware verarbeitet.

Umschließen Sie Funktionen, die eine Genehmigung erfordern, mit ApprovalRequiredAIFunction, und fügen Sie sie dann über HarnessAgentOptions.ChatOptions.Tools hinzu:

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 ist standardmäßig false, daher fügt das Harness ToolApprovalAgent hinzu. Mit der Standardeinstellung ToolApprovalAgentOptionswerden keine heuristischen Regeln konfiguriert. Nicht übereinstimmende ToolApprovalRequestContent Elemente werden weiterhin zur Genehmigung an den Aufrufer zurückgegeben. Um vertrauenswürdige Rückrufe für die automatische Genehmigung hinzuzufügen, legen Sie folgendes fest ToolApprovalAgentOptions.AutoApprovalRules.

Durch das Festlegen von DisableToolAutoApproval = true wird nur die Middleware zur automatischen Genehmigung auf Basis von Dauerregeln, Warteschlangen und Heuristiken entfernt. Dadurch wird die Genehmigungspflicht für ein ApprovalRequiredAIFunction nicht aufgehoben. Genehmigungsantwortbindung und Umgehung von Tools, die keine Genehmigung erfordern, bleiben ebenfalls standardmäßig aktiviert; ihre separaten Opt-Outs sind DisableApprovalResponseBinding und DisableApprovalNotRequiredFunctionBypassing.

Markieren Sie das Tool mit approval_mode="always_require" und übergeben Sie es an 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 fügt standardmäßig hinzu ToolApprovalMiddleware . Die Middleware benötigt dasselbe AgentSession über mehrere Genehmigungszyklen hinweg, reiht mehrere Anfragen in die Warteschlange ein, wendet bereits vorliegende Genehmigungen aus früheren Antworten des Benutzers an und wertet auto_approval_rules aus, bevor sie eine Anfrage an den Aufrufer zurückgibt. Mit auto_approval_rules=None genehmigt kein heuristischer Rückruf einen Anruf automatisch.

Durch das Setzen von disable_tool_auto_approval=True wird diese Harness-Middleware entfernt, aber die approval_mode des Tools wird dadurch nicht geändert; der normale result.user_input_requests-Freigabeprozess gilt weiterhin.

Ein verpacktes Go-Gurtzeug ist derzeit nicht verfügbar. Umschließen Sie genehmigungspflichtige Tools mit tool.ApprovalRequiredFunc und stellen Sie die Genehmigungs-Middleware direkt zusammen.