diff --git a/SWARM_README.md b/SWARM_README.md new file mode 100644 index 0000000..63c318a --- /dev/null +++ b/SWARM_README.md @@ -0,0 +1,247 @@ +# Swarm Mode - 多Agent协作编码 + +## 概述 + +Swarm Mode是agent-manager的多Agent协作功能，允许Claude Code通过API在K8s上创建多个specialized agents协作完成编码任务。 + +## 架构 + +``` +Claude Code (本地) + ↓ HTTP/MCP +Agent Manager (/api/swarm/*) + ↓ +Swarm Orchestrator (任务分解、分配、聚合) + ↓ +K8s Cluster (多个独立Namespace) + ├─ architect-agent (设计) + ├─ coder-agent-1 (实现) + ├─ coder-agent-2 (实现) + └─ reviewer-agent (审查) + ↓ A2A协议通信 + 结果聚合 → SSE Stream → Claude Code +``` + +## 核心功能 + +### 1. 数据库模型 + +- **Swarm**: 蜂群主表，存储任务描述、编排策略、状态等 +- **SwarmAgent**: 蜂群中的Agent实例，存储角色、配置、K8s资源信息 +- **SwarmMessage**: Agent间通信消息记录 + +### 2. API端点 + +#### POST /api/swarm/create +创建新的蜂群。 + +**请求示例：** +```json +{ + "task_description": "实现用户认证模块", + "project_context": { + "repo_url": "https://github.com/user/project", + "branch": "feature/auth", + "language": "python", + "framework": "fastapi" + }, + "agents": [ + { + "role": "architect", + "template": "a2a_litellm_agent", + "model": "gpt-4", + "capabilities": ["design", "planning"], + "replicas": 1 + }, + { + "role": "coder", + "template": "code_manager_agent", + "model": "gpt-4", + "capabilities": ["coding", "git"], + "replicas": 2 + } + ], + "orchestration": { + "strategy": "sequential", + "max_iterations": 3, + "timeout_minutes": 30 + } +} +``` + +**响应示例：** +```json +{ + "swarm_id": "swm_a1b2c3d4e5f6", + "status": "initializing", + "agents": [ + { + "agent_id": "agi_architect_001", + "role": "architect", + "status": "pending", + "namespace": "swarm-swm-a1b2c3-architect" + } + ], + "created_at": "2026-05-17T10:30:00Z" +} +``` + +#### GET /api/swarm/{swarm_id}/status +查询蜂群状态。 + +**响应示例：** +```json +{ + "swarm_id": "swm_a1b2c3d4e5f6", + "status": "running", + "phase": "coding", + "progress": 65, + "agents": [...], + "metrics": { + "total_messages": 45, + "tokens_used": 125000, + "elapsed_seconds": 180 + } +} +``` + +#### GET /api/swarm/{swarm_id}/results +SSE流式获取实时结果。 + +**事件类型：** +- `phase_change`: 阶段变更 +- `agent_message`: Agent消息 +- `artifact`: 生成的工件（代码、文档等） +- `code_diff`: 代码变更 +- `result`: 最终结果 + +#### POST /api/swarm/{swarm_id}/stop +停止蜂群执行。 + +**请求示例：** +```json +{ + "reason": "用户取消", + "cleanup": true +} +``` + +#### GET /api/swarm/{swarm_id}/logs +获取所有Agent的日志聚合。 + +### 3. 编排策略 + +- **sequential**: 顺序执行（architect → coder → reviewer） +- **parallel**: 并行执行（所有agents同时工作） +- **hybrid**: 混合模式（部分顺序、部分并行） + +### 4. K8s资源管理 + +每个Agent运行在独立的K8s namespace中： +- Namespace命名：`swarm-{swarm_id_prefix}-{role}` +- Pod命名：`agent-{agent_id}` +- Service类型：ClusterIP（内部通信） +- 资源限制：CPU 100m-500m, Memory 256Mi-512Mi + +### 5. Agent通信 + +Agents通过A2A协议（HTTP）直接通信： +- 消息格式：JSON-RPC 2.0 +- 传输方式：HTTP POST +- 支持流式：SSE (Server-Sent Events) + +## 使用示例 + +### Python客户端 + +```python +import requests + +# 创建蜂群 +response = requests.post("http://localhost:8000/api/swarm/create", json={ + "task_description": "实现TODO API", + "agents": [ + {"role": "architect", "template": "a2a_litellm_agent"}, + {"role": "coder", "template": "code_manager_agent"} + ], + "orchestration": {"strategy": "sequential"} +}) + +swarm_id = response.json()["swarm_id"] + +# 监听结果 +response = requests.get( + f"http://localhost:8000/api/swarm/{swarm_id}/results", + stream=True +) + +for line in response.iter_lines(): + if line.startswith(b'data: '): + event = json.loads(line[6:]) + print(f"Event: {event['type']}") +``` + +### 测试脚本 + +运行测试脚本验证API： + +```bash +conda activate data && python test_swarm_api.py +``` + +## 技术栈 + +- **FastAPI**: REST API框架 +- **SQLAlchemy**: ORM数据库访问 +- **Kubernetes Python Client**: K8s资源管理 +- **aiohttp**: 异步HTTP客户端 +- **PostgreSQL**: 数据持久化 + +## 文件结构 + +``` +agent-manager/ +├── database.py # 新增Swarm相关模型 +├── k8s_manager.py # 新增swarm方法 +├── app.py # 注册swarm router +├── api/swarm/ +│ ├── __init__.py +│ ├── models.py # Pydantic请求/响应模型 +│ ├── router.py # API端点 +│ ├── orchestrator.py # 核心编排逻辑 +│ └── agent_client.py # A2A客户端 +└── test_swarm_api.py # 测试脚本 +``` + +## 下一步 + +### Phase 5: Agent模板优化 +- 优化a2a_litellm_agent支持swarm环境变量 +- 添加状态上报机制 +- 实现Agent间消息路由 + +### Phase 6: 集成测试 +- 端到端测试完整流程 +- 测试错误处理和重试 +- 性能和资源测试 + +### Phase 7: Claude Code MCP集成 +- 创建MCP server +- 定义swarm工具 +- 实现流式结果返回 + +## 限制和注意事项 + +1. **当前实现为MVP版本**，Agent Pods实际部署需要K8s集群环境 +2. **后台任务执行**使用FastAPI BackgroundTasks，生产环境建议使用Celery +3. **Agent间通信**假设所有agents在同一K8s集群内 +4. **资源清理**需要手动调用stop接口，未实现自动TTL +5. **认证授权**未实现，需要在生产环境添加 + +## 贡献 + +欢迎提交Issue和Pull Request！ + +## License + +MIT diff --git a/test_swarm_api.py b/test_swarm_api.py new file mode 100644 index 0000000..87f9896 --- /dev/null +++ b/test_swarm_api.py @@ -0,0 +1,232 @@ +#!/usr/bin/env python3 +""" +Test script for Swarm API endpoints. +""" + +import requests +import json +import time + +BASE_URL = "http://localhost:8000" + + +def test_create_swarm(): + """Test creating a swarm""" + print("\n=== Testing Swarm Creation ===") + + payload = { + "task_description": "实现用户认证模块，包括登录、注册、密码重置功能", + "project_context": { + "repo_url": "https://github.com/test/project", + "branch": "feature/auth", + "language": "python", + "framework": "fastapi" + }, + "agents": [ + { + "role": "architect", + "template": "a2a_litellm_agent", + "model": "gpt-4", + "capabilities": ["design", "planning"], + "system_prompt": "你是架构师，负责设计系统架构和API接口", + "replicas": 1 + }, + { + "role": "coder", + "template": "code_manager_agent", + "model": "gpt-4", + "capabilities": ["coding", "git"], + "system_prompt": "你是开发工程师，负责实现代码", + "replicas": 2 + }, + { + "role": "reviewer", + "template": "a2a_litellm_agent", + "model": "gpt-4", + "capabilities": ["review", "testing"], + "system_prompt": "你是代码审查员，负责审查代码质量", + "replicas": 1 + } + ], + "orchestration": { + "strategy": "sequential", + "max_iterations": 3, + "timeout_minutes": 30 + }, + "owner_id": "test_user" + } + + try: + response = requests.post(f"{BASE_URL}/api/swarm/create", json=payload) + response.raise_for_status() + result = response.json() + + print(f"✅ Swarm created successfully!") + print(f" Swarm ID: {result['swarm_id']}") + print(f" Status: {result['status']}") + print(f" Agents: {len(result['agents'])}") + + for agent in result['agents']: + print(f" - {agent['role']}: {agent['agent_id']} ({agent['status']})") + + return result['swarm_id'] + + except requests.exceptions.RequestException as e: + print(f"❌ Failed to create swarm: {e}") + if hasattr(e.response, 'text'): + print(f" Response: {e.response.text}") + return None + + +def test_get_swarm_status(swarm_id): + """Test getting swarm status""" + print(f"\n=== Testing Swarm Status (ID: {swarm_id}) ===") + + try: + response = requests.get(f"{BASE_URL}/api/swarm/{swarm_id}/status") + response.raise_for_status() + result = response.json() + + print(f"✅ Swarm status retrieved!") + print(f" Status: {result['status']}") + print(f" Phase: {result.get('phase', 'N/A')}") + print(f" Progress: {result['progress']}%") + print(f" Total Messages: {result['metrics']['total_messages']}") + print(f" Elapsed: {result['metrics']['elapsed_seconds']}s") + + return result + + except requests.exceptions.RequestException as e: + print(f"❌ Failed to get swarm status: {e}") + return None + + +def test_stream_swarm_results(swarm_id, duration=10): + """Test streaming swarm results""" + print(f"\n=== Testing Swarm Results Stream (ID: {swarm_id}) ===") + print(f"Streaming for {duration} seconds...") + + try: + response = requests.get( + f"{BASE_URL}/api/swarm/{swarm_id}/results", + stream=True, + timeout=duration + 5 + ) + response.raise_for_status() + + start_time = time.time() + event_count = 0 + + for line in response.iter_lines(): + if time.time() - start_time > duration: + break + + if line: + line = line.decode('utf-8') + if line.startswith('data: '): + event_count += 1 + data = json.loads(line[6:]) + event_type = data.get('type', 'unknown') + print(f" Event #{event_count}: {event_type}") + + if event_type == 'phase_change': + print(f" Phase: {data.get('phase')}") + elif event_type == 'agent_message': + print(f" From: {data.get('from_agent_id', 'orchestrator')}") + print(f" To: {data.get('to_agent_id', 'orchestrator')}") + elif event_type == 'result': + print(f" Final Status: {data.get('status')}") + break + + print(f"✅ Received {event_count} events") + + except requests.exceptions.RequestException as e: + print(f"❌ Failed to stream results: {e}") + + +def test_stop_swarm(swarm_id): + """Test stopping a swarm""" + print(f"\n=== Testing Swarm Stop (ID: {swarm_id}) ===") + + payload = { + "reason": "Test completed", + "cleanup": False # Don't cleanup for testing + } + + try: + response = requests.post(f"{BASE_URL}/api/swarm/{swarm_id}/stop", json=payload) + response.raise_for_status() + result = response.json() + + print(f"✅ Swarm stopped successfully!") + print(f" Status: {result['status']}") + print(f" Stopped at: {result['stopped_at']}") + + return result + + except requests.exceptions.RequestException as e: + print(f"❌ Failed to stop swarm: {e}") + return None + + +def test_get_swarm_logs(swarm_id): + """Test getting swarm logs""" + print(f"\n=== Testing Swarm Logs (ID: {swarm_id}) ===") + + try: + response = requests.get(f"{BASE_URL}/api/swarm/{swarm_id}/logs") + response.raise_for_status() + result = response.json() + + print(f"✅ Swarm logs retrieved!") + print(f" Agents: {len(result.get('agents', []))}") + + for agent in result.get('agents', []): + print(f" - {agent['role']} ({agent['agent_id']})") + + return result + + except requests.exceptions.RequestException as e: + print(f"❌ Failed to get swarm logs: {e}") + return None + + +def main(): + """Run all tests""" + print("=" * 60) + print("Swarm API Test Suite") + print("=" * 60) + + # Test 1: Create swarm + swarm_id = test_create_swarm() + if not swarm_id: + print("\n❌ Cannot continue tests without swarm_id") + return + + # Wait a bit for initialization + print("\nWaiting 2 seconds for initialization...") + time.sleep(2) + + # Test 2: Get status + test_get_swarm_status(swarm_id) + + # Test 3: Stream results (for 10 seconds) + test_stream_swarm_results(swarm_id, duration=10) + + # Test 4: Get logs + test_get_swarm_logs(swarm_id) + + # Test 5: Stop swarm + test_stop_swarm(swarm_id) + + # Final status check + print("\nFinal status check...") + test_get_swarm_status(swarm_id) + + print("\n" + "=" * 60) + print("Test Suite Completed!") + print("=" * 60) + + +if __name__ == "__main__": + main()