潜江市网站建设_网站建设公司_模板建站_seo优化

10.3.4 Git管理模块

文件agent_mcp/utils/worktree_utils.py是本项目中用于管理Git工作树（worktree）的工具模块，核心功能是为多代理并行开发提供隔离的代码环境。此文件通过一系列函数实现了Git仓库的验证、分支检查、工作树的创建与清理、未提交更改检测等基础操作，还能自动识别项目类型并生成对应的环境配置命令（如Node.js 项目的 npm install、Python项目的 pip install 等），同时支持标准化的工作树路径和分支名称生成，确保每个代理拥有独立的开发空间，避免代码冲突，提升多代理协作时的版本控制效率。

（1）下面代码的功能是创建新的Git工作树，为代理提供独立的代码开发环境。原理是先检查目标路径是否存在，确保父目录存在，再根据分支是否已存在，决定是基于已有分支创建工作树，还是从基准分支新建分支并创建工作树，最后通过subprocess执行Git命令完成操作，并返回包含创建结果的字典。

def create_git_worktree( path: str, branch: str, base_branch: str = "main", repo_path: str = "." ) -> Dict[str, Any]: """ 创建新的 Git 工作树。 参数: path: 工作树的创建路径 branch: 工作树使用的分支名称 base_branch: 用于创建新分支的基准分支 repo_path: 主仓库的路径 返回: 包含创建状态和详细信息的字典 """ try: # 确保路径为绝对路径且不存在 abs_path = os.path.abspath(path) if os.path.exists(abs_path): return { "success": False, "error": f"路径已存在: {abs_path}", "path": abs_path } # 创建父目录（如果不存在） parent_dir = os.path.dirname(abs_path) os.makedirs(parent_dir, exist_ok=True) # 准备 Git 命令 if branch_exists(branch, repo_path): # 使用已存在的分支 cmd = ["git", "worktree", "add", abs_path, branch] action = f"检出已存在的分支 '{branch}'" else: # 从基准分支创建新分支 cmd = ["git", "worktree", "add", abs_path, "-b", branch, base_branch] action = f"从 '{base_branch}' 创建新分支 '{branch}'" # 执行 Git 工作树命令 result = subprocess.run( cmd, cwd=repo_path, capture_output=True, text=True, timeout=60 # 大型仓库的工作树创建可能较慢 ) if result.returncode == 0: logger.info(f"在 {abs_path} 创建了工作树（{action}）") return { "success": True, "path": abs_path, "branch": branch, "base_branch": base_branch, "action": action, "message": f"工作树已创建于 {abs_path}" } else: logger.error(f"创建工作树失败: {result.stderr}") return { "success": False, "error": result.stderr.strip(), "command": " ".join(cmd), "path": abs_path } except subprocess.TimeoutExpired: return { "success": False, "error": "创建工作树超时（仓库可能过大）", "path": path } except Exception as e: logger.error(f"创建工作树时发生异常: {e}") return { "success": False, "error": str(e), "path": path }

（2）下面代码的功能是清理（删除）指定的Git工作树，释放资源并避免冲突。原理是先检查工作树是否存在，若存在则检测是否有未提交的更改（强制删除时跳过此检查），然后通过subprocess执行Git命令删除工作树，并返回包含删除结果的字典。

def cleanup_git_worktree( path: str, force: bool = False, repo_path: str = "." ) -> Dict[str, Any]: """ 删除 Git 工作树。 参数: path: 要删除的工作树路径 force: 即使有未提交的更改也强制删除 repo_path: 主仓库的路径 返回: 包含删除状态和详细信息的字典 """ try: abs_path = os.path.abspath(path) # 检查工作树是否存在 if not os.path.exists(abs_path): return { "success": True, # 已不存在视为成功 "message": f"{abs_path} 处的工作树不存在", "path": abs_path } # 除非强制删除，否则检查未提交的更改 if not force and has_uncommitted_changes(abs_path): return { "success": False, "error": "工作树存在未提交的更改。使用 force=True 可强制覆盖。", "uncommitted_changes": True, "path": abs_path } # 准备删除命令 cmd = ["git", "worktree", "remove", abs_path] if force: cmd.append("--force") # 执行删除操作 result = subprocess.run( cmd, cwd=repo_path, capture_output=True, text=True, timeout=30 ) if result.returncode == 0: logger.info(f"已删除 {abs_path} 处的工作树") return { "success": True, "message": f"{abs_path} 处的工作树已成功删除", "path": abs_path } else: logger.error(f"删除工作树失败: {result.stderr}") return { "success": False, "error": result.stderr.strip(), "command": " ".join(cmd), "path": abs_path } except subprocess.TimeoutExpired: return { "success": False, "error": "删除工作树超时", "path": path } except Exception as e: logger.error(f"删除工作树时发生异常: {e}") return { "success": False, "error": str(e), "path": path }

（3）下面代码的功能是在工作树中运行项目环境配置命令，自动初始化开发环境。原理是先检查工作树路径是否存在，切换到该路径后，依次执行传入的配置命令（如依赖安装、编译等），记录每个命令的执行结果（成功/失败、输出信息等），最终返回包含整体执行状态的字典。

def run_setup_commands( worktree_path: str, commands: List[str], timeout: int = 300 ) -> Dict[str, Any]: """ 在工作树目录中运行环境配置命令。 参数: worktree_path: 工作树路径 commands: 要运行的配置命令列表 timeout: 每个命令的超时时间（秒） 返回: 包含配置命令执行结果的字典 """ results = [] original_cwd = os.getcwd() # 保存原始工作目录 try: if not os.path.exists(worktree_path): return { "success": False, "error": f"工作树路径不存在: {worktree_path}", "results": [] } os.chdir(worktree_path) # 切换到工作树目录 logger.info(f"在 {worktree_path} 中运行 {len(commands)} 个配置命令") for cmd in commands: logger.debug(f"运行命令: {cmd}") try: result = subprocess.run( cmd.split(), cwd=worktree_path, capture_output=True, text=True, timeout=timeout ) cmd_result = { "command": cmd, "success": result.returncode == 0, "returncode": result.returncode, "stdout": result.stdout, "stderr": result.stderr } if result.returncode == 0: logger.debug(f"✅ {cmd} 执行成功") else: logger.warning(f"❌ {cmd} 执行失败，返回码 {result.returncode}") results.append(cmd_result) except subprocess.TimeoutExpired: cmd_result = { "command": cmd, "success": False, "error": f"命令在 {timeout} 秒后超时" } results.append(cmd_result) logger.error(f"⏰ {cmd} 超时") except Exception as e: cmd_result = { "command": cmd, "success": False, "error": str(e) } results.append(cmd_result) logger.error(f"💥 {cmd} 执行异常: {e}") # 统计成功的命令数量 success_count = sum(1 for r in results if r["success"]) overall_success = success_count == len(commands) logger.info(f"配置完成: {success_count}/{len(commands)} 个命令成功执行") return { "success": overall_success, "results": results, "success_count": success_count, "total_commands": len(commands), "worktree_path": worktree_path } except Exception as e: logger.error(f"运行配置命令时出错: {e}") return { "success": False, "error": str(e), "results": results } finally: os.chdir(original_cwd) # 恢复原始工作目录