diff --git a/AGENTS.md b/AGENTS.md new file mode 100644 index 0000000..a369add --- /dev/null +++ b/AGENTS.md @@ -0,0 +1,11 @@ +# Agent Manager Scope + +This repository only owns **Heicode sub-mode runtime** behavior. + +Rules for all files under this repository: + +- Do not add or restore standalone swarm-mode product features. +- Treat `/api/swarms` as a **sub-mode compatibility API**, not a generic swarm product API. +- Do not add `/api/swarm/*` endpoints, swarm-only docs, or swarm-only tests. +- When refactoring, prefer names and comments that reflect **sub-mode runtime** ownership. +- If a feature belongs to the separate swarm system, remove or reject it here instead of integrating it. diff --git a/QUICKSTART.md b/QUICKSTART.md deleted file mode 100644 index da857a1..0000000 --- a/QUICKSTART.md +++ /dev/null @@ -1,150 +0,0 @@ -# 蜂群模式快速开始 - -## 一句话方案 - -**将蜂群的去中心化任务分发、动态角色分配和自组织协作机制，通过在agent-manager中引入任务广播队列、agent能力注册表和竞价响应机制来实现，让agents像蜂群一样根据自身能力主动认领任务而非被动分配。** - -## 实现方式 - -传统方式：Claude Code在本地启动子agent -新方案：Claude Code调用agent-manager API，在K8s上动态创建多个specialized agents协作编码 - -## 快速测试 - -### 1. 启动服务 - -```bash -conda activate data && python app.py -``` - -### 2. 创建蜂群 - -```bash -curl -X POST http://localhost:8000/api/swarm/create \ - -H "Content-Type: application/json" \ - -d '{ - "task_description": "实现简单的TODO API", - "agents": [ - {"role": "architect", "template": "a2a_litellm_agent", "replicas": 1}, - {"role": "coder", "template": "code_manager_agent", "replicas": 2}, - {"role": "reviewer", "template": "a2a_litellm_agent", "replicas": 1} - ], - "orchestration": {"strategy": "sequential"} - }' -``` - -### 3. 查询状态 - -```bash -curl http://localhost:8000/api/swarm/{swarm_id}/status -``` - -### 4. 监听结果（SSE） - -```bash -curl -N http://localhost:8000/api/swarm/{swarm_id}/results -``` - -### 5. 运行测试脚本 - -```bash -python test_swarm_api.py -``` - -## 核心组件 - -### 数据库模型 -- `Swarm`: 蜂群主表 -- `SwarmAgent`: Agent实例 -- `SwarmMessage`: 通信消息 - -### API端点 -- `POST /api/swarm/create` - 创建蜂群 -- `GET /api/swarm/{id}/status` - 查询状态 -- `GET /api/swarm/{id}/results` - SSE流式结果 -- `POST /api/swarm/{id}/stop` - 停止蜂群 -- `GET /api/swarm/{id}/logs` - 获取日志 - -### 编排策略 -- **Sequential**: architect → coder → reviewer（顺序执行） -- **Parallel**: 所有agents并行工作 -- **Hybrid**: 混合模式 - -## 架构特点 - -1. **K8s原生**: 每个agent独立namespace和pod -2. **A2A通信**: 基于HTTP的agent间通信协议 -3. **实时追踪**: SSE流式推送进度和结果 -4. **灵活编排**: 支持多种协作策略 -5. **资源隔离**: 独立的K8s资源和配额管理 - -## 下一步集成 - -### Claude Code MCP工具 - -```json -{ - "name": "create_code_swarm", - "description": "在K8s上创建Agent蜂群协作编码", - "inputSchema": { - "type": "object", - "properties": { - "task": {"type": "string"}, - "agents": {"type": "array"} - } - } -} -``` - -### 使用示例 - -```python -# Claude Code调用 -result = await mcp.call_tool("create_code_swarm", { - "task": "实现用户认证模块", - "agents": [ - {"role": "architect", "model": "gpt-4"}, - {"role": "coder", "model": "gpt-4", "replicas": 2} - ] -}) - -# 监听结果 -async for event in mcp.stream_results(result["swarm_id"]): - if event["type"] == "code_diff": - apply_diff(event["diff"]) -``` - -## 技术债务和改进 - -### 当前限制 -1. Agent Pods实际部署需要K8s集群（当前为模拟） -2. 后台任务使用FastAPI BackgroundTasks（生产建议Celery） -3. 无认证授权机制 -4. 无自动资源清理TTL - -### 建议改进 -1. 实现真实的K8s pod部署和监控 -2. 添加认证和授权机制 -3. 实现自动资源清理和TTL -4. 添加更多编排策略（如基于依赖的DAG执行） -5. 实现agent能力注册和任务竞价机制 - -## 文档 - -详细文档请参考： -- [SWARM_README.md](SWARM_README.md) - 完整文档 -- [test_swarm_api.py](test_swarm_api.py) - API测试示例 - -## Git提交 - -```bash -git log --oneline feature/swarm-mode -# b4b20f0 feat: implement swarm mode for multi-agent collaboration -# 2657ef2 docs: add swarm mode documentation and test script -``` - ---- - -**实现完成！** 🎉 - -核心功能已就绪，可以开始集成到Claude Code或进行进一步的测试和优化。 diff --git a/SWARM_README.md b/SWARM_README.md deleted file mode 100644 index 63c318a..0000000 --- a/SWARM_README.md +++ /dev/null @@ -1,247 +0,0 @@ -# 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/api/swarm/__init__.py b/api/swarm/__init__.py index 2a0f633..856bcd0 100644 --- a/api/swarm/__init__.py +++ b/api/swarm/__init__.py @@ -1,7 +1,5 @@ -""" -Swarm API module for multi-agent collaboration. -""" +"""Sub-mode runtime compatibility module.""" -from .router import router +from .router import swarms_router -__all__ = ["router"] +__all__ = ["swarms_router"] diff --git a/api/swarm/models.py b/api/swarm/models.py index 5f2f7e4..9fb639b 100644 --- a/api/swarm/models.py +++ b/api/swarm/models.py @@ -1,14 +1,12 @@ -""" -Pydantic models for Swarm API requests and responses. -""" +"""Pydantic models for sub-mode runtime compatibility requests and responses.""" from typing import List, Optional, Dict, Any -from pydantic import BaseModel, Field +from pydantic import BaseModel, ConfigDict, Field, model_validator from datetime import datetime class AgentConfig(BaseModel): - """Agent configuration for swarm""" + """Agent configuration for sub-mode runtime execution.""" role: str = Field(..., description="Agent role (architect/coder/reviewer/tester)") template: str = Field(default="a2a_litellm_agent", description="Agent template name") model: str = Field(default="gpt-4", description="Model name") @@ -16,6 +14,19 @@ class AgentConfig(BaseModel): system_prompt: Optional[str] = Field(None, description="Custom system prompt") replicas: int = Field(default=1, description="Number of agent replicas") + @model_validator(mode="before") + @classmethod + def normalize_sub_mode_agent(cls, data: Any) -> Any: + """Accept Manager sub-mode agent fields when /api/swarms is used.""" + if not isinstance(data, dict): + return data + data = dict(data) + if not data.get("role"): + data["role"] = data.get("role_template") or data.get("target_role") or "worker" + if not data.get("model"): + data["model"] = data.get("default_model_id") or data.get("model_ref") or "gpt-4" + return data + class OrchestrationConfig(BaseModel): """Orchestration configuration""" @@ -26,20 +37,29 @@ class OrchestrationConfig(BaseModel): class ProjectContext(BaseModel): """Project context information""" + model_config = ConfigDict(extra="allow") + repo_url: Optional[str] = Field(None, description="Repository URL") branch: Optional[str] = Field(None, description="Git branch") language: Optional[str] = Field(None, description="Programming language") framework: Optional[str] = Field(None, description="Framework") + intent_id: Optional[str] = None + template_hint: Optional[str] = None + sub_mode: Optional[str] = None + agile_context: Optional[Dict[str, Any]] = None + correlation_id: Optional[str] = None class CallbackConfig(BaseModel): """Callback configuration""" url: str = Field(..., description="Callback URL") method: str = Field(default="POST", description="HTTP method") + signing_secret_ref: Optional[str] = Field(None, description="Azure Key Vault reference for callback HMAC signing") + subscribed_events: Optional[List[str]] = Field(None, description="Runtime event types subscribed by Manager") class SwarmCreateRequest(BaseModel): - """Request model for creating a swarm""" + """Request model for creating a sub-mode runtime run.""" task_description: str = Field(..., description="Task description") project_context: Optional[ProjectContext] = Field(None, description="Project context") agents: List[AgentConfig] = Field(..., description="Agent configurations") @@ -49,7 +69,7 @@ class SwarmCreateRequest(BaseModel): class SwarmAgentInfo(BaseModel): - """Swarm agent information""" + """Sub-mode runtime agent information.""" agent_id: str role: str status: str @@ -60,7 +80,8 @@ class SwarmAgentInfo(BaseModel): class SwarmCreateResponse(BaseModel): - """Response model for swarm creation""" + """Response model for sub-mode runtime creation.""" + deployment_id: Optional[str] = None swarm_id: str status: str agents: List[SwarmAgentInfo] @@ -69,14 +90,15 @@ class SwarmCreateResponse(BaseModel): class SwarmMetrics(BaseModel): - """Swarm metrics""" + """Sub-mode runtime metrics.""" total_messages: int tokens_used: int elapsed_seconds: int class SwarmStatusResponse(BaseModel): - """Response model for swarm status""" + """Response model for sub-mode runtime status.""" + deployment_id: Optional[str] = None swarm_id: str status: str phase: Optional[str] = None @@ -85,16 +107,35 @@ class SwarmStatusResponse(BaseModel): metrics: SwarmMetrics artifacts: List[Dict[str, Any]] = [] error_message: Optional[str] = None + created_at: Optional[datetime] = None + updated_at: Optional[datetime] = None class SwarmStopRequest(BaseModel): - """Request model for stopping a swarm""" + """Request model for stopping a sub-mode runtime run.""" reason: Optional[str] = Field(None, description="Reason for stopping") cleanup: bool = Field(default=True, description="Whether to cleanup K8s resources") class SwarmStopResponse(BaseModel): - """Response model for swarm stop""" + """Response model for stopping a sub-mode runtime run.""" + deployment_id: Optional[str] = None swarm_id: str status: str stopped_at: datetime + + +class ApprovalDecisionRequest(BaseModel): + """Manager approval decision for a paused high-risk Runtime action.""" + approval_id: str + decision: str = Field(..., description="approved or rejected") + manager_deployment_id: Optional[str] = None + runtime_deployment_id: Optional[str] = None + operation: Optional[str] = None + resource_id: Optional[str] = None + resource_type: Optional[str] = None + target_role: Optional[str] = None + requires_credential: bool = False + credential_ref: Optional[str] = None + lease_id: Optional[str] = None + lease_expires_at: Optional[int] = None diff --git a/api/swarm/orchestrator.py b/api/swarm/orchestrator.py index 1c791b8..c1edde1 100644 --- a/api/swarm/orchestrator.py +++ b/api/swarm/orchestrator.py @@ -1,6 +1,4 @@ -""" -Swarm Orchestrator - Core logic for multi-agent collaboration. -""" +"""Sub-mode runtime orchestrator.""" import asyncio import json @@ -29,7 +27,7 @@ PHASE_MAP = { class SwarmOrchestrator: - """Swarm orchestrator for managing multi-agent collaboration""" + """Runtime orchestrator for Heicode sub-mode execution.""" def __init__(self, swarm_id: str, db: Session): """ diff --git a/api/swarm/router.py b/api/swarm/router.py index 240e19b..1ca8c82 100644 --- a/api/swarm/router.py +++ b/api/swarm/router.py @@ -1,14 +1,10 @@ -""" -Swarm API router - REST endpoints for swarm management. -""" +"""Sub-mode runtime compatibility router.""" -import asyncio import json import uuid from datetime import datetime, timedelta -from typing import AsyncGenerator, Dict, Any +from typing import Dict, Any from fastapi import APIRouter, Depends, HTTPException, BackgroundTasks, Request -from fastapi.responses import StreamingResponse from sqlalchemy.orm import Session from database import ( @@ -22,7 +18,6 @@ from .models import ( ) from .orchestrator import SwarmOrchestrator -router = APIRouter(prefix="/api/swarm", tags=["swarm"]) swarms_router = APIRouter(prefix="/api/swarms", tags=["swarms"]) @@ -54,7 +49,7 @@ def _agent_infos_for_swarm(db: Session, swarm_id: str) -> list[SwarmAgentInfo]: def _build_swarm_status_response(db: Session, swarm: Swarm) -> SwarmStatusResponse: - """Return a standard status payload for /api/swarm and /api/swarms.""" + """Return a standard status payload for the sub-mode runtime compatibility API.""" elapsed_seconds = int((datetime.utcnow() - swarm.created_at).total_seconds()) return SwarmStatusResponse( deployment_id=swarm.swarm_id, @@ -100,13 +95,7 @@ def _stop_swarm_record(db: Session, swarm_id: str, request: SwarmStopRequest) -> async def initialize_and_execute_swarm(swarm_id: str, db_url: str): - """ - Background task to initialize and execute swarm. - - Args: - swarm_id: Swarm ID - db_url: Database URL for creating new session - """ + """Background task to initialize and execute a sub-mode runtime run.""" from sqlalchemy import create_engine from sqlalchemy.orm import sessionmaker @@ -133,23 +122,12 @@ async def initialize_and_execute_swarm(swarm_id: str, db_url: str): db.close() -@router.post("/create", response_model=SwarmCreateResponse) async def create_swarm( request: SwarmCreateRequest, background_tasks: BackgroundTasks, db: Session = Depends(get_db) ): - """ - Create a new swarm. - - Args: - request: Swarm creation request - background_tasks: FastAPI background tasks - db: Database session - - Returns: - Swarm creation response - """ + """Create a new sub-mode runtime run.""" swarm_id = generate_swarm_id() project_context = request.project_context.dict() if request.project_context else {} @@ -228,12 +206,7 @@ async def create_swarm_compat( background_tasks: BackgroundTasks, db: Session = Depends(get_db) ): - """Compatibility entrypoint for Heicode sub-mode Runtime adapters. - - Accepts the Manager's structured orchestration_plan payload and maps it to - the existing swarm creation path without forcing Manager to call - /api/swarm/create directly. - """ + """Compatibility entrypoint for Heicode sub-mode Runtime adapters.""" if payload.get("dry_run") is True: raise HTTPException(status_code=422, detail="dry_run is not supported by Runtime create; no swarm was created") @@ -318,25 +291,6 @@ async def get_swarm_detail_compat(swarm_id: str, db: Session = Depends(get_db)): return _build_swarm_status_response(db, swarm) -@router.get("/{swarm_id}/status", response_model=SwarmStatusResponse) -async def get_swarm_status(swarm_id: str, db: Session = Depends(get_db)): - """ - Get swarm status. - - Args: - swarm_id: Swarm ID - db: Database session - - Returns: - Swarm status response - """ - swarm = db.query(Swarm).filter(Swarm.swarm_id == swarm_id).first() - if not swarm: - raise HTTPException(status_code=404, detail="Swarm not found") - - return _build_swarm_status_response(db, swarm) - - @swarms_router.get("/{swarm_id}/status", response_model=SwarmStatusResponse) async def get_swarm_status_compat(swarm_id: str, db: Session = Depends(get_db)): """Compatibility status endpoint for Manager Runtime bridge.""" @@ -346,104 +300,6 @@ async def get_swarm_status_compat(swarm_id: str, db: Session = Depends(get_db)): return _build_swarm_status_response(db, swarm) -@router.get("/{swarm_id}/results") -async def stream_swarm_results(swarm_id: str, db: Session = Depends(get_db)): - """ - Stream swarm results via SSE. - - Args: - swarm_id: Swarm ID - db: Database session - - Returns: - SSE stream - """ - swarm = db.query(Swarm).filter(Swarm.swarm_id == swarm_id).first() - if not swarm: - raise HTTPException(status_code=404, detail="Swarm not found") - - async def event_generator() -> AsyncGenerator[str, None]: - last_message_id = 0 - last_phase = None - - while True: - # Refresh swarm status - db.refresh(swarm) - - # Send phase change event - if swarm.phase != last_phase: - last_phase = swarm.phase - event = { - "type": "phase_change", - "phase": swarm.phase, - "timestamp": datetime.utcnow().isoformat() - } - yield f"data: {json.dumps(event)}\n\n" - - # Get new messages - messages = db.query(SwarmMessage).filter( - SwarmMessage.swarm_id == swarm_id, - SwarmMessage.id > last_message_id - ).order_by(SwarmMessage.id).all() - - for msg in messages: - last_message_id = msg.id - event = { - "type": "agent_message", - "message_id": msg.message_id, - "from_agent_id": msg.from_agent_id, - "to_agent_id": msg.to_agent_id, - "message_type": msg.message_type, - "content": msg.content, - "timestamp": msg.created_at.isoformat() - } - yield f"data: {json.dumps(event)}\n\n" - - # Check if completed - if swarm.status in [SwarmStatus.COMPLETED, SwarmStatus.FAILED, SwarmStatus.STOPPED]: - result_event = { - "type": "result", - "status": swarm.status.value, - "artifacts": swarm.artifacts, - "error_message": swarm.error_message, - "timestamp": datetime.utcnow().isoformat() - } - yield f"data: {json.dumps(result_event)}\n\n" - break - - await asyncio.sleep(1) - - return StreamingResponse( - event_generator(), - media_type="text/event-stream", - headers={ - "Cache-Control": "no-cache", - "Connection": "keep-alive", - "X-Accel-Buffering": "no" - } - ) - - -@router.post("/{swarm_id}/stop", response_model=SwarmStopResponse) -async def stop_swarm( - swarm_id: str, - request: SwarmStopRequest, - db: Session = Depends(get_db) -): - """ - Stop a swarm. - - Args: - swarm_id: Swarm ID - request: Stop request - db: Database session - - Returns: - Stop response - """ - return _stop_swarm_record(db, swarm_id, request) - - @swarms_router.post("/{swarm_id}/stop", response_model=SwarmStopResponse) async def stop_swarm_compat( swarm_id: str, @@ -454,18 +310,8 @@ async def stop_swarm_compat( return _stop_swarm_record(db, swarm_id, request) -@router.get("/{swarm_id}/logs") async def get_swarm_logs(swarm_id: str, db: Session = Depends(get_db)): - """ - Get aggregated logs from all agents in swarm. - - Args: - swarm_id: Swarm ID - db: Database session - - Returns: - Aggregated logs - """ + """Get aggregated logs from all agents in a sub-mode runtime run.""" swarm = db.query(Swarm).filter(Swarm.swarm_id == swarm_id).first() if not swarm: raise HTTPException(status_code=404, detail="Swarm not found") diff --git a/app.py b/app.py index d84ff73..3e0b09b 100644 --- a/app.py +++ b/app.py @@ -42,9 +42,9 @@ app.include_router(external_tool_router) from api.agnet.router import router as agnet_router app.include_router(agnet_router) -# 注册 Swarm API Router -from api.swarm.router import router as swarm_router -app.include_router(swarm_router) +# 注册 Heicode sub-mode Runtime 兼容 Router +from api.swarm.router import swarms_router +app.include_router(swarms_router) # 初始化K8s管理器 NAMESPACE = os.getenv("NAMESPACE", "ai-agents") diff --git a/database.py b/database.py index e457ae0..fe0d635 100644 --- a/database.py +++ b/database.py @@ -378,7 +378,7 @@ class AuditLog(Base): # ============================================================================ -# Swarm Mode Models (NEW) +# Sub-mode Runtime Internal Models # ============================================================================ class SwarmStatus(str, enum.Enum): @@ -399,7 +399,7 @@ class SwarmAgentStatus(str, enum.Enum): class Swarm(Base): - """Swarm model for multi-agent collaboration""" + """Internal runtime run model used by Heicode sub-mode execution.""" __tablename__ = "swarms" id = Column(Integer, primary_key=True, index=True) @@ -445,7 +445,7 @@ class Swarm(Base): class SwarmAgent(Base): - """Swarm agent model for agents in a swarm""" + """Internal runtime agent model for sub-mode execution.""" __tablename__ = "swarm_agents" id = Column(Integer, primary_key=True, index=True) @@ -479,7 +479,7 @@ class SwarmAgent(Base): class SwarmMessage(Base): - """Swarm message model for agent-to-agent communication""" + """Internal runtime message model for sub-mode execution.""" __tablename__ = "swarm_messages" id = Column(Integer, primary_key=True, index=True) @@ -491,7 +491,7 @@ class SwarmMessage(Base): to_agent_id = Column(String(100), index=True) # NULL for broadcast message_type = Column(String(50)) # task, response, broadcast, artifact content = Column(Text) - metadata = Column(JSON) + message_metadata = Column("metadata", JSON) # Timestamp created_at = Column(DateTime, default=datetime.utcnow, index=True) diff --git a/k8s_manager.py b/k8s_manager.py index 3eaa534..529f151 100644 --- a/k8s_manager.py +++ b/k8s_manager.py @@ -1306,7 +1306,7 @@ class K8sManager: raise Exception(f"列出Pod失败: {e.reason}") # ============================================================================ - # Swarm Mode Methods (NEW) + # Sub-mode runtime helper methods # ============================================================================ def create_swarm_namespace(self, swarm_id: str, role: str) -> str: diff --git a/test_swarm_api.py b/test_swarm_api.py deleted file mode 100644 index 87f9896..0000000 --- a/test_swarm_api.py +++ /dev/null @@ -1,232 +0,0 @@ -#!/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()