استخدام أدوات الدالة مع الإنسان في الموافقات الحلقية

توضح لك خطوة البرنامج التعليمي هذه كيفية استخدام أدوات الوظائف التي تتطلب موافقة بشرية مع وكيل، حيث يتم بناء العامل على خدمة إكمال دردشة OpenAI Azure.

عندما يطلب الوكلاء أي إدخال مستخدم، على سبيل المثال للموافقة على استدعاء دالة، يشار إلى هذا على أنه نمط بشري في التكرار الحلقي. سيكتمل تشغيل العامل الذي يتطلب إدخال المستخدم، مع استجابة تشير إلى الإدخال المطلوب من المستخدم، بدلا من إكماله بإجابة نهائية. ثم يكون المتصل بالعامل مسؤولا عن الحصول على الإدخال المطلوب من المستخدم، وتمريره مرة أخرى إلى العامل كجزء من تشغيل عامل جديد.

المتطلبات الأساسية

للحصول على المتطلبات الأساسية وتثبيت حزم NuGet، راجع خطوة إنشاء وتشغيل عامل بسيط في هذا البرنامج التعليمي.

إنشاء العامل باستخدام أدوات الدالة

عند استخدام الدالات، من الممكن الإشارة إلى كل دالة، ما إذا كانت تتطلب موافقة بشرية قبل تنفيذها. يتم ذلك عن طريق التفاف المثيل AIFunction في مثيل ApprovalRequiredAIFunction .

هنا مثال على أداة وظيفة بسيطة مزيفة الحصول على الطقس لموقع معين.

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

لإنشاء AIFunction ثم تضمينه في ApprovalRequiredAIFunction، يمكنك القيام بما يلي:

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

عند إنشاء العامل، يمكنك الآن توفير أداة الوظيفة التي تتطلب الموافقة للعامل، عن طريق تمرير قائمة الأدوات إلى AsAIAgent الأسلوب .

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

تحذير

DefaultAzureCredential مناسب للتنمية ولكنه يتطلب دراسة متأنية في الإنتاج. في الإنتاج، ضع في اعتبارك استخدام بيانات اعتماد محددة (على سبيل المثال، ManagedIdentityCredential) لتجنب مشكلات زمن الانتقال، وبحث بيانات الاعتماد غير المقصودة، والمخاطر الأمنية المحتملة من الآليات الاحتياطية.

نظرا لأن لديك الآن دالة تتطلب الموافقة، فقد يستجيب العامل بطلب للموافقة بدلا من تنفيذ الدالة مباشرة وإرجاع النتيجة. يمكنك التحقق من محتوى الاستجابة لأي ToolApprovalRequestContent مثيلات، ما يشير إلى أن العامل يتطلب موافقة المستخدم على دالة.

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

إذا كانت هناك أي طلبات الموافقة على الدالة، فإن استدعاء الدالة بما في ذلك اسمها ووسيطاتها متاح من الخاصية ToolCall على المثيل ToolApprovalRequestContent . يمكن عرض هذا للمستخدم، بحيث يمكنه تحديد ما إذا كان سيتم الموافقة على استدعاء الدالة أو رفضه. على سبيل المثال، افترض أن هناك طلبا واحدا.

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

بمجرد أن يقدم المستخدم إدخاله، استخدم CreateResponse الأسلوب على ToolApprovalRequestContent لإنشاء استجابة الموافقة. مرر true للموافقة على استدعاء الدالة، أو false لرفضه.

يمكن بعد ذلك تمرير محتوى الاستجابة إلى العامل في جديد UserChatMessage، جنبا إلى جنب مع نفس كائن جلسة العمل للحصول على النتيجة مرة أخرى من العامل.

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

كلما كنت تستخدم أدوات الدالة مع الإنسان في الموافقات الحلقية، تذكر التحقق من ToolApprovalRequestContent وجود مثيلات في الاستجابة، بعد تشغيل كل عامل، حتى تتم الموافقة على جميع استدعاءات الدالة أو رفضها.

Tip

راجع الخطوة 01 .NET Agents: استخدام أدوات الدالة مع نموذج الموافقات للحصول على مثال كامل وقابل للتشغيل.

توضح لك خطوة البرنامج التعليمي هذه كيفية استخدام أدوات الوظيفة التي تتطلب موافقة بشرية مع عامل.

عندما يطلب الوكلاء أي إدخال مستخدم، على سبيل المثال للموافقة على استدعاء دالة، يشار إلى هذا على أنه نمط بشري في التكرار الحلقي. سيكتمل تشغيل العامل الذي يتطلب إدخال المستخدم، مع استجابة تشير إلى الإدخال المطلوب من المستخدم، بدلا من إكماله بإجابة نهائية. ثم يكون المتصل بالعامل مسؤولا عن الحصول على الإدخال المطلوب من المستخدم، وتمريره مرة أخرى إلى العامل كجزء من تشغيل عامل جديد.

المتطلبات الأساسية

للحصول على المتطلبات الأساسية وتثبيت حزم Python، راجع خطوة إنشاء وتشغيل عامل بسيط في هذا البرنامج التعليمي.

إنشاء العامل باستخدام أدوات الوظيفة التي تتطلب الموافقة

عند استخدام الدالات، من الممكن الإشارة إلى كل دالة، ما إذا كانت تتطلب موافقة بشرية قبل تنفيذها. يتم ذلك عن طريق تعيين المعلمة approval_mode إلى "always_require" عند استخدام @tool مصمم الديكور.

هنا مثال على أداة وظيفة بسيطة مزيفة الحصول على الطقس لموقع معين.

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

لإنشاء دالة تتطلب الموافقة، يمكنك استخدام المعلمة 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%."

عند إنشاء العامل، يمكنك الآن توفير أداة الوظيفة التي تتطلب الموافقة للعامل، عن طريق تمرير قائمة الأدوات إلى الدالة Agent الإنشائية.

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

نظرا لأن لديك الآن دالة تتطلب الموافقة، فقد يستجيب العامل بطلب للموافقة بدلا من تنفيذ الدالة مباشرة وإرجاع النتيجة. يمكنك التحقق من الاستجابة لأي طلبات إدخال مستخدم، ما يشير إلى أن العامل يتطلب موافقة المستخدم على وظيفة.

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

إذا كانت هناك أي طلبات الموافقة على الدالة، يمكن العثور على تفاصيل استدعاء الدالة بما في ذلك الاسم والوسيطات في الخاصية function_call على طلب إدخال المستخدم. يمكن عرض هذا للمستخدم، بحيث يمكنه تحديد ما إذا كان سيتم الموافقة على استدعاء الدالة أو رفضه.

بمجرد أن يقدم المستخدم إدخاله، يمكنك إنشاء استجابة باستخدام to_function_approval_response الأسلوب على طلب إدخال المستخدم. مرر True للموافقة على استدعاء الدالة، أو False لرفضه.

يمكن بعد ذلك تمرير الاستجابة إلى العامل في جديد Message، للحصول على النتيجة مرة أخرى من العامل.

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)

معالجة الموافقات في حلقة

عند العمل مع استدعاءات دالات متعددة تتطلب الموافقة، قد تحتاج إلى معالجة الموافقات في حلقة حتى تتم الموافقة على جميع الدالات أو رفضها:

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)

كلما كنت تستخدم أدوات الدالة مع الإنسان في الموافقات التكرارية، تذكر التحقق من طلبات إدخال المستخدم في الاستجابة، بعد تشغيل كل عامل، حتى تتم الموافقة على جميع استدعاءات الدالة أو رفضها.

مثال كامل

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

الموافقة على الأداة

يمكنك طلب موافقة بشرية قبل تنفيذ أداة عن طريق التفافها مع tool.ApprovalRequiredFunc:

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

عندما يطلب النموذج استدعاء أداة، يعترضه إطار العمل وينتظر الموافقة قبل التنفيذ. تتم معالجة تدفق الموافقة من خلال البرامج الوسيطة.

استخدام الموافقة على الأداة مع Harness Agent

يتطلب التكوين العادي/اليدوي أداة مميزة بالموافقة وحلقة الموافقة على الاستجابة. يستخدم عامل Harness نفس الأدوات التي وضعت عليها علامة الموافقة ومحتوى الاستجابة، ولكنه يقوم أيضا بتثبيت البرامج الوسيطة للطلبات المدرجة في قائمة الانتظار، وقواعد "الموافقة دائما" الدائمة، والموافقة التلقائية الإختيارية.

التفاف الدالات التي تتطلب الموافقة في ApprovalRequiredAIFunction، ثم إضافتها من خلال HarnessAgentOptions.ChatOptions.Tools:

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 الافتراضيات إلى false، لذلك يضيف ToolApprovalAgentالتسخير . مع الإعداد الافتراضي ToolApprovalAgentOptions، لا يتم تكوين أي قواعد توجيهية؛ لا تزال العناصر غير المتطابقة ToolApprovalRequestContent تعود إلى المتصل للموافقة عليها. لإضافة عمليات رد اتصال الموافقة التلقائية الموثوق بها، قم بتعيين ToolApprovalAgentOptions.AutoApprovalRules.

يزيل الإعداد DisableToolAutoApproval = true فقط القاعدة الدائمة، والاعتماد في قائمة الانتظار، والبرامج الوسيطة للموافقة التلقائية الموجهة. لا يزيل متطلبات الموافقة من ApprovalRequiredAIFunction. كما يظل ربط الموافقة على الاستجابة وتجاوز الأدوات التي لا تتطلب الموافقة ممكنا بشكل افتراضي؛ عمليات إلغاء الاشتراك المنفصلة الخاصة بهم هي DisableApprovalResponseBinding و DisableApprovalNotRequiredFunctionBypassing.

وضع علامة على الأداة مع approval_mode="always_require" وتمريرها إلى 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 ToolApprovalMiddleware يضيف بشكل افتراضي. يتطلب البرنامج الوسيط نفس الشيء AgentSession عبر الموافقة ذهابا وجولة، ويصف طلبات متعددة، ويطبق الموافقات الدائمة من استجابات المستخدم السابقة، ويقيم auto_approval_rules قبل إرجاع طلب إلى المتصل. مع auto_approval_rules=None، لا يوافق رد الاتصال الاستدلالي تلقائيا على مكالمة.

يزيل الإعداد disable_tool_auto_approval=True الذي يسخر البرامج الوسيطة، ولكنه لا يغير الأداة approval_mode؛ لا يزال تدفق الموافقة العادي result.user_input_requests ساريا.

لا يتوفر حاليا تسخير Go المحزم. قم بتضمين الأدوات المطلوبة للموافقة مع tool.ApprovalRequiredFunc البرنامج الوسيط للموافقة وإنشاءه مباشرة.

الخطوات التالية

مترجم شفرة