| 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
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:
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"""
priority = -task.priority
await self.task_queue.put((priority, task))
async def run(self):
"""Main orchestration loop"""
while True:
_, task = await self.task_queue.get()
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:
await agent.receive_message(message)
async def send_direct(self, message: AgentMessage):
"""Send message to specific agent"""
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] = {}
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 = {}
self.alpha = 1.0
self.beta = 2.0
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 = self.pheromones.get(key, 0.1)
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
- Keep agents simple: Complex behavior emerges from simple rules
- Define clear interfaces: Message formats, task structures
- Plan for failure: Agents will fail; system should continue
- Monitor collective behavior: Individual agents may be fine but swarm stuck
- 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 patternsreferences/coordination-protocols.md - Consensus and voting algorithmsreferences/emergent-patterns.md - Stigmergy and self-organization
