From 026588bba3a43f3b2ef7f404133e3eaf05c6e88e Mon Sep 17 00:00:00 2001 From: JOJO <1498581755@qq.com> Date: Mon, 19 Jan 2026 20:58:38 +0800 Subject: [PATCH] fix: align toolbox python env for run_command --- modules/terminal_ops.py | 102 +++++++++++++++++++++++------- sub_agent/modules/terminal_ops.py | 90 ++++++++++++++++++-------- 2 files changed, 140 insertions(+), 52 deletions(-) diff --git a/modules/terminal_ops.py b/modules/terminal_ops.py index f779079..303ff8b 100644 --- a/modules/terminal_ops.py +++ b/modules/terminal_ops.py @@ -7,6 +7,7 @@ import subprocess import shutil import time import signal +import re from pathlib import Path from typing import Dict, Optional, Tuple, TYPE_CHECKING try: @@ -39,8 +40,9 @@ class TerminalOperator: def __init__(self, project_path: str, container_session: Optional["ContainerHandle"] = None): self.project_path = Path(project_path).resolve() self.process = None - # 自动检测Python命令 - self.python_cmd = self._detect_python_command() + # 自动检测Python命令，并记录虚拟环境变量（若存在） + self._python_env: Dict[str, str] = {} + self.python_cmd = self._detect_python_runtime() print(f"{OUTPUT_FORMATS['info']} 检测到Python命令: {self.python_cmd}") self._toolbox: Optional[ToolboxContainer] = None self.container_session: Optional["ContainerHandle"] = container_session @@ -54,28 +56,64 @@ class TerminalOperator: pass self._toolbox = None - def _detect_python_command(self) -> str: + def _detect_python_runtime(self) -> str: """ - 自动检测可用的Python命令 - - Returns: - 可用的Python命令（python、python3、py） + 自动检测可用的 Python 命令，优先使用预装虚拟环境。 + + 1) 优先查找 AGENT_TOOLBOX_VENV /opt/agent-venv / 当前进程 VIRTUAL_ENV。 + 2) 找不到预装虚拟环境时，回退到系统可执行文件。 """ - # 按优先级尝试不同的Python命令 + preferred = self._detect_preinstalled_python() + if preferred: + # 记录虚拟环境相关环境变量，供宿主机子进程复用 + self._python_env = self._build_python_env(preferred) + return preferred + + return self._detect_system_python() + + def _detect_preinstalled_python(self) -> Optional[str]: + """尝试定位预装虚拟环境的 python 可执行文件。""" + candidates = [] + + # 环境变量优先（Dockerfile 已设置 AGENT_TOOLBOX_VENV=/opt/agent-venv） + env_venv = os.environ.get("AGENT_TOOLBOX_VENV") + if env_venv: + candidates.append(env_venv) + + # 默认预装路径 + candidates.append("/opt/agent-venv") + + # 当前进程若已激活虚拟环境，也纳入候选 + current_venv = os.environ.get("VIRTUAL_ENV") + if current_venv: + candidates.append(current_venv) + + # 去重并依次检查 + seen = set() + for raw in candidates: + if not raw or raw in seen: + continue + seen.add(raw) + root = Path(raw).expanduser() + bin_dir = root / ("Scripts" if sys.platform == "win32" else "bin") + for name in ("python3", "python"): + candidate = bin_dir / name + if candidate.exists() and os.access(candidate, os.X_OK): + return str(candidate.resolve()) + return None + + def _detect_system_python(self) -> str: + """在系统 PATH 中探测可用的 Python3 可执行文件。""" commands_to_try = [] - + if sys.platform == "win32": - # Windows优先顺序 commands_to_try = ["python", "py", "python3"] else: - # Unix-like系统优先顺序 commands_to_try = ["python3", "python"] - - # 检测哪个命令可用 + for cmd in commands_to_try: if shutil.which(cmd): try: - # 验证是否真的可以运行 result = subprocess.run( [cmd, "--version"], capture_output=True, @@ -83,15 +121,32 @@ class TerminalOperator: timeout=5 ) if result.returncode == 0: - # 检查版本是否为Python 3 output = result.stdout + result.stderr if "Python 3" in output or "Python 2" not in output: return cmd - except: + except Exception: continue - - # 如果都没找到，根据平台返回默认值 + return "python" if sys.platform == "win32" else "python3" + + def _build_python_env(self, python_path: str) -> Dict[str, str]: + """构造与预装虚拟环境匹配的环境变量（仅作用于宿主机子进程）。""" + env: Dict[str, str] = {} + try: + py_path = Path(python_path).resolve() + bin_dir = py_path.parent + venv_dir = bin_dir.parent + env["VIRTUAL_ENV"] = str(venv_dir) + current_path = os.environ.get("PATH", "") + # 避免重复添加 + prefix = str(bin_dir) + if current_path.startswith(prefix + os.pathsep) or current_path == prefix: + env["PATH"] = current_path + else: + env["PATH"] = f"{prefix}{os.pathsep}{current_path}" if current_path else prefix + except Exception: + pass + return env def _get_toolbox(self) -> ToolboxContainer: if self._toolbox is None: @@ -165,12 +220,9 @@ class TerminalOperator: } # 每次执行前重置工具容器（保持隔离），但下面改用一次性子进程执行，仍保留重置以兼容后续逻辑 self._reset_toolbox() - # 替换命令中的python3为实际可用的命令 - if "python3" in command and self.python_cmd != "python3": - command = command.replace("python3", self.python_cmd) - elif "python" in command and "python3" not in command and self.python_cmd == "python3": - # 如果命令中有python（但不是python3），而系统使用python3 - command = command.replace("python", self.python_cmd) + # 统一替换 python/python3，为保障虚拟环境命中只替换独立单词 + if re.search(r"\bpython3?\b", command): + command = re.sub(r"\bpython3?\b", self.python_cmd, command) # 验证命令 valid, error = self._validate_command(command) @@ -290,6 +342,8 @@ class TerminalOperator: # 统一环境，确保 Python 输出无缓冲 env = os.environ.copy() env.setdefault("PYTHONUNBUFFERED", "1") + if self._python_env: + env.update(self._python_env) if use_shell: process = await asyncio.create_subprocess_shell( diff --git a/sub_agent/modules/terminal_ops.py b/sub_agent/modules/terminal_ops.py index 346dda4..2f940c7 100644 --- a/sub_agent/modules/terminal_ops.py +++ b/sub_agent/modules/terminal_ops.py @@ -5,6 +5,7 @@ import sys import asyncio import subprocess import shutil +import re from pathlib import Path from typing import Dict, Optional, Tuple try: @@ -29,32 +30,49 @@ class TerminalOperator: def __init__(self, project_path: str): self.project_path = Path(project_path).resolve() self.process = None - # 自动检测Python命令 - self.python_cmd = self._detect_python_command() + # 自动检测Python命令，并尝试复用预装虚拟环境 + self._python_env: Dict[str, str] = {} + self.python_cmd = self._detect_python_runtime() print(f"{OUTPUT_FORMATS['info']} 检测到Python命令: {self.python_cmd}") - def _detect_python_command(self) -> str: - """ - 自动检测可用的Python命令 - - Returns: - 可用的Python命令（python、python3、py） - """ - # 按优先级尝试不同的Python命令 - commands_to_try = [] - + def _detect_python_runtime(self) -> str: + """优先选择预装虚拟环境的 Python，其次回退系统可执行文件。""" + preferred = self._detect_preinstalled_python() + if preferred: + self._python_env = self._build_python_env(preferred) + return preferred + return self._detect_system_python() + + def _detect_preinstalled_python(self) -> Optional[str]: + candidates = [] + env_venv = os.environ.get("AGENT_TOOLBOX_VENV") or os.environ.get("VIRTUAL_ENV") + if env_venv: + candidates.append(env_venv) + + candidates.append("/opt/agent-venv") + + seen = set() + for raw in candidates: + if not raw or raw in seen: + continue + seen.add(raw) + root = Path(raw).expanduser() + bin_dir = root / ("Scripts" if sys.platform == "win32" else "bin") + for name in ("python3", "python"): + cand = bin_dir / name + if cand.exists() and os.access(cand, os.X_OK): + return str(cand.resolve()) + return None + + def _detect_system_python(self) -> str: if sys.platform == "win32": - # Windows优先顺序 commands_to_try = ["python", "py", "python3"] else: - # Unix-like系统优先顺序 commands_to_try = ["python3", "python"] - - # 检测哪个命令可用 + for cmd in commands_to_try: if shutil.which(cmd): try: - # 验证是否真的可以运行 result = subprocess.run( [cmd, "--version"], capture_output=True, @@ -62,15 +80,29 @@ class TerminalOperator: timeout=5 ) if result.returncode == 0: - # 检查版本是否为Python 3 output = result.stdout + result.stderr if "Python 3" in output or "Python 2" not in output: return cmd - except: + except Exception: continue - - # 如果都没找到，根据平台返回默认值 return "python" if sys.platform == "win32" else "python3" + + def _build_python_env(self, python_path: str) -> Dict[str, str]: + env: Dict[str, str] = {} + try: + py_path = Path(python_path).resolve() + bin_dir = py_path.parent + venv_dir = bin_dir.parent + env["VIRTUAL_ENV"] = str(venv_dir) + current_path = os.environ.get("PATH", "") + prefix = str(bin_dir) + if current_path.startswith(prefix + os.pathsep) or current_path == prefix: + env["PATH"] = current_path + else: + env["PATH"] = f"{prefix}{os.pathsep}{current_path}" if current_path else prefix + except Exception: + pass + return env def _validate_command(self, command: str) -> Tuple[bool, str]: """验证命令安全性""" @@ -111,12 +143,9 @@ class TerminalOperator: Returns: 执行结果字典 """ - # 替换命令中的python3为实际可用的命令 - if "python3" in command and self.python_cmd != "python3": - command = command.replace("python3", self.python_cmd) - elif "python" in command and "python3" not in command and self.python_cmd == "python3": - # 如果命令中有python（但不是python3），而系统使用python3 - command = command.replace("python", self.python_cmd) + # 替换 python/python3，确保命中预装虚拟环境 + if re.search(r"\bpython3?\b", command): + command = re.sub(r"\bpython3?\b", self.python_cmd, command) # 验证命令 valid, error = self._validate_command(command) @@ -151,12 +180,17 @@ class TerminalOperator: try: # 创建进程 + env = os.environ.copy() + if self._python_env: + env.update(self._python_env) + process = await asyncio.create_subprocess_shell( command, stdout=asyncio.subprocess.PIPE, stderr=asyncio.subprocess.PIPE, cwd=str(work_path), - shell=True + shell=True, + env=env ) # 等待执行完成