agent-swarm-orchestrator

What this skill does

# Agent Swarm Orchestrator

This skill provides guidance for designing multi-agent systems where multiple AI agents coordinate to accomplish complex tasks through distributed execution and emergent behavior.

## Core Competencies

- **Swarm Architecture**: Agent topologies, communication patterns
- **Task Distribution**: Work allocation, load balancing
- **Coordination Protocols**: Consensus, voting, delegation
- **Emergent Behavior**: Collective intelligence from simple rules

## Multi-Agent Fundamentals

### Why Multi-Agent Systems

```
Single Agent:                    Multi-Agent Swarm:
┌─────────────────┐             ┌─────────────────────────────┐
│                 │             │  ┌───┐ ┌───┐ ┌───┐ ┌───┐   │
│   One Agent     │             │  │ A │ │ A │ │ A │ │ A │   │
│   Sequential    │     vs      │  └───┘ └───┘ └───┘ └───┘   │
│   Single POV    │             │  Parallel, Diverse POV     │
│                 │             │  Specialization possible    │
└─────────────────┘             └─────────────────────────────┘

Benefits:
- Parallelism: Multiple agents work simultaneously
- Specialization: Agents can have different capabilities
- Resilience: System continues if one agent fails
- Diverse perspectives: Multiple approaches to problems
```

### Agent Roles

| Role | Responsibility | Characteristics |
|------|----------------|-----------------|
| Orchestrator | Coordinate swarm | Global view, task assignment |
| Worker | Execute tasks | Specialized skills, focused |
| Supervisor | Quality control | Review, approve, redirect |
| Specialist | Domain expertise | Deep knowledge, narrow scope |
| Scout | Exploration | Information gathering, research |

## Swarm Topologies

### Hierarchical

```
                    ┌──────────────┐
                    │ Orchestrator │
                    └──────┬───────┘
                           │
          ┌────────────────┼────────────────┐
          │                │                │
    ┌─────┴─────┐    ┌─────┴─────┐    ┌─────┴─────┐
    │Supervisor │    │Supervisor │    │Supervisor │
    └─────┬─────┘    └─────┬─────┘    └─────┬─────┘
          │                │                │
    ┌─────┼─────┐    ┌─────┼─────┐    ┌─────┼─────┐
    │     │     │    │     │     │    │     │     │
   ┌┴┐   ┌┴┐   ┌┴┐  ┌┴┐   ┌┴┐   ┌┴┐  ┌┴┐   ┌┴┐   ┌┴┐
   │W│   │W│   │W│  │W│   │W│   │W│  │W│   │W│   │W│
   └─┘   └─┘   └─┘  └─┘   └─┘   └─┘  └─┘   └─┘   └─┘
         Workers

Best for: Clear task decomposition, quality control needed
```

### Peer-to-Peer

```
        ┌───┐───────────────┌───┐
        │ A │               │ A │
        └───┘               └───┘
          │ \             / │
          │   \         /   │
          │     \     /     │
          │       \ /       │
        ┌───┐     ╳       ┌───┐
        │ A │   /   \     │ A │
        └───┘ /       \   └───┘
            /           \
        ┌───┐           ┌───┐
        │ A │───────────│ A │
        └───┘           └───┘

Best for: Collaborative problem-solving, no single point of failure
```

### Blackboard

```
┌─────────────────────────────────────────────────────┐
│                     Blackboard                       │
│  ┌─────────────┐ ┌─────────────┐ ┌───────────────┐ │
│  │ Problem     │ │ Partial     │ │ Solutions     │ │
│  │ State       │ │ Results     │ │               │ │
│  └─────────────┘ └─────────────┘ └───────────────┘ │
└───────────────────────┬─────────────────────────────┘
                        │
    ┌───────────────────┼───────────────────┐
    │         Read/Write│                   │
    ▼                   ▼                   ▼
┌───────┐          ┌───────┐          ┌───────┐
│Agent A│          │Agent B│          │Agent C│
│Analyst│          │Builder│          │Critic │
└───────┘          └───────┘          └───────┘

Best for: Complex problems, agents contribute asynchronously
```

## Agent Implementation

### Base Agent Structure

```python
from abc import ABC, abstractmethod
from dataclasses import dataclass, field
from typing import Any, Optional, List
from enum import Enum
import asyncio

class AgentStatus(Enum):
    IDLE = "idle"
    WORKING = "working"
    WAITING = "waiting"
    COMPLETED = "completed"
    FAILED = "failed"

@dataclass
class AgentMessage:
    sender_id: str
    recipient_id: str  # Or "broadcast"
    message_type: str
    content: Any
    timestamp: float = field(default_factory=lambda: time.time())
    correlation_id: Optional[str] = None

@dataclass
class Task:
    id: str
    description: str
    priority: int = 0
    dependencies: List[str] = field(default_factory=list)
    assigned_to: Optional[str] = None
    status: str = "pending"
    result: Any = None


class BaseAgent(ABC):
    """Base class for swarm agents"""

    def __init__(self, agent_id: str, capabilities: List[str]):
        self.id = agent_id
        self.capabilities = capabilities
        self.status = AgentStatus.IDLE
        self.message_queue: asyncio.Queue = asyncio.Queue()
        self.current_task: Optional[Task] = None

    @abstractmethod
    async def process_task(self, task: Task) -> Any:
        """Process assigned task - implement in subclass"""
        pass

    @abstractmethod
    def can_handle(self, task: Task) -> bool:
        """Check if agent can handle this task type"""
        pass

    async def receive_message(self, message: AgentMessage):
        """Add message to queue for processing"""
        await self.message_queue.put(message)

    async def run(self):
        """Main agent loop"""
        while True:
            # Check for messages
            try:
                message = await asyncio.wait_for(
                    self.message_queue.get(),
                    timeout=0.1
                )
                await self._handle_message(message)
            except asyncio.TimeoutError:
                pass

            # Work on current task
            if self.current_task and self.status == AgentStatus.WORKING:
                await self._work_on_task()

    async def _handle_message(self, message: AgentMessage):
        """Handle incoming message"""
        if message.message_type == "assign_task":
            self.current_task = message.content
            self.status = AgentStatus.WORKING
        elif message.message_type == "cancel_task":
            self.current_task = None
            self.status = AgentStatus.IDLE
        elif message.message_type == "status_request":
            await self._send_status(message.sender_id)

    async def _work_on_task(self):
        """Execute current task"""
        try:
            result = await self.process_task(self.current_task)
            self.current_task.result = result
            self.current_task.status = "completed"
            self.status = AgentStatus.COMPLETED
        except Exception as e:
            self.current_task.status = "failed"
            self.status = AgentStatus.FAILED
```

### Specialized Agents

```python
class ResearchAgent(BaseAgent):
    """Agent specialized for information gathering"""

    def __init__(self, agent_id: str):
        super().__init__(agent_id, ["research", "search", "analyze"])
        self.search_tools = []

    def can_handle(self, task: Task) -> bool:
        return any(cap in task.description.lower()
                   for cap in ["research", "find", "search", "investigate"])

    async def process_task(self, task: Task) -> dict:
        # Research implementation
        results = await self._search(task.description)
        analysis = await self._analyze(results)
        return {
            "sources": results,
            "analysis": analysis,
            "confidence": self._calculate_confidence(results)
        }


class CodeAgent(BaseAgent):
    """Agent specialized for code generation"""

    def __init__(self, agent_id: str):
        super().__init__(agent_id, ["code", "implement", "debug"])

    def can_handle(self, task: Task) -> bool:
        return any(cap in task.description.lower()