使用干预处理程序进行工具执行的用户批准

此食谱演示了如何使用干预处理程序拦截工具执行，并提示用户批准执行该工具。

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

让我们创建一个简单的工具使用 Agent，该 Agent 能够通过 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")]
        # Use the tool agent to execute the tools, and get the output messages.
        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,
        )
        # Extract the final response from the output messages.
        final_response = output_messages[-1].content
        assert isinstance(final_response, str)
        return Message(content=final_response)

工具使用 Agent 向工具 Agent 发送工具调用请求以执行工具，因此我们可以拦截工具使用 Agent 发送给工具 Agent 的消息，以提示用户批准执行该工具。

让我们创建一个干预处理程序，该处理程序拦截消息并在允许工具执行之前提示用户。

class ToolInterventionHandler(DefaultInterventionHandler):
    async def on_send(
        self, message: Any, *, message_context: MessageContext, recipient: AgentId
    ) -> Any | type[DropMessage]:
        if isinstance(message, FunctionCall):
            # Request user prompt for tool execution.
            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

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

# Create the runtime with the intervention handler.
runtime = SingleThreadedAgentRuntime(intervention_handlers=[ToolInterventionHandler()])

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

# Create the docker executor for the Python code execution tool.
docker_executor = DockerCommandLineCodeExecutor()

# Create the Python code execution tool.
python_tool = PythonCodeExecutionTool(executor=docker_executor)

使用工具和工具模式注册 Agent。

# Register agents.
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')

通过启动运行时并将消息发送到工具使用 Agent 来运行 Agent。 干预处理程序将提示您批准执行该工具。

# Start the runtime and the docker executor.
await docker_executor.start()
runtime.start()

# Send a task to the tool user.
response = await runtime.send_message(
    Message("Run the following Python code: print('Hello, World!')"), AgentId("tool_enabled_agent", "default")
)
print(response.content)

# Stop the runtime and the docker executor.
await runtime.stop()
await docker_executor.stop()

# Close the connection to the model client.
await model_client.close()

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