agent-swarm-orchestration

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