Imported from terminalskills/skills (skills/agent-swarm-orchestration/SKILL.md). Install upstream with npx skills add terminalskills/skills --skill agent-swarm-orchestration. Copyright stays with the author (Apache-2.0).

Agent Swarm Orchestration

Overview

Coordinate multiple AI agents working together on complex tasks. Design topologies, implement routing, handle handoffs, share memory, and enforce quality gates.

Instructions

Why multi-agent?

Single-agent limitations: context window fills up, generalist performance degrades on specialist tasks, no parallel execution, single point of failure. Multi-agent benefits: focused expertise per agent, parallel subtasks, quality agents review others' work, failed agents retry without losing all progress.

Topologies

Pipeline (sequential):
Task → Agent A → Agent B → Agent C → Result
Best for: Linear workflows (Spec → Code → Test → Deploy)

Hierarchical (manager + workers):
         Orchestrator
        /     |      \
   Coder  Tester  Reviewer
Best for: Complex tasks decomposing into independent subtasks

Hub-and-spoke (router):
       ┌→ Specialist A
Router → Specialist B
       └→ Specialist C
Best for: Task classification and routing to the right expert

Orchestrator pattern

# orchestrator.py — Central coordinator managing agent pipeline

from dataclasses import dataclass, field
from enum import Enum

class AgentRole(Enum):
    PLANNER = "planner"
    CODER = "coder"
    REVIEWER = "reviewer"
    TESTER = "tester"

@dataclass
class AgentTask:
    id: str
    role: AgentRole
    input_data: dict
    output_data: dict = field(default_factory=dict)
    status: str = "pending"
    retries: int = 0
    max_retries: int = 3

class Orchestrator:
    def __init__(self, agents: dict[AgentRole, 'Agent']):
        self.agents = agents
        self.tasks: list[AgentTask] = []
        self.context: dict = {}  # Shared memory

    async def run_pipeline(self, spec: str) -> dict:
        plan = await self._run_agent(AgentRole.PLANNER, {"spec": spec})
        self.context["plan"] = plan

        for subtask in plan.get("subtasks", []):
            result = await self._run_agent(AgentRole.CODER, {
                "task": subtask, "plan": plan
            })
            review = await self._run_agent(AgentRole.REVIEWER, {
                "code": result, "requirements": subtask
            })
            retries = 0
            while not review.get("approved") and retries < 3:
                result = await self._run_agent(AgentRole.CODER, {
                    "task": subtask, "previous_attempt": result,
                    "feedback": review.get("feedback")
                })
                review = await self._run_agent(AgentRole.REVIEWER, {
                    "code": result, "requirements": subtask
                })
                retries += 1
            self.context[f"subtask_{subtask['id']}"] = result

        tests = await self._run_agent(AgentRole.TESTER, {"code": self.context})
        return {"plan": plan, "results": self.context, "tests": tests}

    async def _run_agent(self, role: AgentRole, input_data: dict) -> dict:
        agent = self.agents[role]
        task = AgentTask(id=f"{role.value}_{len(self.tasks)}", role=role, input_data=input_data)
        self.tasks.append(task)
        try:
            task.status = "running"
            result = await agent.execute(input_data)
            task.output_data = result
            task.status = "completed"
            return result
        except Exception:
            task.status = "failed"
            if task.retries < task.max_retries:
                task.retries += 1
                return await self._run_agent(role, input_data)
            raise

Router pattern

# router.py — Classify and route tasks to specialists

class TaskRouter:
    ROUTING_PROMPT = """Classify this task and select the best agent:
Task: {task}
Available agents: {agents}
Return JSON: {{"agent": "name", "confidence": 0.0-1.0, "reasoning": "why"}}"""

    def __init__(self, agents: dict[str, 'Agent']):
        self.agents = agents

    async def route(self, task: str) -> dict:
        agent_descriptions = "\n".join(
            f"- {name}: {agent.description}" for name, agent in self.agents.items()
        )
        routing = await self._classify(task, agent_descriptions)
        return await self.agents[routing["agent"]].execute({"task": task})

Shared memory

# shared_memory.py — Inter-agent communication layer

class SharedMemory:
    def __init__(self):
        self.facts: list[dict] = []
        self.decisions: list[dict] = []
        self.artifacts: dict = {}

    def add_fact(self, agent: str, fact: str, confidence: float = 1.0):
        self.facts.append({"agent": agent, "fact": fact, "confidence": confidence})

    def add_decision(self, agent: str, decision: str, reasoning: str):
        self.decisions.append({"agent": agent, "decision": decision, "reasoning": reasoning})

    def get_context_for_agent(self, agent_role: str, max_items: int = 20) -> str:
        parts = []
        for f in self.facts[-max_items:]:
            parts.append(f"[{f['agent']}] {f['fact']}")
        for d in self.decisions[-max_items:]:
            parts.append(f"[{d['agent']}] {d['decision']}: {d['reasoning']}")
        return "\n".join(parts)

Quality gates

Enforce quality between pipeline stages:

# quality_gate.py — Validate agent output before handoff

@dataclass
class QualityCheck:
    name: str
    passed: bool
    details: str
    severity: str  # "blocking" or "warning"

class QualityGate:
    async def check(self, stage: str, output: dict) -> list[QualityCheck]:
        checks = []
        if stage == "code":
            checks.append(self._check_syntax(output.get("code", "")))
            checks.append(self._check_tests_present(output))
            checks.append(self._check_no_secrets(output.get("code", "")))
        elif stage == "review":
            checks.append(self._check_review_depth(output.get("review", "")))
        elif stage == "test":
            checks.append(self._check_tests_pass(output.get("test_results", {})))
        return checks

    def gate_passed(self, checks: list[QualityCheck]) -> bool:
        return all(c.passed for c in checks if c.severity == "blocking")

Examples

Build a code review pipeline

Build a multi-agent pipeline for automated code review. Agent 1 (Analyzer) reads the PR diff and identifies potential issues. Agent 2 (Security Reviewer) checks for security vulnerabilities. Agent 3 (Style Checker) verifies coding standards. The Orchestrator collects all findings, deduplicates, prioritizes by severity, and produces a structured review. Include retry logic for when agents produce low-quality reviews.

Create a research swarm

Build a research swarm where 4 agents each search different sources (web, academic papers, news, social media) for information about a topic, then a Synthesizer agent combines their findings into a comprehensive brief. Use shared memory so agents can see what others have found and avoid duplication. Include confidence scores and source citations.

Design a customer support routing system

Build a support ticket routing system with 5 specialist agents: Billing, Technical, Account, Feature Requests, and Escalation. The Router agent classifies incoming tickets and routes to the right specialist. If confidence is below 70%, route to a generalist. Track routing accuracy and retrain the classifier weekly based on resolution data.

Guidelines

• Start with the simplest topology (pipeline) and only add complexity when needed
• Always include quality gates between pipeline stages — never pass unchecked output forward
• Use shared memory to prevent agents from duplicating work or contradicting each other
• Set retry limits (typically 3) to prevent infinite loops when agents fail repeatedly
• Route to a generalist or escalate to human when classifier confidence is below 70%
• Log every agent decision and handoff for debugging and optimization
• Keep individual agent contexts small and focused — specialist agents outperform generalists