A ToolExecutor that delegates tool calls to either a local executor or a Docker environment based on the tool's name.
| 7 | |
| 8 | |
| 9 | class DockerToolExecutor: |
| 10 | """ |
| 11 | A ToolExecutor that delegates tool calls to either a local executor |
| 12 | or a Docker environment based on the tool's name. |
| 13 | """ |
| 14 | |
| 15 | def __init__( |
| 16 | self, |
| 17 | original_executor: ToolExecutor, |
| 18 | docker_manager: DockerManager, |
| 19 | docker_tools: list[str], |
| 20 | host_workspace_dir: str | None, |
| 21 | container_workspace_dir: str, |
| 22 | ): |
| 23 | """ |
| 24 | Initializes the DockerToolExecutor. |
| 25 | """ |
| 26 | self._original_executor = original_executor |
| 27 | self._docker_manager = docker_manager |
| 28 | self._docker_tools_set = set(docker_tools) |
| 29 | # Get path from __init__ --- |
| 30 | self._host_workspace_dir = ( |
| 31 | os.path.abspath(host_workspace_dir) if host_workspace_dir else None |
| 32 | ) |
| 33 | self._container_workspace_dir = container_workspace_dir |
| 34 | |
| 35 | def _translate_path(self, host_path: str) -> str: |
| 36 | """Robust path translation function: Translate the host path into the corresponding path within the container.""" |
| 37 | if not self._host_workspace_dir: |
| 38 | return host_path # 如果没有配置主机工作区，则不翻译 |
| 39 | abs_host_path = os.path.abspath(host_path) |
| 40 | if ( |
| 41 | os.path.commonpath([abs_host_path, self._host_workspace_dir]) |
| 42 | == self._host_workspace_dir |
| 43 | ): |
| 44 | relative_path = os.path.relpath(abs_host_path, self._host_workspace_dir) |
| 45 | container_path = os.path.join(self._container_workspace_dir, relative_path) |
| 46 | return os.path.normpath(container_path) |
| 47 | return host_path |
| 48 | |
| 49 | async def close_tools(self): |
| 50 | """ |
| 51 | Closes any resources held by the underlying original executor. |
| 52 | This method fulfills the contract expected by BaseAgent. |
| 53 | """ |
| 54 | if self._original_executor: |
| 55 | return await self._original_executor.close_tools() |
| 56 | |
| 57 | async def sequential_tool_call(self, tool_calls: list[ToolCall]) -> list[ToolResult]: |
| 58 | """Executes tool calls sequentially, routing to Docker if necessary.""" |
| 59 | results = [] |
| 60 | for tool_call in tool_calls: |
| 61 | if tool_call.name in self._docker_tools_set: |
| 62 | result = self._execute_in_docker(tool_call) |
| 63 | else: |
| 64 | # Execute locally |
| 65 | result_list = await self._original_executor.sequential_tool_call([tool_call]) |
| 66 | result = result_list[0] |