Files

T

elipitc 2657ef23db 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

2026-05-17 20:22:51 +08:00

5.6 KiB

Raw Blame History

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

创建新的蜂群。

请求示例：

{
  "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
  }
}

响应示例：

{
  "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

查询蜂群状态。

响应示例：

{
  "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

停止蜂群执行。

请求示例：

{
  "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客户端

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：

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工具
实现流式结果返回

限制和注意事项

当前实现为MVP版本，Agent Pods实际部署需要K8s集群环境
后台任务执行使用FastAPI BackgroundTasks，生产环境建议使用Celery
Agent间通信假设所有agents在同一K8s集群内
资源清理需要手动调用stop接口，未实现自动TTL
认证授权未实现，需要在生产环境添加

贡献

欢迎提交Issue和Pull Request！

License

MIT

5.6 KiB Raw Blame History Unescape Escape