From 09a42d29ea5100517caf3a51c451fb51fc157861 Mon Sep 17 00:00:00 2001 From: ben Date: Sun, 31 Aug 2025 01:42:50 +0000 Subject: [PATCH] =?UTF-8?q?=E5=8F=91=E5=B8=83=E7=B1=BB=E5=9E=8B:=20[featur?= =?UTF-8?q?e/bugfix/hotfix/docs]=20=E5=BD=B1=E5=93=8D=E8=8C=83=E5=9B=B4:?= =?UTF-8?q?=20[core/api/ui/config]=20=E6=B5=8B=E8=AF=95=E7=8A=B6=E6=80=81:?= =?UTF-8?q?=20[passed/failed/pending]=20=E5=9B=9E=E6=BB=9A=E7=AD=96?= =?UTF-8?q?=E7=95=A5:=20[=E5=B7=B2=E5=87=86=E5=A4=87/=E6=97=A0=E9=9C=80?= =?UTF-8?q?=E5=9B=9E=E6=BB=9A]?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit --- agents/agent_identity_manager.py | 227 ++++++++++++++++++++++ agents/cli.py | 0 agents/cli_tool.py | 0 agents/demo_collaboration.py | 270 ++++++++++++++++++++++++++ agents/git_collaboration.py | 314 +++++++++++++++++++++++++++++++ agents/identity_manager.py | 237 +++++++++++++++++++++++ agents/setup_agents.sh | 167 ++++++++++++++++ 7 files changed, 1215 insertions(+) create mode 100644 agents/agent_identity_manager.py create mode 100644 agents/cli.py create mode 100644 agents/cli_tool.py create mode 100644 agents/demo_collaboration.py create mode 100644 agents/git_collaboration.py create mode 100644 agents/identity_manager.py create mode 100644 agents/setup_agents.sh diff --git a/agents/agent_identity_manager.py b/agents/agent_identity_manager.py new file mode 100644 index 0000000..1d58927 --- /dev/null +++ b/agents/agent_identity_manager.py @@ -0,0 +1,227 @@ +#!/usr/bin/env python3 +""" +Agent Identity Manager +为每个AI agent提供独立的git身份和提交能力 + +这个系统让每个agent拥有： +- 独立的SSH key对 +- 独立的GPG签名key +- 独立的git配置（name, email） +- 可追溯的提交历史 + +模拟真实团队协作，而非内部讨论 +""" + +import os +import json +import subprocess +import shutil +from pathlib import Path +from typing import Dict, List, Optional +import logging + +class AgentIdentity: + """单个agent的身份信息""" + def __init__(self, name: str, email: str, role: str): + self.name = name + self.email = email + self.role = role + self.ssh_key_path = None + self.gpg_key_id = None + + def to_dict(self) -> Dict: + return { + "name": self.name, + "email": self.email, + "role": self.role, + "ssh_key_path": str(self.ssh_key_path) if self.ssh_key_path else None, + "gpg_key_id": self.gpg_key_id + } + +class AgentIdentityManager: + """管理所有agent的身份和git操作""" + + def __init__(self, base_dir: str = "/home/ben/github/liurenchaxin"): + self.base_dir = Path(base_dir) + self.agents_dir = self.base_dir / "agents" + self.keys_dir = self.agents_dir / "keys" + self.config_file = self.agents_dir / "identities.json" + + # 确保目录存在 + self.agents_dir.mkdir(exist_ok=True) + self.keys_dir.mkdir(exist_ok=True) + + self.identities: Dict[str, AgentIdentity] = {} + self.load_identities() + + def load_identities(self): + """从配置文件加载agent身份""" + if self.config_file.exists(): + with open(self.config_file, 'r', encoding='utf-8') as f: + data = json.load(f) + for name, identity_data in data.items(): + identity = AgentIdentity( + identity_data["name"], + identity_data["email"], + identity_data["role"] + ) + identity.ssh_key_path = Path(identity_data["ssh_key_path"]) if identity_data["ssh_key_path"] else None + identity.gpg_key_id = identity_data["gpg_key_id"] + self.identities[name] = identity + + def save_identities(self): + """保存agent身份到配置文件""" + data = {name: identity.to_dict() for name, identity in self.identities.items()} + with open(self.config_file, 'w', encoding='utf-8') as f: + json.dump(data, f, indent=2, ensure_ascii=False) + + def create_agent(self, name: str, email: str, role: str) -> AgentIdentity: + """创建新的agent身份""" + if name in self.identities: + raise ValueError(f"Agent {name} 已存在") + + identity = AgentIdentity(name, email, role) + + # 生成SSH key + ssh_key_path = self.keys_dir / f"{name}_rsa" + self._generate_ssh_key(name, email, ssh_key_path) + identity.ssh_key_path = ssh_key_path + + # 生成GPG key + gpg_key_id = self._generate_gpg_key(name, email) + identity.gpg_key_id = gpg_key_id + + self.identities[name] = identity + self.save_identities() + + logging.info(f"创建agent: {name} ({role})") + return identity + + def _generate_ssh_key(self, name: str, email: str, key_path: Path): + """为agent生成SSH key""" + cmd = [ + "ssh-keygen", + "-t", "rsa", + "-b", "4096", + "-C", email, + "-f", str(key_path), + "-N", "" # 空密码 + ] + + try: + subprocess.run(cmd, check=True, capture_output=True) + logging.info(f"SSH key已生成: {key_path}") + except subprocess.CalledProcessError as e: + logging.error(f"生成SSH key失败: {e}") + raise + + def _generate_gpg_key(self, name: str, email: str) -> str: + """为agent生成GPG key""" + # 这里简化处理，实际应该使用python-gnupg库 + # 返回模拟的key ID + return f"{name.upper()}12345678" + + def switch_to_agent(self, agent_name: str): + """切换到指定agent身份""" + if agent_name not in self.identities: + raise ValueError(f"Agent {agent_name} 不存在") + + identity = self.identities[agent_name] + + # 设置git配置 + commands = [ + ["git", "config", "user.name", identity.name], + ["git", "config", "user.email", identity.email], + ["git", "config", "user.signingkey", identity.gpg_key_id], + ["git", "config", "commit.gpgsign", "true"] + ] + + for cmd in commands: + try: + subprocess.run(cmd, check=True, cwd=self.base_dir) + except subprocess.CalledProcessError as e: + logging.error(f"设置git配置失败: {e}") + raise + + # 设置SSH key (通过ssh-agent) + if identity.ssh_key_path and identity.ssh_key_path.exists(): + self._setup_ssh_agent(identity.ssh_key_path) + + logging.info(f"已切换到agent: {agent_name}") + + def _setup_ssh_agent(self, key_path: Path): + """设置SSH agent使用指定key""" + # 这里简化处理，实际应该管理ssh-agent + os.environ["GIT_SSH_COMMAND"] = f"ssh -i {key_path}" + + def commit_as_agent(self, agent_name: str, message: str, files: List[str] = None): + """以指定agent身份提交代码""" + self.switch_to_agent(agent_name) + + # 添加文件 + if files: + subprocess.run(["git", "add"] + files, check=True, cwd=self.base_dir) + else: + subprocess.run(["git", "add", "."], check=True, cwd=self.base_dir) + + # 提交 + subprocess.run(["git", "commit", "-S", "-m", message], check=True, cwd=self.base_dir) + + logging.info(f"Agent {agent_name} 提交: {message}") + + def list_agents(self) -> List[Dict]: + """列出所有agent""" + return [identity.to_dict() for identity in self.identities.values()] + + def get_agent_stats(self, agent_name: str) -> Dict: + """获取agent的git统计信息""" + if agent_name not in self.identities: + raise ValueError(f"Agent {agent_name} 不存在") + + identity = self.identities[agent_name] + + # 获取提交统计 + cmd = [ + "git", "log", "--author", identity.email, + "--pretty=format:%h|%an|%ae|%ad|%s", + "--date=short" + ] + + try: + result = subprocess.run(cmd, capture_output=True, text=True, cwd=self.base_dir) + commits = result.stdout.strip().split('\n') if result.stdout.strip() else [] + + return { + "agent_name": agent_name, + "total_commits": len(commits), + "commits": commits[:10] # 最近10条 + } + except subprocess.CalledProcessError: + return { + "agent_name": agent_name, + "total_commits": 0, + "commits": [] + } + +# 使用示例和初始化 +if __name__ == "__main__": + manager = AgentIdentityManager() + + # 创建示例agents + agents_config = [ + {"name": "claude-ai", "email": "claude@ai-collaboration.local", "role": "架构师"}, + {"name": "gemini-dev", "email": "gemini@ai-collaboration.local", "role": "开发者"}, + {"name": "qwen-ops", "email": "qwen@ai-collaboration.local", "role": "运维"}, + {"name": "llama-research", "email": "llama@ai-collaboration.local", "role": "研究员"} + ] + + for agent in agents_config: + try: + manager.create_agent(agent["name"], agent["email"], agent["role"]) + print(f"✅ 创建agent: {agent['name']}") + except ValueError as e: + print(f"⚠️ {e}") + + print("\n📊 当前agent列表:") + for agent in manager.list_agents(): + print(f" - {agent['name']} ({agent['role']}) - {agent['email']}") \ No newline at end of file diff --git a/agents/cli.py b/agents/cli.py new file mode 100644 index 0000000..e69de29 diff --git a/agents/cli_tool.py b/agents/cli_tool.py new file mode 100644 index 0000000..e69de29 diff --git a/agents/demo_collaboration.py b/agents/demo_collaboration.py new file mode 100644 index 0000000..4191095 --- /dev/null +++ b/agents/demo_collaboration.py @@ -0,0 +1,270 @@ +#!/usr/bin/env python3 +""" +Agent协作演示 +展示如何让不同AI agent以真实身份协作完成任务 + +这个演示模拟以下场景： +1. 架构师agent设计系统架构 +2. 开发者agent实现核心功能 +3. 运维agent配置部署 +4. 研究员agent撰写文档 + +每个步骤都有真实的git提交记录 +""" + +import os +import subprocess +import time +from pathlib import Path +from agent_identity_manager import AgentIdentityManager + +class AgentCollaborationDemo: + def __init__(self): + self.manager = AgentIdentityManager() + self.base_dir = Path("/home/ben/github/liurenchaxin") + + def create_demo_files(self): + """创建演示用的文件""" + demo_dir = self.base_dir / "demo_feature" + demo_dir.mkdir(exist_ok=True) + + # 架构师的设计文档 + architecture_file = demo_dir / "architecture.md" + architecture_content = """# 新功能架构设计 + +## 概述 +设计一个智能监控系统，用于跟踪AI agent的工作状态。 + +## 组件设计 +- 状态收集器：收集各agent的运行状态 +- 分析引擎：分析agent行为模式 +- 告警系统：异常行为实时通知 + +## 技术栈 +- Python 3.9+ +- Redis作为消息队列 +- PostgreSQL存储状态数据 +- FastAPI提供REST接口 +""" + architecture_file.write_text(architecture_content) + + # 开发者的实现代码 + core_file = demo_dir / "monitor.py" + core_content = """#!/usr/bin/env python3 +import asyncio +import json +from datetime import datetime +from typing import Dict, Any + +class AgentMonitor: + def __init__(self): + self.agents_status = {} + + async def collect_status(self, agent_name: str) -> Dict[str, Any]: + return { + "name": agent_name, + "timestamp": datetime.now().isoformat(), + "status": "active", + "tasks_completed": 0 + } + + async def run(self): + while True: + # 模拟状态收集 + await asyncio.sleep(1) + +if __name__ == "__main__": + monitor = AgentMonitor() + asyncio.run(monitor.run()) +""" + core_file.write_text(core_content) + + # 运维的配置文件 + config_file = demo_dir / "deploy.yaml" + config_content = """version: '3.8' +services: + agent-monitor: + build: . + ports: + - "8000:8000" + environment: + - REDIS_URL=redis://redis:6379 + - DB_URL=postgresql://user:pass@postgres:5432/agentdb + depends_on: + - redis + - postgres + + redis: + image: redis:alpine + ports: + - "6379:6379" + + postgres: + image: postgres:13 + environment: + POSTGRES_DB: agentdb + POSTGRES_USER: user + POSTGRES_PASSWORD: pass +""" + config_file.write_text(config_content) + + # 研究员的文档 + docs_file = demo_dir / "usage_guide.md" + docs_content = """# Agent监控系统使用指南 + +## 快速开始 + +### 1. 启动监控服务 +```bash +docker-compose up -d +``` + +### 2. 查看agent状态 +```bash +curl http://localhost:8000/api/agents +``` + +### 3. 配置告警 +编辑 `config/alerts.yaml` 文件设置告警规则。 + +## API文档 + +### GET /api/agents +获取所有agent的当前状态 + +### POST /api/agents/{name}/task +记录agent完成的任务 +""" + docs_file.write_text(docs_content) + + return [architecture_file, core_file, config_file, docs_file] + + def run_collaboration_demo(self): + """运行协作演示""" + print("🎭 开始Agent协作演示") + print("=" * 50) + + # 1. 架构师设计 + print("1️⃣ 架构师agent开始设计...") + files = self.create_demo_files() + self.manager.commit_as_agent( + "claude-ai", + "📐 设计智能监控系统架构 - 添加架构设计文档", + [str(f) for f in files[:1]] + ) + time.sleep(1) + + # 2. 开发者实现 + print("2️⃣ 开发者agent开始编码...") + self.manager.commit_as_agent( + "gemini-dev", + "💻 实现监控系统核心功能 - 添加AgentMonitor类", + [str(files[1])] + ) + time.sleep(1) + + # 3. 运维配置 + print("3️⃣ 运维agent配置部署...") + self.manager.commit_as_agent( + "qwen-ops", + "⚙️ 添加Docker部署配置 - 支持一键启动", + [str(files[2])] + ) + time.sleep(1) + + # 4. 研究员文档 + print("4️⃣ 研究员agent撰写文档...") + self.manager.commit_as_agent( + "llama-research", + "📚 完善使用文档 - 添加API说明和快速指南", + [str(files[3])] + ) + time.sleep(1) + + # 5. 架构师review + print("5️⃣ 架构师review并优化...") + optimize_file = self.base_dir / "demo_feature" / "optimization.md" + optimize_content = """# 架构优化建议 + +基于实现代码的review，提出以下优化： + +## 性能优化 +- 使用asyncio.create_task替换直接调用 +- 添加连接池管理 + +## 监控增强 +- 添加prometheus指标收集 +- 实现健康检查端点 + +## 下一步计划 +1. 实现告警系统 +2. 添加Web界面 +3. 集成日志分析 +""" + optimize_file.write_text(optimize_content) + + self.manager.commit_as_agent( + "claude-ai", + "🔍 架构review - 提出性能优化和监控增强建议", + [str(optimize_file)] + ) + + print("\n✅ 协作演示完成！") + + def show_git_history(self): + """显示git提交历史""" + print("\n📊 Git提交历史（按agent分组）:") + print("=" * 50) + + for agent_name in ["claude-ai", "gemini-dev", "qwen-ops", "llama-research"]: + stats = self.manager.get_agent_stats(agent_name) + if stats["commits"]: + print(f"\n👤 {agent_name}:") + for commit in stats["commits"]: + parts = commit.split("|", 4) + if len(parts) >= 5: + hash_id, name, email, date, message = parts + print(f" {hash_id[:8]} {date} {message}") + + def cleanup_demo(self): + """清理演示文件""" + demo_dir = self.base_dir / "demo_feature" + if demo_dir.exists(): + # 保留git历史，只移除工作区文件 + subprocess.run(["git", "rm", "-rf", "demo_feature"], + cwd=self.base_dir, capture_output=True) + subprocess.run(["git", "commit", "-m", "🧹 清理演示文件 - 保留协作历史"], + cwd=self.base_dir, capture_output=True) + print("🧹 演示文件已清理，git历史保留") + +def main(): + """主函数""" + demo = AgentCollaborationDemo() + + print("🎭 AI Agent协作演示") + print("=" * 50) + print("这个演示将展示如何让不同agent以真实身份协作") + print("每个agent都有独立的git身份和提交记录") + print("") + + # 检查agent是否已创建 + if not demo.manager.list_agents(): + print("❌ 请先运行 ./agents/setup_agents.sh 创建agent") + return + + # 运行演示 + demo.run_collaboration_demo() + demo.show_git_history() + + print("\n💡 下一步:") + print("1. 查看git log --oneline --graph 查看提交历史") + print("2. 使用 ./agents/stats.sh 查看agent统计") + print("3. 开始你自己的多agent协作项目！") + + # 询问是否清理 + response = input("\n是否清理演示文件？(y/N): ") + if response.lower() == 'y': + demo.cleanup_demo() + +if __name__ == "__main__": + main() \ No newline at end of file diff --git a/agents/git_collaboration.py b/agents/git_collaboration.py new file mode 100644 index 0000000..b612ca2 --- /dev/null +++ b/agents/git_collaboration.py @@ -0,0 +1,314 @@ +""" +Git 协作管理系统 +管理 Agent 之间基于 Git 的真实协作 +""" + +import os +import subprocess +import json +from pathlib import Path +from typing import Dict, List, Optional, Tuple, Any +from dataclasses import dataclass +import logging +from .identity_manager import AgentIdentityManager + +@dataclass +class Repository: + """仓库信息""" + name: str + local_path: str + remotes: Dict[str, str] # remote_name -> url + current_agent: Optional[str] = None + +class GitCollaborationManager: + """Git 协作管理器""" + + def __init__(self, identity_manager: AgentIdentityManager): + self.identity_manager = identity_manager + self.logger = logging.getLogger(__name__) + self.repositories = {} + self._load_repositories() + + def _load_repositories(self): + """加载仓库配置""" + config_file = Path("config/repositories.json") + if config_file.exists(): + with open(config_file, 'r', encoding='utf-8') as f: + data = json.load(f) + self.repositories = { + name: Repository(**repo_data) + for name, repo_data in data.items() + } + + def _save_repositories(self): + """保存仓库配置""" + config_file = Path("config/repositories.json") + config_file.parent.mkdir(exist_ok=True) + + data = { + name: { + 'name': repo.name, + 'local_path': repo.local_path, + 'remotes': repo.remotes, + 'current_agent': repo.current_agent + } + for name, repo in self.repositories.items() + } + + with open(config_file, 'w', encoding='utf-8') as f: + json.dump(data, f, indent=2, ensure_ascii=False) + + def setup_progressive_deployment(self, + repo_name: str, + gitea_url: str, + bitbucket_url: str, + github_url: str, + local_path: Optional[str] = None): + """设置渐进发布的三个远程仓库""" + + if not local_path: + local_path_str = f"./repos/{repo_name}" + else: + local_path_str = local_path + + local_path_obj = Path(local_path_str) + local_path_obj.mkdir(parents=True, exist_ok=True) + + # 初始化本地仓库（如果不存在） + if not (local_path_obj / ".git").exists(): + subprocess.run(["git", "init"], cwd=local_path) + + # 设置远程仓库 + remotes = { + "gitea": gitea_url, + "bitbucket": bitbucket_url, + "github": github_url + } + + for remote_name, remote_url in remotes.items(): + # 检查远程是否已存在 + result = subprocess.run([ + "git", "remote", "get-url", remote_name + ], cwd=local_path, capture_output=True, text=True) + + if result.returncode != 0: + # 添加新的远程 + subprocess.run([ + "git", "remote", "add", remote_name, remote_url + ], cwd=local_path) + else: + # 更新现有远程 + subprocess.run([ + "git", "remote", "set-url", remote_name, remote_url + ], cwd=local_path) + + # 创建仓库记录 + repository = Repository( + name=repo_name, + local_path=str(local_path), + remotes=remotes + ) + + self.repositories[repo_name] = repository + self._save_repositories() + + self.logger.info(f"设置渐进发布仓库: {repo_name}") + return repository + + def switch_agent_context(self, repo_name: str, agent_name: str): + """切换仓库的 Agent 上下文""" + if repo_name not in self.repositories: + raise ValueError(f"仓库 {repo_name} 不存在") + + repository = self.repositories[repo_name] + + # 设置 Git 配置 + self.identity_manager.setup_git_config(agent_name, repository.local_path) + + # 设置 SSH 密钥 + identity = self.identity_manager.get_agent_identity(agent_name) + if identity: + self._setup_ssh_agent(identity.ssh_key_path) + + repository.current_agent = agent_name + self._save_repositories() + + self.logger.info(f"切换仓库 {repo_name} 到 Agent: {agent_name}") + + def _setup_ssh_agent(self, ssh_key_path: str): + """设置 SSH Agent""" + try: + # 启动 ssh-agent（如果未运行） + result = subprocess.run([ + "ssh-add", "-l" + ], capture_output=True, text=True) + + if result.returncode != 0: + # 启动 ssh-agent + result = subprocess.run([ + "ssh-agent", "-s" + ], capture_output=True, text=True) + + if result.returncode == 0: + # 解析环境变量 + for line in result.stdout.split('\n'): + if 'SSH_AUTH_SOCK' in line: + sock = line.split('=')[1].split(';')[0] + os.environ['SSH_AUTH_SOCK'] = sock + elif 'SSH_AGENT_PID' in line: + pid = line.split('=')[1].split(';')[0] + os.environ['SSH_AGENT_PID'] = pid + + # 添加 SSH 密钥 + subprocess.run(["ssh-add", ssh_key_path]) + + except Exception as e: + self.logger.warning(f"SSH Agent 设置失败: {e}") + + def commit_as_agent(self, + repo_name: str, + message: str, + files: Optional[List[str]] = None, + sign: bool = True) -> bool: + """以当前 Agent 身份提交代码""" + + if repo_name not in self.repositories: + raise ValueError(f"仓库 {repo_name} 不存在") + + repository = self.repositories[repo_name] + repo_path = Path(repository.local_path) + + try: + # 添加文件 + if files: + for file in files: + subprocess.run(["git", "add", file], cwd=repo_path) + else: + subprocess.run(["git", "add", "."], cwd=repo_path) + + # 提交 + commit_cmd = ["git", "commit", "-m", message] + if sign: + commit_cmd.append("-S") + + result = subprocess.run(commit_cmd, cwd=repo_path, capture_output=True, text=True) + + if result.returncode == 0: + self.logger.info(f"Agent {repository.current_agent} 提交成功: {message}") + return True + else: + self.logger.error(f"提交失败: {result.stderr}") + return False + + except Exception as e: + self.logger.error(f"提交过程出错: {e}") + return False + + def progressive_push(self, repo_name: str, branch: str = "main") -> Dict[str, bool]: + """渐进式推送到三个平台""" + + if repo_name not in self.repositories: + raise ValueError(f"仓库 {repo_name} 不存在") + + repository = self.repositories[repo_name] + repo_path = Path(repository.local_path) + + results = {} + + # 按顺序推送：Gitea -> Bitbucket -> GitHub + push_order = ["gitea", "bitbucket", "github"] + + for remote in push_order: + if remote in repository.remotes: + try: + result = subprocess.run([ + "git", "push", remote, branch + ], cwd=repo_path, capture_output=True, text=True) + + results[remote] = result.returncode == 0 + + if result.returncode == 0: + self.logger.info(f"推送到 {remote} 成功") + else: + self.logger.error(f"推送到 {remote} 失败: {result.stderr}") + # 如果某个平台失败，停止后续推送 + break + + except Exception as e: + self.logger.error(f"推送到 {remote} 出错: {e}") + results[remote] = False + break + + return results + + def create_pull_request_workflow(self, + repo_name: str, + source_agent: str, + target_agent: str, + feature_branch: str, + title: str, + description: str = "") -> bool: + """创建 Agent 间的 Pull Request 工作流""" + + repository = self.repositories[repo_name] + repo_path = Path(repository.local_path) + + try: + # 1. 切换到源 Agent + self.switch_agent_context(repo_name, source_agent) + + # 2. 创建功能分支 + subprocess.run([ + "git", "checkout", "-b", feature_branch + ], cwd=repo_path) + + # 3. 推送功能分支 + subprocess.run([ + "git", "push", "-u", "gitea", feature_branch + ], cwd=repo_path) + + # 4. 这里可以集成 API 调用来创建实际的 PR + # 具体实现取决于使用的 Git 平台 + + self.logger.info(f"创建 PR 工作流: {source_agent} -> {target_agent}") + return True + + except Exception as e: + self.logger.error(f"创建 PR 工作流失败: {e}") + return False + + def get_repository_status(self, repo_name: str) -> Dict[str, Any]: + """获取仓库状态""" + if repo_name not in self.repositories: + raise ValueError(f"仓库 {repo_name} 不存在") + + repository = self.repositories[repo_name] + repo_path = Path(repository.local_path) + + status = { + "current_agent": repository.current_agent, + "branch": None, + "uncommitted_changes": False, + "remotes": repository.remotes + } + + try: + # 获取当前分支 + result = subprocess.run([ + "git", "branch", "--show-current" + ], cwd=repo_path, capture_output=True, text=True) + + if result.returncode == 0: + status["branch"] = result.stdout.strip() + + # 检查未提交的更改 + result = subprocess.run([ + "git", "status", "--porcelain" + ], cwd=repo_path, capture_output=True, text=True) + + status["uncommitted_changes"] = bool(result.stdout.strip()) + + except Exception as e: + self.logger.error(f"获取仓库状态失败: {e}") + + return status \ No newline at end of file diff --git a/agents/identity_manager.py b/agents/identity_manager.py new file mode 100644 index 0000000..115159d --- /dev/null +++ b/agents/identity_manager.py @@ -0,0 +1,237 @@ +""" +Agent Identity Management System +管理多个 AI Agent 的身份信息，包括 SSH/GPG 密钥、Git 配置等 +""" + +import os +import json +import subprocess +from pathlib import Path +from typing import Dict, List, Optional +from dataclasses import dataclass, asdict +import logging + +@dataclass +class AgentIdentity: + """Agent 身份信息""" + name: str + email: str + ssh_key_path: str + gpg_key_id: Optional[str] = None + git_username: str = "" + description: str = "" + repositories: List[str] = None + + def __post_init__(self): + if self.repositories is None: + self.repositories = [] + if not self.git_username: + self.git_username = self.name.lower().replace(" ", "_") + +class AgentIdentityManager: + """Agent 身份管理器""" + + def __init__(self, config_dir: str = "config/agents"): + self.config_dir = Path(config_dir) + self.config_dir.mkdir(parents=True, exist_ok=True) + self.identities_file = self.config_dir / "identities.json" + self.ssh_keys_dir = self.config_dir / "ssh_keys" + self.gpg_keys_dir = self.config_dir / "gpg_keys" + + # 创建必要的目录 + self.ssh_keys_dir.mkdir(exist_ok=True) + self.gpg_keys_dir.mkdir(exist_ok=True) + + self.logger = logging.getLogger(__name__) + self._load_identities() + + def _load_identities(self): + """加载已有的身份信息""" + if self.identities_file.exists(): + with open(self.identities_file, 'r', encoding='utf-8') as f: + data = json.load(f) + self.identities = { + name: AgentIdentity(**identity_data) + for name, identity_data in data.items() + } + else: + self.identities = {} + + def _save_identities(self): + """保存身份信息到文件""" + data = { + name: asdict(identity) + for name, identity in self.identities.items() + } + with open(self.identities_file, 'w', encoding='utf-8') as f: + json.dump(data, f, indent=2, ensure_ascii=False) + + def create_agent_identity(self, + name: str, + email: str, + description: str = "", + generate_keys: bool = True) -> AgentIdentity: + """创建新的 Agent 身份""" + + if name in self.identities: + raise ValueError(f"Agent {name} 已存在") + + # 生成 SSH 密钥路径 + ssh_key_path = str(self.ssh_keys_dir / f"{name.lower().replace(' ', '_')}_rsa") + + identity = AgentIdentity( + name=name, + email=email, + ssh_key_path=ssh_key_path, + description=description + ) + + if generate_keys: + self._generate_ssh_key(identity) + self._generate_gpg_key(identity) + + self.identities[name] = identity + self._save_identities() + + self.logger.info(f"创建 Agent 身份: {name}") + return identity + + def _generate_ssh_key(self, identity: AgentIdentity): + """为 Agent 生成 SSH 密钥对""" + try: + cmd = [ + "ssh-keygen", + "-t", "rsa", + "-b", "4096", + "-C", identity.email, + "-f", identity.ssh_key_path, + "-N", "" # 无密码 + ] + + result = subprocess.run(cmd, capture_output=True, text=True) + if result.returncode != 0: + raise Exception(f"SSH 密钥生成失败: {result.stderr}") + + # 设置正确的权限 + os.chmod(identity.ssh_key_path, 0o600) + os.chmod(f"{identity.ssh_key_path}.pub", 0o644) + + self.logger.info(f"为 {identity.name} 生成 SSH 密钥: {identity.ssh_key_path}") + + except Exception as e: + self.logger.error(f"SSH 密钥生成失败: {e}") + raise + + def _generate_gpg_key(self, identity: AgentIdentity): + """为 Agent 生成 GPG 密钥""" + try: + # GPG 密钥生成配置 + gpg_config = f""" +Key-Type: RSA +Key-Length: 4096 +Subkey-Type: RSA +Subkey-Length: 4096 +Name-Real: {identity.name} +Name-Email: {identity.email} +Expire-Date: 0 +%no-protection +%commit +""" + + # 写入临时配置文件 + config_file = self.gpg_keys_dir / f"{identity.git_username}_gpg_config" + with open(config_file, 'w') as f: + f.write(gpg_config) + + # 生成 GPG 密钥 + cmd = ["gpg", "--batch", "--generate-key", str(config_file)] + result = subprocess.run(cmd, capture_output=True, text=True) + + if result.returncode != 0: + self.logger.warning(f"GPG 密钥生成失败: {result.stderr}") + return + + # 获取生成的密钥 ID + cmd = ["gpg", "--list-secret-keys", "--keyid-format", "LONG", identity.email] + result = subprocess.run(cmd, capture_output=True, text=True) + + if result.returncode == 0: + # 解析密钥 ID + lines = result.stdout.split('\n') + for line in lines: + if 'sec' in line and 'rsa4096/' in line: + key_id = line.split('rsa4096/')[1].split(' ')[0] + identity.gpg_key_id = key_id + break + + # 清理临时文件 + config_file.unlink() + + self.logger.info(f"为 {identity.name} 生成 GPG 密钥: {identity.gpg_key_id}") + + except Exception as e: + self.logger.warning(f"GPG 密钥生成失败: {e}") + + def get_agent_identity(self, name: str) -> Optional[AgentIdentity]: + """获取 Agent 身份信息""" + return self.identities.get(name) + + def list_agents(self) -> List[str]: + """列出所有 Agent""" + return list(self.identities.keys()) + + def setup_git_config(self, agent_name: str, repo_path: str = "."): + """为指定仓库设置 Agent 的 Git 配置""" + identity = self.get_agent_identity(agent_name) + if not identity: + raise ValueError(f"Agent {agent_name} 不存在") + + repo_path = Path(repo_path) + + # 设置 Git 用户信息 + subprocess.run([ + "git", "config", "--local", "user.name", identity.name + ], cwd=repo_path) + + subprocess.run([ + "git", "config", "--local", "user.email", identity.email + ], cwd=repo_path) + + # 设置 GPG 签名 + if identity.gpg_key_id: + subprocess.run([ + "git", "config", "--local", "user.signingkey", identity.gpg_key_id + ], cwd=repo_path) + + subprocess.run([ + "git", "config", "--local", "commit.gpgsign", "true" + ], cwd=repo_path) + + self.logger.info(f"为仓库 {repo_path} 设置 {agent_name} 的 Git 配置") + + def get_ssh_public_key(self, agent_name: str) -> str: + """获取 Agent 的 SSH 公钥""" + identity = self.get_agent_identity(agent_name) + if not identity: + raise ValueError(f"Agent {agent_name} 不存在") + + pub_key_path = f"{identity.ssh_key_path}.pub" + if not os.path.exists(pub_key_path): + raise FileNotFoundError(f"SSH 公钥文件不存在: {pub_key_path}") + + with open(pub_key_path, 'r') as f: + return f.read().strip() + + def export_gpg_public_key(self, agent_name: str) -> str: + """导出 Agent 的 GPG 公钥""" + identity = self.get_agent_identity(agent_name) + if not identity or not identity.gpg_key_id: + raise ValueError(f"Agent {agent_name} 没有 GPG 密钥") + + cmd = ["gpg", "--armor", "--export", identity.gpg_key_id] + result = subprocess.run(cmd, capture_output=True, text=True) + + if result.returncode != 0: + raise Exception(f"GPG 公钥导出失败: {result.stderr}") + + return result.stdout \ No newline at end of file diff --git a/agents/setup_agents.sh b/agents/setup_agents.sh new file mode 100644 index 0000000..f5eb5e7 --- /dev/null +++ b/agents/setup_agents.sh @@ -0,0 +1,167 @@ +#!/bin/bash +# Agent协作系统设置脚本 +# 为一人公司创建多agent git协作环境 + +set -e + +echo "🚀 设置AI Agent协作系统..." + +# 创建必要的目录 +mkdir -p agents/keys +mkdir -p agents/logs + +# 设置权限 +chmod 700 agents/keys + +# 检查依赖 +check_dependency() { + if ! command -v $1 &> /dev/null; then + echo "❌ 需要安装: $1" + exit 1 + fi +} + +check_dependency "git" +check_dependency "ssh-keygen" + +echo "✅ 依赖检查通过" + +# 初始化agent身份管理器 +echo "🤖 初始化agent身份..." +python3 agents/agent_identity_manager.py + +# 创建git hooks模板 +cat > agents/pre-commit-hook << 'EOF' +#!/bin/bash +# Agent提交前的钩子 + +echo "🔍 检查agent身份..." +AGENT_NAME=$(git config user.name) +if [[ -z "$AGENT_NAME" ]]; then + echo "❌ 未设置agent身份，请先使用agent协作系统" + exit 1 +fi + +echo "✅ 当前agent: $AGENT_NAME" +EOF + +chmod +x agents/pre-commit-hook + +# 创建快速切换脚本 +cat > agents/switch_agent.sh << 'EOF' +#!/bin/bash +# 快速切换agent身份 + +if [[ $# -eq 0 ]]; then + echo "用法: ./switch_agent.sh " + echo "可用agents:" + python3 -c " +import sys +sys.path.append('agents') +from agent_identity_manager import AgentIdentityManager +manager = AgentIdentityManager() +for agent in manager.list_agents(): + print(f' - {agent[\"name\"]} ({agent[\"role\"]})') + " + exit 1 +fi + +AGENT_NAME=$1 +echo "🔄 切换到agent: $AGENT_NAME" +python3 -c " +import sys +sys.path.append('agents') +from agent_identity_manager import AgentIdentityManager +manager = AgentIdentityManager() +try: + manager.switch_to_agent('$AGENT_NAME') + print('✅ 切换成功') +except Exception as e: + print(f'❌ 切换失败: {e}') + exit(1) +" +EOF + +chmod +x agents/switch_agent.sh + +# 创建agent提交脚本 +cat > agents/commit_as_agent.sh << 'EOF' +#!/bin/bash +# 以指定agent身份提交 + +if [[ $# -lt 2 ]]; then + echo "用法: ./commit_as_agent.sh \"提交信息\" [文件...]" + exit 1 +fi + +AGENT_NAME=$1 +MESSAGE=$2 +shift 2 +FILES=$@ + +echo "📝 Agent $AGENT_NAME 正在提交..." +python3 -c " +import sys +sys.path.append('agents') +from agent_identity_manager import AgentIdentityManager +manager = AgentIdentityManager() +try: + manager.commit_as_agent('$AGENT_NAME', '$MESSAGE', '$FILES'.split() if '$FILES' else None) + print('✅ 提交成功') +except Exception as e: + print(f'❌ 提交失败: {e}') + exit(1) +" +EOF + +chmod +x agents/commit_as_agent.sh + +# 创建统计脚本 +cat > agents/stats.sh << 'EOF' +#!/bin/bash +# 查看agent统计信息 + +echo "📊 Agent协作统计" +echo "==================" + +python3 -c " +import sys +sys.path.append('agents') +from agent_identity_manager import AgentIdentityManager +manager = AgentIdentityManager() + +for agent in manager.list_agents(): + name = agent['name'] + stats = manager.get_agent_stats(name) + print(f'👤 {name} ({agent["role"]})') + print(f' 📧 {agent["email"]}') + print(f' 📈 提交数: {stats["total_commits"]}') + if stats["commits"]: + print(f' 📝 最近提交: {stats["commits"][0]}') + print() +" +EOF + +chmod +x agents/stats.sh + +echo "🎉 设置完成！" +echo "" +echo "📋 使用说明:" +echo "1. 查看agent列表: ./agents/stats.sh" +echo "2. 切换agent: ./agents/switch_agent.sh " +echo "3. agent提交: ./agents/commit_as_agent.sh \"消息\"" +echo "4. 查看统计: ./agents/stats.sh" +echo "" +echo "🔑 SSH公钥位置:" +for key in agents/keys/*_rsa.pub; do + if [[ -f "$key" ]]; then + agent_name=$(basename "$key" _rsa.pub) + echo " $agent_name: $key" + fi +done + +echo "" +echo "💡 下一步:" +echo "1. 将SSH公钥添加到GitHub/Gitea/Bitbucket" +echo "2. 测试agent切换和提交功能" +echo "3. 开始真正的多agent协作开发！" \ No newline at end of file