AI Agent Orchestration Patterns

single_agent_limits:
  context_constraints:
    - Limited context window
    - Information overload
    - Lost in the middle problem

  capability_constraints:
    - Jack of all trades
    - No deep specialization
    - Generic approaches

  reliability_constraints:
    - Single point of failure
    - No verification
    - Compounding errors

multi_agent_benefits:
  specialization:
    - Deep expertise per agent
    - Optimized prompts
    - Appropriate models

  verification:
    - Cross-checking
    - Multiple perspectives
    - Error detection

  scalability:
    - Parallel execution
    - Divide and conquer
    - Modular updates

class SequentialOrchestrator:
    """Agents execute in sequence, passing results."""

    async def run_pipeline(
        self,
        task: str,
        agents: list[Agent]
    ) -> PipelineResult:
        context = {"original_task": task}

        for agent in agents:
            result = await agent.execute(
                task=self._build_task(agent, context),
                context=context
            )

            context[agent.name] = result
            context["previous_result"] = result

        return PipelineResult(
            final_result=context["previous_result"],
            intermediate_results=context
        )

# Example: Content creation pipeline
pipeline = [
    ResearchAgent(name="researcher"),
    OutlineAgent(name="outliner"),
    WriterAgent(name="writer"),
    EditorAgent(name="editor")
]

class ParallelOrchestrator:
    """Execute independent agents in parallel."""

    async def run_parallel(
        self,
        task: str,
        agents: list[Agent]
    ) -> ParallelResult:
        # Launch all agents
        tasks = [
            agent.execute(task)
            for agent in agents
        ]

        # Wait for all to complete
        results = await asyncio.gather(*tasks, return_exceptions=True)

        # Handle failures
        successful = []
        failed = []
        for agent, result in zip(agents, results):
            if isinstance(result, Exception):
                failed.append((agent.name, result))
            else:
                successful.append((agent.name, result))

        return ParallelResult(
            successful=successful,
            failed=failed
        )

# Example: Multi-perspective analysis
agents = [
    TechnicalAnalyst(),
    BusinessAnalyst(),
    RiskAnalyst()
]

class HierarchicalOrchestrator:
    """Manager agent delegates to worker agents."""

    def __init__(self, manager: Agent, workers: dict[str, Agent]):
        self.manager = manager
        self.workers = workers

    async def run(self, task: str) -> HierarchicalResult:
        # Manager creates plan
        plan = await self.manager.plan(
            task=task,
            available_workers=list(self.workers.keys())
        )

        results = {}
        for step in plan.steps:
            if step.parallel:
                # Run parallel workers
                step_results = await self._run_parallel(step.workers)
            else:
                # Run sequential
                step_results = await self._run_sequential(step.workers)

            results[step.name] = step_results

            # Manager reviews and may replan
            review = await self.manager.review(results)
            if review.needs_replan:
                plan = await self.manager.replan(task, results)

        # Manager synthesizes final result
        final = await self.manager.synthesize(task, results)
        return HierarchicalResult(results=results, final=final)

class DebateOrchestrator:
    """Agents debate to reach better conclusions."""

    async def debate(
        self,
        question: str,
        debaters: list[Agent],
        rounds: int = 3
    ) -> DebateResult:
        positions = {}

        # Initial positions
        for agent in debaters:
            positions[agent.name] = await agent.initial_position(question)

        # Debate rounds
        for round in range(rounds):
            new_positions = {}
            for agent in debaters:
                other_positions = {
                    k: v for k, v in positions.items()
                    if k != agent.name
                }

                new_positions[agent.name] = await agent.respond(
                    question=question,
                    own_position=positions[agent.name],
                    other_positions=other_positions
                )

            positions = new_positions

        # Find consensus
        consensus = await self._find_consensus(question, positions)

        return DebateResult(
            final_positions=positions,
            consensus=consensus
        )

agent_communication:
  challenges:
    - Information loss between agents
    - Context misunderstanding
    - Format inconsistencies

  solutions:
    - Structured message formats
    - Explicit context passing
    - Shared memory/state

class ResilientOrchestrator:
    """Handle agent failures gracefully."""

    async def run_with_fallback(
        self,
        task: str,
        primary: Agent,
        fallback: Agent
    ) -> AgentResult:
        try:
            return await primary.execute(task)
        except AgentError as e:
            logger.warning(f"Primary failed: {e}, trying fallback")
            return await fallback.execute(task)

    async def run_with_retry(
        self,
        agent: Agent,
        task: str,
        max_retries: int = 3
    ) -> AgentResult:
        for attempt in range(max_retries):
            try:
                return await agent.execute(task)
            except RetryableError as e:
                if attempt == max_retries - 1:
                    raise
                await asyncio.sleep(2 ** attempt)