本教學課程步驟示範如何使用需要人工核准的函式工具與代理程式搭配使用，其中代理程式是以 Azure OpenAI 聊天完成服務為基礎。
當代理需要使用者的任何輸入時，例如核准功能呼叫，這稱為人機互動模式。 需要使用者輸入的代理程式執行將以回應完成，指出使用者需要哪些輸入，而不是以最終答案完成。 然後，代理程式的呼叫者負責從使用者取得所需的輸入，並將其作為新代理程式執行的一部分傳遞回代理程式。
先決條件
如需必要條件和安裝 NuGet 套件，請參閱本教學課程中的 建立並執行簡單代理程式 步驟。
使用函數工具建立代理程式
使用函數時，可以為每個函數指出是否需要人工批准才能執行。
這可以透過將 AIFunction 實例包裝在 ApprovalRequiredAIFunction 實例中來完成。
這是一個簡單的函數工具的示例，該工具可以偽造獲取給定位置的天氣。
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.";
若要建立 ， AIFunction 然後將其包裝在 ApprovalRequiredAIFunction中，您可以執行下列動作：
AIFunction weatherFunction = AIFunctionFactory.Create(GetWeather);
AIFunction approvalRequiredWeatherFunction = new ApprovalRequiredAIFunction(weatherFunction);
建立代理程式時，您現在可以透過將工具清單傳遞給 AsAIAgent 方法，將需要核准的功能工具提供給代理程式。
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]);
由於您現在有一個需要核准的函數，因此代理程式可能會以核准請求來回應，而不是直接執行函數並傳回結果。
您可以檢查任何 FunctionApprovalRequestContent 執行個體的回應內容，這表示代理程式需要使用者核准才能使用函式。
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();
如果有任何函式核准要求，則可以在實例的FunctionCall屬性中找到FunctionApprovalRequestContent函式呼叫的詳細資料，包括名稱和引數。
這可以顯示給使用者，以便他們決定是否要核准或拒絕函數呼叫。
在此範例中，假設有一個要求。
FunctionApprovalRequestContent requestContent = functionApprovalRequests.First();
Console.WriteLine($"We require approval to execute '{requestContent.FunctionCall.Name}'");
使用者提供輸入後，您可以在FunctionApprovalResponseContent上使用CreateResponse 方法來建立FunctionApprovalRequestContent實例。
傳遞 true 以核准函式呼叫，或傳遞 false 以拒絕函式呼叫。
回應內容接著可以以新的 UserChatMessage，連同相同的會話物件一起傳遞給代理，從代理那裡取得結果。
var approvalMessage = new ChatMessage(ChatRole.User, [requestContent.CreateResponse(true)]);
Console.WriteLine(await agent.RunAsync(approvalMessage, session));
每當您使用功能工具搭配人為介入核准時，請記得在每次代理程式執行後，檢查回應中的 FunctionApprovalRequestContent 出現，直到所有函式呼叫都已核准或拒絕為止。
小提示
完整可執行範例請參閱 .NET 範例 。
本教學步驟示範如何使用需人工核准的函數工具與代理。
當代理需要使用者的任何輸入時，例如核准功能呼叫，這稱為人機互動模式。 需要使用者輸入的代理程式執行將以回應完成，指出使用者需要哪些輸入，而不是以最終答案完成。 然後，代理程式的呼叫者負責從使用者取得所需的輸入，並將其作為新代理程式執行的一部分傳遞回代理程式。
先決條件
如需必要條件和安裝 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 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
由於您現在有一個需要核准的函數，因此代理程式可能會以核准請求來回應，而不是直接執行函數並傳回結果。 您可以檢查任何使用者輸入要求的回應，這表示代理程式需要使用者核准才能使用功能。
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}")
如果有任何函式核准要求，則可以在使用者輸入要求的屬性中找到 function_call 函式呼叫的詳細資料，包括名稱和引數。
這可以顯示給使用者，以便他們決定是否要核准或拒絕函數呼叫。
使用者提供輸入後，您可以使用使用者輸入請求上的方法 create_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.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)
在迴圈中處理核准
當處理需要核准的多個函數呼叫時，您可能需要在循環中處理核准，直到所有函數都被核准或拒絕為止：
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)
每當您使用函數工具搭配人機迴圈核准時，請記得在每次代理程式執行之後，檢查回應中的使用者輸入要求，直到所有函數呼叫都已核准或拒絕為止。
完整範例
# 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())