agent-swarm-orchestrator

name	agent-swarm-orchestrator
description	Designs multi-agent systems with coordinated agent swarms, task distribution, inter-agent communication, and emergent collective behavior.
license	MIT

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

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 = 


 ():
    

     ():
        . = agent_id
        .capabilities = capabilities
        .status = AgentStatus.IDLE
        .message_queue: asyncio.Queue = asyncio.Queue()
        .current_task: [Task] = 


      () -> :
        
        


     () -> :
        
        

      ():
        
         .message_queue.put(message)

      ():
        
         :
            
            :
                message =  asyncio.wait_for(
                    .message_queue.get(),
                    timeout=
                )
                 ._handle_message(message)
             asyncio.TimeoutError:
                

            
             .current_task  .status == AgentStatus.WORKING:
                 ._work_on_task()

      ():
        
         message.message_type == :
            .current_task = message.content
            .status = AgentStatus.WORKING
         message.message_type == :
            .current_task = 
            .status = AgentStatus.IDLE
         message.message_type == :
             ._send_status(message.sender_id)

      ():
        
        :
            result =  .process_task(.current_task)
            .current_task.result = result
            .current_task.status = 
            .status = AgentStatus.COMPLETED
         Exception  e:
            .current_task.status = 
            .status = AgentStatus.FAILED

Specialized Agents

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__():
        ().__init__(agent_id, [, , ])

     () -> :
         (cap  task.description.lower()
                    cap  [, , , , ])

      () -> :
        
        code =  ._generate_code(task.description)
        tests =  ._generate_tests(code)
         {
            : code,
            : tests,
            : ._detect_language(code)
        }


 ():
    

     ():
        ().__init__(agent_id, [, , ])

     () -> :
           task.description.lower()

      () -> :
        artifact = task.content  
        issues =  ._find_issues(artifact)
        suggestions =  ._generate_suggestions(issues)
         {
            : (issues) == ,
            : issues,
            : suggestions
        }

Orchestration Patterns

Task-Based Orchestration

class SwarmOrchestrator:
    """Coordinate agent swarm for task completion"""

    def __init__(self):
        self.agents: dict[str, BaseAgent] = {}
        self.task_queue: asyncio.PriorityQueue = asyncio.PriorityQueue()
        self.completed_tasks: dict[str, Task] = {}
        self.message_bus = MessageBus()

    def register_agent(self, agent: BaseAgent):
        """Add agent to swarm"""
        self.agents[agent.id] = agent

    async def submit_task(self, task: Task):
        """Submit task for processing"""
        # Calculate effective priority (lower = higher priority)
        priority = -task.priority  # Negate for min-heap behavior
        await self.task_queue.put((priority, task))

    async def run(self):
        """Main orchestration loop"""
        while True:
            # Get highest priority task
            _, task = await self.task_queue.get()

            # Check dependencies
            if not self._dependencies_met(task):
                
                 .task_queue.put((, task))
                

            
            agent =  ._find_available_agent(task)
             agent:
                 ._assign_task(agent, task)
            :
                
                 asyncio.sleep()
                 .task_queue.put((, task))

     () -> :
        
         dep_id  task.dependencies:
             dep_id   .completed_tasks:
                 
             .completed_tasks[dep_id].status != :
                 
         

      () -> [BaseAgent]:
        
         agent  .agents.values():
             agent.status == AgentStatus.IDLE  agent.can_handle(task):
                 agent
         

      ():
        
        task.assigned_to = agent.
        task.status = 

        message = AgentMessage(
            sender_id=,
            recipient_id=agent.,
            message_type=,
            content=task
        )
         agent.receive_message(message)

Workflow Orchestration

from dataclasses import dataclass
from typing import Callable, List

@dataclass
class WorkflowStep:
    name: str
    agent_type: str
    input_transform: Callable[[dict], dict] = lambda x: x
    required_approval: bool = False

class WorkflowOrchestrator:
    """Execute multi-step workflows with agent swarm"""

    def __init__(self, swarm: SwarmOrchestrator):
        self.swarm = swarm
        self.workflows: dict[str, List[WorkflowStep]] = {}

    def register_workflow(self, name: str, steps: List[WorkflowStep]):
        """Register a multi-step workflow"""
        self.workflows[name] = steps

    async def execute_workflow(
        self,
        workflow_name: str,
        initial_input: dict
    ) -> dict:
        """Execute workflow through agent swarm"""
        steps = self.workflows[workflow_name]
        current_data = initial_input
        results = []

        for i, step in (steps):
            
            step_input = step.input_transform(current_data)

            
            task = Task(
                =,
                description=,
                context=step_input
            )

            
             .swarm.submit_task(task)
            result =  ._wait_for_completion(task.)

            
             step.required_approval:
                approved =  ._request_approval(step, result)
                  approved:
                     {: , : step.name}

            results.append(result)
            current_data = {**current_data, **result}

         {
            : ,
            : results,
            : current_data
        }

Inter-Agent Communication

Message Patterns

class MessageBus:
    """Central message routing for swarm communication"""

    def __init__(self):
        self.subscribers: dict[str, List[BaseAgent]] = {}
        self.message_history: List[AgentMessage] = []

    def subscribe(self, topic: str, agent: BaseAgent):
        """Subscribe agent to topic"""
        if topic not in self.subscribers:
            self.subscribers[topic] = []
        self.subscribers[topic].append(agent)

    async def publish(self, topic: str, message: AgentMessage):
        """Publish message to topic subscribers"""
        self.message_history.append(message)

        for agent in self.subscribers.get(topic, []):
            if agent.id != message.sender_id:  # Don't send to self
                await agent.receive_message(message)

    async def send_direct(self, message: AgentMessage):
        """Send message to specific agent"""
        # Route to recipient
        pass

    async  ():
        
         topic_subscribers  .subscribers.values():
             agent  topic_subscribers:
                 agent. != message.sender_id:
                     agent.receive_message(message)

Consensus Protocol

class ConsensusProtocol:
    """Achieve consensus among agents"""

    def __init__(self, agents: List[BaseAgent], threshold: float = 0.66):
        self.agents = agents
        self.threshold = threshold

    async def vote(self, proposal: Any) -> dict:
        """Collect votes from agents"""
        votes = {}

        for agent in self.agents:
            message = AgentMessage(
                sender_id="consensus",
                recipient_id=agent.id,
                message_type="vote_request",
                content=proposal
            )
            response = await self._request_vote(agent, message)
            votes[agent.id] = response

        return self._tally_votes(votes)

    def _tally_votes(self, votes: dict) -> dict:
        """Calculate consensus result"""
        approve_count = sum(1 for v in votes.values() if v.get("approve"))
        total = len(votes)

        consensus_reached = (approve_count / total) >= self.threshold

        return {
            : consensus_reached,
            : approve_count,
            : total - approve_count,
            : .threshold,
            : votes
        }

Emergent Behavior

Stigmergy Pattern

Indirect coordination through environment modification:

class SharedWorkspace:
    """Shared environment for stigmergic coordination"""

    def __init__(self):
        self.artifacts: dict[str, Any] = {}
        self.pheromones: dict[str, float] = {}  # Strength indicators
        self.decay_rate = 0.1

    def deposit(self, key: str, artifact: Any, strength: float = 1.0):
        """Add artifact to shared space"""
        self.artifacts[key] = artifact
        self.pheromones[key] = strength

    def sense(self, pattern: str) -> List[tuple[str, Any, float]]:
        """Find artifacts matching pattern"""
        matches = []
        for key, artifact in self.artifacts.items():
            if pattern in key:
                strength = self.pheromones.get(key, 0)
                matches.append((key, artifact, strength))
        return sorted(matches, key=lambda x: -x[])  

     ():
        
         key  .pheromones:
            .pheromones[key] *= ( - .decay_rate)
             .pheromones[key] < :
                 .pheromones[key]
                 .artifacts[key]

     ():
        
         key  .pheromones:
            .pheromones[key] = (, .pheromones[key] + amount)

Ant Colony Optimization

class AntColonyTaskAllocator:
    """Allocate tasks using ant colony optimization"""

    def __init__(self, agents: List[BaseAgent], tasks: List[Task]):
        self.agents = agents
        self.tasks = tasks
        self.pheromones = {}  # (agent_id, task_id) -> strength
        self.alpha = 1.0  # Pheromone importance
        self.beta = 2.0   # Heuristic importance
        self.evaporation = 0.1

    def _calculate_probability(
        self,
        agent: BaseAgent,
        task: Task
    ) -> float:
        """Calculate probability of assigning task to agent"""
        key = (agent.id, task.id)

        # Pheromone component
        pheromone = self.pheromones.get(key, 0.1)

        # Heuristic: agent capability match
        heuristic = 1.0 if agent.can_handle(task) else 0.1

        return (pheromone ** self.alpha) * (heuristic ** self.beta)

    def allocate(self) -> dict[str, str]:
        """Generate task allocation"""
        allocation = {}
        available_tasks = (t.  t  .tasks)

         agent  .agents:
              available_tasks:
                

            
            probs = {}
             task  .tasks:
                 task.  available_tasks:
                    probs[task.] = ._calculate_probability(agent, task)

            
            total = (probs.values())
             total > :
                selected = ._weighted_choice(probs, total)
                allocation[agent.] = selected
                available_tasks.remove(selected)

         allocation

     ():
        
        
         key  .pheromones:
            .pheromones[key] *= ( - .evaporation)

        
         agent_id, task_id  allocation.items():
            key = (agent_id, task_id)
            deposit = quality.get(task_id, )
            .pheromones[key] = .pheromones.get(key, ) + deposit

Best Practices

Design Guidelines

1. Keep agents simple: Complex behavior emerges from simple rules
2. Define clear interfaces: Message formats, task structures
3. Plan for failure: Agents will fail; system should continue
4. Monitor collective behavior: Individual agents may be fine but swarm stuck
5. Version coordination protocols: Agents may run different versions

Anti-Patterns to Avoid

• God orchestrator: One agent that does everything
• Chatty agents: Too much inter-agent communication
• Tight coupling: Agents depending on specific other agents
• Missing deadlines: No timeouts on task completion
• State explosion: Agents maintaining too much state

References

• references/swarm-topologies.md - Detailed topology patterns
• references/coordination-protocols.md - Consensus and voting algorithms
• references/emergent-patterns.md - Stigmergy and self-organization