{
  "cells": [
    {
      "cell_type": "markdown",
      "metadata": {},
      "source": [
        "# 使用干预处理器实现工具执行的用户审批\n\n本教程展示如何通过干预处理器拦截工具执行，并提示用户授权执行工具。\n"
      ]
    },
    {
      "cell_type": "code",
      "execution_count": 27,
      "metadata": {},
      "outputs": [],
      "source": [
        "from dataclasses import dataclass\n",
        "from typing import Any, List\n",
        "\n",
        "from autogen_core import (\n",
        "    AgentId,\n",
        "    AgentType,\n",
        "    DefaultInterventionHandler,\n",
        "    DropMessage,\n",
        "    FunctionCall,\n",
        "    MessageContext,\n",
        "    RoutedAgent,\n",
        "    SingleThreadedAgentRuntime,\n",
        "    message_handler,\n",
        ")\n",
        "from autogen_core.models import (\n",
        "    ChatCompletionClient,\n",
        "    LLMMessage,\n",
        "    SystemMessage,\n",
        "    UserMessage,\n",
        ")\n",
        "from autogen_core.tool_agent import ToolAgent, ToolException, tool_agent_caller_loop\n",
        "from autogen_core.tools import ToolSchema\n",
        "from autogen_ext.code_executors.docker import DockerCommandLineCodeExecutor\n",
        "from autogen_ext.models.openai import OpenAIChatCompletionClient\n",
        "from autogen_ext.tools.code_execution import PythonCodeExecutionTool"
      ]
    },
    {
      "cell_type": "markdown",
      "metadata": {},
      "source": [
        "让我们定义一个携带字符串内容的简单消息类型。\n"
      ]
    },
    {
      "cell_type": "code",
      "execution_count": 28,
      "metadata": {},
      "outputs": [],
      "source": [
        "@dataclass\n",
        "class Message:\n",
        "    content: str"
      ]
    },
    {
      "cell_type": "markdown",
      "metadata": {},
      "source": [
        "让我们创建一个能通过 {py:class}`~autogen_core.tool_agent.ToolAgent` 使用工具的简单工具代理。\n"
      ]
    },
    {
      "cell_type": "code",
      "execution_count": 29,
      "metadata": {},
      "outputs": [],
      "source": [
        "class ToolUseAgent(RoutedAgent):\n",
        "    \"\"\"An agent that uses tools to perform tasks. It executes the tools\n",
        "    by itself by sending the tool execution task to a ToolAgent.\"\"\"\n",
        "\n",
        "    def __init__(\n",
        "        self,\n",
        "        description: str,\n",
        "        system_messages: List[SystemMessage],\n",
        "        model_client: ChatCompletionClient,\n",
        "        tool_schema: List[ToolSchema],\n",
        "        tool_agent_type: AgentType,\n",
        "    ) -> None:\n",
        "        super().__init__(description)\n",
        "        self._model_client = model_client\n",
        "        self._system_messages = system_messages\n",
        "        self._tool_schema = tool_schema\n",
        "        self._tool_agent_id = AgentId(type=tool_agent_type, key=self.id.key)\n",
        "\n",
        "    @message_handler\n",
        "    async def handle_user_message(self, message: Message, ctx: MessageContext) -> Message:\n",
        "        \"\"\"Handle a user message, execute the model and tools, and returns the response.\"\"\"\n",
        "        session: List[LLMMessage] = [UserMessage(content=message.content, source=\"User\")]\n",
        "        # 使用工具代理执行工具，并获取输出消息。\n",
        "        output_messages = await tool_agent_caller_loop(\n",
        "            self,\n",
        "            tool_agent_id=self._tool_agent_id,\n",
        "            model_client=self._model_client,\n",
        "            input_messages=session,\n",
        "            tool_schema=self._tool_schema,\n",
        "            cancellation_token=ctx.cancellation_token,\n",
        "        )\n",
        "        # 从输出消息中提取最终响应。\n",
        "        final_response = output_messages[-1].content\n",
        "        assert isinstance(final_response, str)\n",
        "        return Message(content=final_response)"
      ]
    },
    {
      "cell_type": "markdown",
      "metadata": {},
      "source": [
        "工具使用代理向工具代理发送工具调用请求以执行工具，\n因此我们可以拦截工具使用代理发送给工具代理的消息，\n在执行工具前提示用户获取执行权限。\n\n让我们创建一个干预处理器来拦截这些消息，\n并在允许工具执行前提示用户确认。\n"
      ]
    },
    {
      "cell_type": "code",
      "execution_count": 30,
      "metadata": {},
      "outputs": [],
      "source": [
        "class ToolInterventionHandler(DefaultInterventionHandler):\n",
        "    async def on_send(\n",
        "        self, message: Any, *, message_context: MessageContext, recipient: AgentId\n",
        "    ) -> Any | type[DropMessage]:\n",
        "        if isinstance(message, FunctionCall):\n",
        "            # 请求用户确认工具执行权限。\n",
        "            user_input = input(\n",
        "                f\"Function call: {message.name}\\nArguments: {message.arguments}\\nDo you want to execute the tool? (y/n): \"\n",
        "            )\n",
        "            if user_input.strip().lower() != \"y\":\n",
        "                raise ToolException(content=\"User denied tool execution.\", call_id=message.id, name=message.name)\n",
        "        return message"
      ]
    },
    {
      "cell_type": "markdown",
      "metadata": {},
      "source": [
        "现在，我们可以创建一个注册了干预处理器的运行时环境。\n"
      ]
    },
    {
      "cell_type": "code",
      "execution_count": 31,
      "metadata": {},
      "outputs": [],
      "source": [
        "# 创建带有干预处理器的运行时环境。\n",
        "runtime = SingleThreadedAgentRuntime(intervention_handlers=[ToolInterventionHandler()])"
      ]
    },
    {
      "cell_type": "markdown",
      "metadata": {},
      "source": [
        "在本示例中，我们将使用一个Python代码执行工具。\n首先，我们使用{py:class}`~autogen_ext.code_executors.docker.DockerCommandLineCodeExecutor`\n创建一个基于Docker的命令行代码执行器，\n然后使用它实例化内置的Python代码执行工具\n{py:class}`~autogen_core.tools.PythonCodeExecutionTool`\n该工具在Docker容器中运行代码。\n"
      ]
    },
    {
      "cell_type": "code",
      "execution_count": 32,
      "metadata": {},
      "outputs": [],
      "source": [
        "# 创建用于Python代码执行工具的docker执行器。\n",
        "docker_executor = DockerCommandLineCodeExecutor()\n",
        "\n",
        "# 创建Python代码执行工具。\n",
        "python_tool = PythonCodeExecutionTool(executor=docker_executor)"
      ]
    },
    {
      "cell_type": "markdown",
      "metadata": {},
      "source": [
        "使用工具和工具模式注册代理。\n"
      ]
    },
    {
      "cell_type": "code",
      "execution_count": 33,
      "metadata": {},
      "outputs": [
        {
          "data": {
            "text/plain": [
              "AgentType(type='tool_enabled_agent')"
            ]
          },
          "execution_count": 33,
          "metadata": {},
          "output_type": "execute_result"
        }
      ],
      "source": [
        "# 注册代理。\n",
        "tool_agent_type = await ToolAgent.register(\n",
        "    runtime,\n",
        "    \"tool_executor_agent\",\n",
        "    lambda: ToolAgent(\n",
        "        description=\"Tool Executor Agent\",\n",
        "        tools=[python_tool],\n",
        "    ),\n",
        ")\n",
        "model_client = OpenAIChatCompletionClient(model=\"gpt-4o-mini\")\n",
        "await ToolUseAgent.register(\n",
        "    runtime,\n",
        "    \"tool_enabled_agent\",\n",
        "    lambda: ToolUseAgent(\n",
        "        description=\"Tool Use Agent\",\n",
        "        system_messages=[SystemMessage(content=\"You are a helpful AI Assistant. Use your tools to solve problems.\")],\n",
        "        model_client=model_client,\n",
        "        tool_schema=[python_tool.schema],\n",
        "        tool_agent_type=tool_agent_type,\n",
        "    ),\n",
        ")"
      ]
    },
    {
      "cell_type": "markdown",
      "metadata": {},
      "source": [
        "通过启动运行时并向工具使用代理发送消息来运行代理。\n干预处理程序将提示您是否允许执行该工具。\n"
      ]
    },
    {
      "cell_type": "code",
      "execution_count": 34,
      "metadata": {},
      "outputs": [
        {
          "name": "stdout",
          "output_type": "stream",
          "text": [
            "The output of the code is: **Hello, World!**\n"
          ]
        }
      ],
      "source": [
        "# 启动运行时和docker执行器\n",
        "await docker_executor.start()\n",
        "runtime.start()\n",
        "\n",
        "# 向工具用户发送任务\n",
        "response = await runtime.send_message(\n",
        "    Message(\"Run the following Python code: print('Hello, World!')\"), AgentId(\"tool_enabled_agent\", \"default\")\n",
        ")\n",
        "print(response.content)\n",
        "\n",
        "# 停止运行时和docker执行器\n",
        "await runtime.stop()\n",
        "await docker_executor.stop()\n",
        "\n",
        "# 关闭与模型客户端的连接\n",
        "await model_client.close()"
      ]
    }
  ],
  "metadata": {
    "kernelspec": {
      "display_name": ".venv",
      "language": "python",
      "name": "python3"
    },
    "language_info": {
      "codemirror_mode": {
        "name": "ipython",
        "version": 3
      },
      "file_extension": ".py",
      "mimetype": "text/x-python",
      "name": "python",
      "nbconvert_exporter": "python",
      "pygments_lexer": "ipython3",
      "version": "3.11.9"
    }
  },
  "nbformat": 4,
  "nbformat_minor": 2
}