docs: add swarm mode documentation and test script
- Add comprehensive SWARM_README.md with architecture and usage - Add test_swarm_api.py for API validation - Document all endpoints, models, and orchestration strategies
This commit is contained in:
elipitc
+247
@@ -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
|
||||
@@ -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()
|
||||
Reference in New Issue
Block a user