Multi-Agent Systems & Orchestration

Multi-agent systems use multiple specialized AI agents that work together to solve complex problems. Each agent has a specific role, and an orchestrator coordinates their work.

Why Multi-Agent?

Single agents struggle with:

Tasks requiring diverse expertise (coding + design + testing)
Very complex workflows with many steps
Tasks needing different perspectives (analyst + critic + writer)
Parallel processing of independent subtasks

Multi-agent systems solve these by dividing work among specialists.

Multi-Agent Patterns

Pattern 1: Sequential Pipeline

Agents work in sequence, each processing the output of the previous one:

User Request → Researcher → Analyst → Writer → Editor → Final Output

python

class SequentialPipeline:
    def __init__(self):
        self.agents = [
            Agent("researcher", "Research the topic thoroughly"),
            Agent("analyst", "Analyze the research and find key insights"),
            Agent("writer", "Write a clear, engaging article"),
            Agent("editor", "Review and improve the writing")
        ]

    def run(self, task):
        result = task
        for agent in self.agents:
            result = agent.process(result)
            print(f"[{agent.name}] completed")
        return result

Pattern 2: Supervisor (Hub and Spoke)

A supervisor agent delegates to specialized workers:

                    ┌──────────┐
                    │Supervisor│
                    └────┬─────┘
              ┌──────────┼──────────┐
              ▼          ▼          ▼
         ┌────────┐ ┌────────┐ ┌────────┐
         │Coder   │ │Tester  │ │Reviewer│
         └────────┘ └────────┘ └────────┘

python

class SupervisorAgent:
    def __init__(self):
        self.workers = {
            "coder": CoderAgent(),
            "tester": TesterAgent(),
            "reviewer": ReviewerAgent()
        }

    def run(self, task):
        messages = [
            {
                "role": "system",
                "content": """You are a project supervisor. Break tasks down
                and delegate to your team:
                - coder: Writes implementation code
                - tester: Writes and runs tests
                - reviewer: Reviews code for quality and security

                Coordinate their work to complete the task."""
            },
            {"role": "user", "content": task}
        ]

        # Supervisor decides which worker to use
        for step in range(10):
            decision = self.decide_next_step(messages)

            if decision["action"] == "delegate":
                worker = self.workers[decision["worker"]]
                result = worker.execute(decision["instruction"])
                messages.append({
                    "role": "assistant",
                    "content": f"[{decision['worker']}]: {result}"
                })

            elif decision["action"] == "complete":
                return decision["final_answer"]

Pattern 3: Debate / Adversarial

Agents with opposing perspectives debate to find better answers:

python

def debate(topic, rounds=3):
    advocate = Agent("advocate", "Argue FOR this position with evidence")
    critic = Agent("critic", "Challenge claims and find weaknesses")
    judge = Agent("judge", "Evaluate both sides and reach a conclusion")

    arguments = []

    for round in range(rounds):
        # Advocate presents argument
        pro = advocate.respond(topic, arguments)
        arguments.append(f"Advocate: {pro}")

        # Critic challenges
        con = critic.respond(topic, arguments)
        arguments.append(f"Critic: {con}")

    # Judge makes final determination
    verdict = judge.respond(
        f"Based on this debate about '{topic}', what's the best conclusion?",
        arguments
    )
    return verdict

Pattern 4: Parallel Workers

Independent subtasks processed simultaneously:

python

import asyncio

async def parallel_research(topics):
    agents = [ResearchAgent() for _ in topics]

    # Run all agents simultaneously
    results = await asyncio.gather(
        *[agent.research(topic) for agent, topic in zip(agents, topics)]
    )

    # Synthesize results
    synthesizer = Agent("synthesizer", "Combine research into unified report")
    return synthesizer.process(results)

Building a Multi-Agent System with CrewAI

CrewAI is a popular framework for multi-agent collaboration:

bash

pip install crewai crewai-tools

python

from crewai import Agent, Task, Crew, Process

# Define specialized agents
researcher = Agent(
    role="Senior Researcher",
    goal="Find comprehensive, accurate information on the topic",
    backstory="You are an expert researcher with a keen eye for detail.",
    verbose=True
)

writer = Agent(
    role="Technical Writer",
    goal="Create clear, engaging technical content",
    backstory="You are a skilled writer who makes complex topics accessible.",
    verbose=True
)

editor = Agent(
    role="Editor",
    goal="Ensure content is accurate, well-structured, and polished",
    backstory="You are a meticulous editor with high standards.",
    verbose=True
)

# Define tasks
research_task = Task(
    description="Research the latest developments in quantum computing",
    expected_output="A comprehensive research brief with key findings",
    agent=researcher
)

writing_task = Task(
    description="Write a blog post based on the research brief",
    expected_output="A 1000-word engaging blog post",
    agent=writer
)

editing_task = Task(
    description="Review and polish the blog post",
    expected_output="A final, publication-ready blog post",
    agent=editor
)

# Create crew and run
crew = Crew(
    agents=[researcher, writer, editor],
    tasks=[research_task, writing_task, editing_task],
    process=Process.sequential,
    verbose=True
)

result = crew.kickoff()
print(result)

Agent Communication Protocols

Shared Memory (Blackboard)

All agents read and write to a shared state:

python

class SharedMemory:
    def __init__(self):
        self.state = {}
        self.history = []

    def write(self, agent_name, key, value):
        self.state[key] = value
        self.history.append({
            "agent": agent_name,
            "action": "write",
            "key": key,
            "value": value
        })

    def read(self, key):
        return self.state.get(key)

    def get_context(self):
        return dict(self.state)

Message Passing

Agents communicate through structured messages:

python

class Message:
    def __init__(self, sender, receiver, content, msg_type="info"):
        self.sender = sender
        self.receiver = receiver
        self.content = content
        self.type = msg_type  # "info", "request", "result", "error"

class MessageBus:
    def __init__(self):
        self.messages = []

    def send(self, message):
        self.messages.append(message)

    def get_messages(self, receiver):
        return [m for m in self.messages if m.receiver == receiver]

Orchestration Strategies

Strategy	Best For	Complexity
Sequential	Linear workflows	Low
Supervisor	Diverse subtasks	Medium
Parallel	Independent tasks	Medium
Debate	Decision-making	Medium
Hierarchical	Large complex projects	High

Real-World Multi-Agent Examples

Software Development Team

Product Manager Agent → breaks down requirements
    ↓
Architect Agent → designs the solution
    ↓
Developer Agent → writes the code
    ↓
Tester Agent → writes and runs tests
    ↓
Reviewer Agent → code review
    ↓
DevOps Agent → deployment

Content Creation Pipeline

Research Agent → gathers information
    ↓
Writer Agent → drafts content
    ↓
Fact-Checker Agent → verifies claims
    ↓
SEO Agent → optimizes for search
    ↓
Editor Agent → final polish

Challenges and Solutions

1. Agent Coordination

Problem: Agents may contradict each other or produce incompatible outputs. Solution: Clear role definitions, shared context, and a supervisor to resolve conflicts.

2. Cost Management

Problem: Multiple agents = multiple LLM calls = higher costs. Solution: Use cheaper models for simple agents, expensive models only for critical decisions.

3. Error Propagation

Problem: One agent's error cascades through the pipeline. Solution: Validation between steps, ability to retry failed steps.

4. Infinite Loops

Problem: Agents keep delegating to each other endlessly. Solution: Maximum iteration limits, cycle detection, timeout.

python

class SafeOrchestrator:
    MAX_TOTAL_STEPS = 50
    MAX_AGENT_CALLS = 10

    def __init__(self):
        self.step_count = 0
        self.agent_calls = {}

    def can_proceed(self, agent_name):
        if self.step_count >= self.MAX_TOTAL_STEPS:
            return False
        calls = self.agent_calls.get(agent_name, 0)
        if calls >= self.MAX_AGENT_CALLS:
            return False
        self.step_count += 1
        self.agent_calls[agent_name] = calls + 1
        return True

Key Takeaways

Use multi-agent for complex, multi-skill tasks — don't over-engineer simple ones
Start with sequential pipelines — easiest to debug and understand
Define clear roles — ambiguous roles cause conflicts
Set hard limits — iterations, cost, time
Use existing frameworks (CrewAI, AutoGen) — don't build orchestration from scratch
Monitor and log — visibility into agent interactions is critical

With agents mastered, the final module covers deploying AI applications to production and responsible AI practices.