本教程步骤介绍如何使用需要人工审批的函数工具，该代理是在 Azure OpenAI 聊天完成服务上构建的。
当代理需要任何用户输入（例如批准函数调用）时，这称为人机循环模式。 需要用户输入的代理运行完成时将提供说明用户需要输入哪些内容的响应，而不是提供最终答案。 然后，代理的调用方负责从用户获取所需的输入，并在新代理运行过程中将其传回代理。
Prerequisites
有关先决条件和安装 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]);
Warning
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}'");
用户提供输入后，使用 ToolApprovalRequestContent 上的 CreateResponse 方法创建审批响应。
传递 true 以批准函数调用，或 false 拒绝它。
然后，可以将响应内容与同一会话对象一起传递进新的UserChatMessage，以便从代理中获取结果。
var approvalMessage = new ChatMessage(ChatRole.User, [requestContent.CreateResponse(true)]);
Console.WriteLine(await agent.RunAsync(approvalMessage, session));
每当使用包含人类介入的审批流程的功能工具时，请记住在每次代理运行后检查 ToolApprovalRequestContent 响应中的实例，直到所有功能调用都获得批准或拒绝。
Tip
有关完整的可运行示例，请参阅 .NET 代理步骤 01：将函数工具与审批示例配合使用 。
本教程步骤展示如何将需要人工审批的函数工具与代理结合使用。
当代理需要任何用户输入（例如批准函数调用）时，这称为人机循环模式。 需要用户输入的代理运行完成时将提供说明用户需要输入哪些内容的响应，而不是提供最终答案。 然后，代理的调用方负责从用户获取所需的输入，并在新代理运行过程中将其传回代理。
Prerequisites
有关先决条件和安装 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},
},
})
当模型请求工具调用时，框架会截获它，并在执行前等待审批。 审批流通过中间件进行处理。
将工具审批与工具套件代理结合使用
纯手动合成需要带有审批标记的工具和审批响应闭环。 工具套件代理使用相同的标记为需要审批的工具和响应内容，但还会安装中间件，用于请求排队、长期有效的“始终批准”规则和可选的启发式自动审批。
将需要批准的函数包裹在 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 会移除该 harness 中间件，但不会改变该工具的 approval_mode；常规的 result.user_input_requests 审批流程仍然适用。
目前还没有可用的打包版 Go harness。 使用 tool.ApprovalRequiredFunc 封装需要审批的工具，并直接组合审批中间件。