使用干预处理器实现工具执行的用户审批

本教程展示如何通过干预处理器拦截工具执行，并提示用户授权执行工具。

from dataclasses import dataclass
from typing import Any, List

from autogen_core import (
    AgentId,
    AgentType,
    DefaultInterventionHandler,
    DropMessage,
    FunctionCall,
    MessageContext,
    RoutedAgent,
    SingleThreadedAgentRuntime,
    message_handler,
)
from autogen_core.models import (
    ChatCompletionClient,
    LLMMessage,
    SystemMessage,
    UserMessage,
)
from autogen_core.tool_agent import ToolAgent, ToolException, tool_agent_caller_loop
from autogen_core.tools import ToolSchema
from autogen_ext.code_executors.docker import DockerCommandLineCodeExecutor
from autogen_ext.models.openai import OpenAIChatCompletionClient
from autogen_ext.tools.code_execution import PythonCodeExecutionTool

让我们定义一个携带字符串内容的简单消息类型。

@dataclass
class Message:
    content: str

让我们创建一个能通过 ToolAgent 使用工具的简单工具代理。

class ToolUseAgent(RoutedAgent):
    """An agent that uses tools to perform tasks. It executes the tools
    by itself by sending the tool execution task to a ToolAgent."""

    def __init__(
        self,
        description: str,
        system_messages: List[SystemMessage],
        model_client: ChatCompletionClient,
        tool_schema: List[ToolSchema],
        tool_agent_type: AgentType,
    ) -> None:
        super().__init__(description)
        self._model_client = model_client
        self._system_messages = system_messages
        self._tool_schema = tool_schema
        self._tool_agent_id = AgentId(type=tool_agent_type, key=self.id.key)

    @message_handler
    async def handle_user_message(self, message: Message, ctx: MessageContext) -> Message:
        """Handle a user message, execute the model and tools, and returns the response."""
        session: List[LLMMessage] = [UserMessage(content=message.content, source="User")]
        # 使用工具代理执行工具，并获取输出消息。
        output_messages = await tool_agent_caller_loop(
            self,
            tool_agent_id=self._tool_agent_id,
            model_client=self._model_client,
            input_messages=session,
            tool_schema=self._tool_schema,
            cancellation_token=ctx.cancellation_token,
        )
        # 从输出消息中提取最终响应。
        final_response = output_messages[-1].content
        assert isinstance(final_response, str)
        return Message(content=final_response)

工具使用代理向工具代理发送工具调用请求以执行工具， 因此我们可以拦截工具使用代理发送给工具代理的消息， 在执行工具前提示用户获取执行权限。

让我们创建一个干预处理器来拦截这些消息， 并在允许工具执行前提示用户确认。

class ToolInterventionHandler(DefaultInterventionHandler):
    async def on_send(
        self, message: Any, *, message_context: MessageContext, recipient: AgentId
    ) -> Any | type[DropMessage]:
        if isinstance(message, FunctionCall):
            # 请求用户确认工具执行权限。
            user_input = input(
                f"Function call: {message.name}\nArguments: {message.arguments}\nDo you want to execute the tool? (y/n): "
            )
            if user_input.strip().lower() != "y":
                raise ToolException(content="User denied tool execution.", call_id=message.id, name=message.name)
        return message

现在，我们可以创建一个注册了干预处理器的运行时环境。

# 创建带有干预处理器的运行时环境。
runtime = SingleThreadedAgentRuntime(intervention_handlers=[ToolInterventionHandler()])

在本示例中，我们将使用一个Python代码执行工具。 首先，我们使用DockerCommandLineCodeExecutor 创建一个基于Docker的命令行代码执行器， 然后使用它实例化内置的Python代码执行工具 PythonCodeExecutionTool 该工具在Docker容器中运行代码。

# 创建用于Python代码执行工具的docker执行器。
docker_executor = DockerCommandLineCodeExecutor()

# 创建Python代码执行工具。
python_tool = PythonCodeExecutionTool(executor=docker_executor)

使用工具和工具模式注册代理。

# 注册代理。
tool_agent_type = await ToolAgent.register(
    runtime,
    "tool_executor_agent",
    lambda: ToolAgent(
        description="Tool Executor Agent",
        tools=[python_tool],
    ),
)
model_client = OpenAIChatCompletionClient(model="gpt-4o-mini")
await ToolUseAgent.register(
    runtime,
    "tool_enabled_agent",
    lambda: ToolUseAgent(
        description="Tool Use Agent",
        system_messages=[SystemMessage(content="You are a helpful AI Assistant. Use your tools to solve problems.")],
        model_client=model_client,
        tool_schema=[python_tool.schema],
        tool_agent_type=tool_agent_type,
    ),
)

AgentType(type='tool_enabled_agent')

通过启动运行时并向工具使用代理发送消息来运行代理。 干预处理程序将提示您是否允许执行该工具。

# 启动运行时和docker执行器
await docker_executor.start()
runtime.start()

# 向工具用户发送任务
response = await runtime.send_message(
    Message("Run the following Python code: print('Hello, World!')"), AgentId("tool_enabled_agent", "default")
)
print(response.content)

# 停止运行时和docker执行器
await runtime.stop()
await docker_executor.stop()

# 关闭与模型客户端的连接
await model_client.close()

The output of the code is: **Hello, World!**