Döngü onaylarında işlev araçlarını insanla kullanma

Bu öğretici adımında, Azure OpenAI Sohbet Tamamlama hizmeti üzerinde oluşturulmuş bir ajan ile insan onayı gerektiren fonksiyon araçlarının nasıl kullanılacağı gösterilir.

Aracılar, örneğin bir işlev çağrısını onaylamak için herhangi bir kullanıcı girdisi gerektirdiğinde, bu insan katılımı döngüsü örüntüsü olarak adlandırılır. Kullanıcı girişi gerektiren bir aracı çalıştırmak, son bir yanıtla tamamlanmak yerine, kullanıcıdan hangi girişin gerekli olduğunu belirten bir yanıtla tamamlanır. Aracı çağıran kişi, kullanıcıdan gerekli girdiği almaktan ve bunu yeni bir agent çalıştırmasının parçası olarak aracıya geri iletmekten sorumludur.

Önkoşullar

Önkoşullar ve NuGet paketlerini yükleme için bu öğreticideki Basit bir aracı oluşturma ve çalıştırma adımına bakın.

İşlev araçlarıyla aracı oluşturma

İşlevleri kullanırken, her işlev için yürütülmeden önce insan onayı gerekip gerekmediğini belirtmek mümkündür. Bu, bir AIFunction örneği ApprovalRequiredAIFunction örneğinin içinde sarmalayarak yapılır.

Burada, belirli bir konumun hava durumunu alma sahtesini gösteren basit bir işlev aracı örneği verilmiştir.

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

Önce bir oluşturmak ve ardından içinde sarmalamak için şunları yapabilirsiniz:

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

Aracıyı oluştururken, artık AsAIAgent yöntemine bir araç listesi geçirerek, onay gerektiren bir işlev aracını aracıya sağlayabilirsiniz.

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]);

Artık onay gerektiren bir işleviniz olduğundan aracı, işlevi doğrudan yürütmek ve sonucu döndürmek yerine bir onay isteğiyle yanıt verebilir. Herhangi bir FunctionApprovalRequestContent örneğin yanıt içeriğini denetleyebilirsiniz. Bu, aracının bir işlev için kullanıcı onayı gerektirdiğini gösterir.

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

Herhangi bir işlev onay isteği varsa, işlev çağrısının ad ve bağımsız değişkenler de dahil olmak üzere ayrıntıları örnekteki özelliğinde FunctionCallFunctionApprovalRequestContent bulunabilir. Bu, kullanıcıya gösterilebilir, böylece işlev çağrısını onaylamaya veya reddetmeye karar verebilir. Bu örnekte bir istek olduğunu varsayalım.

FunctionApprovalRequestContent requestContent = functionApprovalRequests.First();
Console.WriteLine($"We require approval to execute '{requestContent.FunctionCall.Name}'");

Kullanıcı girdisini sağladıktan sonra, FunctionApprovalResponseContent yöntemini CreateResponse üzerinde kullanarak bir FunctionApprovalRequestContent örneği oluşturabilirsiniz. true işlev çağrısını onaylamak için, veya false reddetmek için geçirin.

Yanıt içeriği, sonuçları geri almak için aynı oturum nesnesiyle birlikte yeni bir UserChatMessage içinde aracıya geçirilebilir.

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

İşlev araçlarını insan onaylarıyla birlikte kullandığınızda, her ajanın çalıştırılmasından sonra, tüm işlev çağrıları onaylanana veya reddedilene kadar, yanıttaki FunctionApprovalRequestContent örnekleri denetlemeyi unutmayın.

Tip

Çalıştırılabilir örneklerin tamamı için .NET örneklerine bakın.

Bu adımda, bir aracın yardımıyla insan onayı gereksinimi duyan fonksiyon araçlarının nasıl kullanılacağı gösterilir.

Aracılar, örneğin bir işlev çağrısını onaylamak için herhangi bir kullanıcı girdisi gerektirdiğinde, bu insan katılımı döngüsü örüntüsü olarak adlandırılır. Kullanıcı girişi gerektiren bir aracı çalıştırmak, son bir yanıtla tamamlanmak yerine, kullanıcıdan hangi girişin gerekli olduğunu belirten bir yanıtla tamamlanır. Aracı çağıran kişi, kullanıcıdan gerekli girdiği almaktan ve bunu yeni bir agent çalıştırmasının parçası olarak aracıya geri iletmekten sorumludur.

Önkoşullar

Önkoşullar ve Python paketlerini yükleme için bu öğreticideki Basit bir aracı oluşturma ve çalıştırma adımına bakın.

Onay gerektiren işlev araçlarıyla aracı oluşturma

İşlevleri kullanırken, her işlev için yürütülmeden önce insan onayı gerekip gerekmediğini belirtmek mümkündür. Bu işlem, approval_mode dekoratörü kullanılırken "always_require" parametresini @tool olarak ayarlayarak gerçekleştirilir.

Burada, belirli bir konumun hava durumunu alma sahtesini gösteren basit bir işlev aracı örneği verilmiştir.

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

Onay gerektiren bir işlev oluşturmak için parametresini approval_mode kullanabilirsiniz:

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

Aracıyı oluştururken, artık oluşturucuya bir araç listesi geçirerek Agent onay gerektiren işlev aracını araca sağlayabilirsiniz.

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

Artık onay gerektiren bir işleviniz olduğundan aracı, işlevi doğrudan yürütmek ve sonucu döndürmek yerine bir onay isteğiyle yanıt verebilir. Herhangi bir kullanıcı giriş isteği için yanıtı denetleyebilirsiniz. Bu, aracının bir işlev için kullanıcı onayı gerektirdiğini gösterir.

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

Herhangi bir işlev onayı isteği varsa, işlev çağrısının ad ve bağımsız değişkenler de dahil olmak üzere ayrıntıları kullanıcı giriş isteğindeki function_call özelliğinde bulunabilir. Bu, kullanıcıya gösterilebilir, böylece işlev çağrısını onaylamaya veya reddetmeye karar verebilir.

Kullanıcı girdisini sağladıktan sonra, kullanıcı giriş isteğindeki create_response yöntemini kullanarak bir yanıt oluşturabilirsiniz. True işlev çağrısını onaylamak için, veya False reddetmek için geçirin.

Ardından, yanıt, sonucu aracıdan geri almak için yeni bir Message içinde aracıya iletilebilir.

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)

Bir döngü içinde onayları işleme

Onay gerektiren birden çok işlev çağrısıyla çalışırken, tüm işlevler onaylanana veya reddedilene kadar döngüdeki onayları işlemeniz gerekebilir:

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)

İnsan müdahalesi gerekli olduğunda işlev araçlarını kullanırken, her araç çalıştırıldığında, tüm işlev çağrıları onaylanana veya reddedilene kadar yanıt üzerinde kullanıcı giriş isteklerini kontrol etmeyi unutmayın.

Tam örnek

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

Sonraki Adımlar

Aracılarla Yapılandırılmış Çıkış Üretme