Multi-Agent Architectures

When a single agent is not enough, multiple agents can collaborate on complex tasks. Multi-agent systems divide work across specialised agents, each with distinct roles, tools, and expertise. This lesson covers the major architectural patterns, communication protocols, and when to choose multi-agent over single-agent designs.

---

Why Multi-Agent?

Challenge with Single Agent	How Multi-Agent Helps
Too many tools (context bloat)	Each agent has a focused tool set
No specialisation	Agents can be experts in specific domains
Single point of failure	Other agents can compensate
Complex tasks overwhelm one model	Divide and conquer
Hard to scale	Add more agents as complexity grows

---

Architecture Patterns

Pattern 1: Supervisor

A single supervisor agent delegates tasks to worker agents and aggregates results.

                 ┌──────────────┐
                 │  SUPERVISOR  │
                 │  (orchestrator)│
                 └──────┬───────┘
                        │
            ┌───────────┼───────────┐
            │           │           │
            ▼           ▼           ▼
     ┌───────────┐ ┌──────────┐ ┌───────────┐
     │ Research  │ │ Analysis │ │  Writing   │
     │ Agent     │ │ Agent    │ │  Agent     │
     └───────────┘ └──────────┘ └───────────┘

from openai import OpenAI
import json

client = OpenAI()

class Agent:
    def __init__(self, name: str, system_prompt: str, model: str = "gpt-4o-mini"):
        self.name = name
        self.system_prompt = system_prompt
        self.model = model

    def run(self, task: str, context: str = "") -> str:
        messages = [
            {"role": "system", "content": self.system_prompt},
        ]
        if context:
            messages.append({"role": "user", "content": f"Context:\n{context}"})
        messages.append({"role": "user", "content": task})

        response = client.chat.completions.create(
            model=self.model,
            messages=messages,
        )
        return response.choices[0].message.content

class SupervisorAgent:
    def __init__(self, workers: dict[str, Agent]):
        self.workers = workers
        self.supervisor = Agent(
            name="supervisor",
            system_prompt=(
                "You are a supervisor agent. Given a task, decide which worker "
                "agents to delegate to and in what order. Return JSON: "
                '{"delegations": [{"agent": "name", "task": "description"}]}'
            ),
            model="gpt-4o",
        )

    def run(self, task: str) -> str:
        # Step 1: Plan delegations
        worker_list = ", ".join(
            f"{name} ({agent.system_prompt[:50]}...)"
            for name, agent in self.workers.items()
        )
        plan_response = self.supervisor.run(
            f"Task: {task}\nAvailable workers: {worker_list}"
        )
        plan = json.loads(plan_response)

        # Step 2: Execute delegations
        results = {}
        context = ""
        for delegation in plan["delegations"]:
            agent_name = delegation["agent"]
            agent_task = delegation["task"]

            if agent_name not in self.workers:
                results[agent_name] = f"Error: Unknown agent {agent_name}"
                continue

            print(f"Delegating to {agent_name}: {agent_task}")
            result = self.workers[agent_name].run(agent_task, context)
            results[agent_name] = result
            context += f"\n{agent_name} result: {result}"

        # Step 3: Synthesise results
        synthesis = self.supervisor.run(
            f"Synthesise the following worker results into a final answer for: {task}",
            context=json.dumps(results, indent=2),
        )
        return synthesis

# Usage
team = SupervisorAgent(workers={
    "researcher": Agent("researcher", "You are a research expert. Find and summarise information."),
    "analyst": Agent("analyst", "You are a data analyst. Analyse data and find patterns."),
    "writer": Agent("writer", "You are a technical writer. Write clear, structured reports."),
})

result = team.run("Analyse the impact of AI on healthcare in 2025.")

---

Pattern 2: Peer-to-Peer

Agents communicate directly with each other without a central supervisor.

     ┌───────────┐       ┌──────────┐
     │ Agent A   │◄─────▶│ Agent B  │
     │ (research)│       │ (analyse)│
     └─────┬─────┘       └────┬─────┘
           │                   │
           └────────┬──────────┘
                    │
                    ▼
              ┌──────────┐
              │ Agent C  │
              │ (write)  │
              └──────────┘

class PeerAgent:
    def __init__(self, name: str, system_prompt: str):
        self.name = name
        self.system_prompt = system_prompt
        self.inbox: list[dict] = []

    def receive_message(self, sender: str, message: str):
        self.inbox.append({"from": sender, "message": message})