Multi-Agent Systems

Multi-Agent Patterns

When to Use Multi-Agent

Multi-agent systems add complexity. Only use them when a single agent genuinely isn't enough:

Supervisor Pattern

One supervisor agent receives the task, breaks it into subtasks, delegates to specialized worker agents, collects results, and assembles the final output. This is the most common multi-agent pattern.

# Supervisor pattern: One boss, multiple workers
class SupervisorAgent:
    def __init__(self):
        self.workers = {
            "researcher": ResearchAgent(),
            "coder": CodingAgent(),
            "reviewer": ReviewAgent(),
        }

    def run(self, task):
        # Plan: Decide which workers to use
        plan = self.llm.generate(
            f"Break this task into subtasks and assign each to a worker.\n"
            f"Available workers: {list(self.workers.keys())}\n"
            f"Task: {task}"
        )
        # plan: [("researcher", "find best practices"), ("coder", "implement"), ("reviewer", "review")]

        results = {}
        for worker_name, subtask in plan:
            worker = self.workers[worker_name]
            # Pass relevant context from previous results
            result = worker.run(subtask, context=results)
            results[worker_name] = result

        # Synthesize final output
        return self.llm.generate(
            f"Combine these results into a final response:\n{results}"
        )

Debate Pattern

Two or more agents independently answer the same question, then critique each other's answers in rounds. A judge agent (or the same agents) converges on the best answer. This improves accuracy on complex reasoning tasks.

# Debate pattern: Multiple agents argue to find truth
def debate(question, num_rounds=3):
    # Generate initial positions
    agent_a_answer = agent_a.generate(question)
    agent_b_answer = agent_b.generate(question)

    for round in range(num_rounds):
        # Agent A critiques Agent B
        a_critique = agent_a.generate(
            f"Question: {question}\n"
            f"Your answer: {agent_a_answer}\n"
            f"Opponent's answer: {agent_b_answer}\n"
            f"Critique their answer and defend or update yours."
        )

        # Agent B critiques Agent A
        b_critique = agent_b.generate(
            f"Question: {question}\n"
            f"Your answer: {agent_b_answer}\n"
            f"Opponent's answer: {agent_a_answer}\n"
            f"Critique their answer and defend or update yours."
        )

        agent_a_answer = a_critique
        agent_b_answer = b_critique

    # Judge picks the best answer
    return judge.generate(
        f"Pick the better answer:\nA: {agent_a_answer}\nB: {agent_b_answer}"
    )

Pipeline Pattern

Agents process work sequentially, like an assembly line. Each agent transforms the output and passes it to the next. Great for workflows with clear stages.

# Pipeline pattern: Sequential processing
pipeline = [
    ("planner", "Break this feature request into technical tasks"),
    ("coder", "Implement each task with clean, tested code"),
    ("reviewer", "Review the code for bugs, security issues, and style"),
    ("documenter", "Write documentation for the new feature"),
]

def run_pipeline(initial_input):
    current_output = initial_input
    for agent_name, instruction in pipeline:
        agent = get_agent(agent_name)
        current_output = agent.run(
            f"{instruction}\n\nInput from previous stage:\n{current_output}"
        )
    return current_output

Swarm Pattern

In a swarm, agents dynamically hand off conversations to each other based on the current need. There's no fixed hierarchy — any agent can transfer control to another. This is ideal for customer-facing workflows where the topic shifts.

# Swarm pattern: Dynamic hand-offs between peers
agents = {
    "triage": TriageAgent(),       # Classifies the request
    "billing": BillingAgent(),     # Handles billing questions
    "technical": TechnicalAgent(), # Handles tech support
    "escalation": HumanAgent(),    # Escalates to human
}

def swarm_loop(user_message):
    current_agent = agents["triage"]
    messages = [{"role": "user", "content": user_message}]

    while True:
        response = current_agent.run(messages)

        if response.handoff_to:
            # Agent decided to hand off to a specialist
            current_agent = agents[response.handoff_to]
            messages.append({"role": "system",
                "content": f"Transferred from {response.agent_name}. Context: {response.summary}"})
        elif response.done:
            return response.final_answer
        else:
            messages.append(response.message)

Agent Communication

How agents share information is as important as how they reason. Key approaches:

Challenges & Pitfalls

• Cost explosion — N agents × M steps × token cost adds up fast. Budget carefully.
• Error propagation — one agent's mistake cascades to all downstream agents. Build validation between stages.
• Debugging difficulty — tracing a bug across 5 agents with separate conversations is painful. Invest in logging.
• Coordination overhead — agents may duplicate work, contradict each other, or deadlock. Clear protocols prevent this.
• Latency — sequential multi-agent pipelines multiply response time. Parallelize where possible.

Interview Questions