Multi-Agent Systems Design: Architecture Guide

How to design and build systems of collaborating AI agents.

Multi-agent systems represent the frontier of AI architecture. Instead of one agent doing everything, multiple specialized agents collaborate — each handling what they do best. This guide covers the design patterns, trade-offs, and implementation details.

When to Use Multiple Agents

Single Agent Is Fine When:

• Task is well-defined and narrow
• All context fits in one model's context window
• No need for parallel processing
• Quality requirements are moderate

Multi-Agent Is Better When:

• Task spans multiple domains
• Different tasks need different models (cost vs quality)
• Parallel processing speeds up results
• Specialization improves quality
• System needs to scale horizontally

Architecture Patterns

Pattern 1: Pipeline (Sequential)

Input → Agent A → Agent B → Agent C → Output

class PipelineMAS:
    def __init__(self):
        self.researcher = Agent(model="gpt-4o", tools=["search"])
        self.analyst = Agent(model="gpt-4o", tools=["analyze"])
        self.writer = Agent(model="claude-sonnet", tools=["write"])

    async def process(self, topic):
        research = await self.researcher.process(f"Research: {topic}")
        analysis = await self.analyst.process(f"Analyze: {research}")
        article = await self.writer.process(f"Write about: {analysis}")
        return article

Pattern 2: Fan-Out/Fan-In (Parallel)

       ┌→ Agent A ┐
Input →├→ Agent B ├→ Aggregator → Output
       └→ Agent C ┘

class FanOutFanInMAS:
    async def process(self, data):
        # Fan out — parallel execution
        results = await asyncio.gather(
            self.finance_agent.process(data),
            self.legal_agent.process(data),
            self.technical_agent.process(data),
        )

        # Fan in — aggregate results
        return await self.synthesizer.process(
            finance=results[0],
            legal=results[1],
            technical=results[2],
        )

Pattern 3: Hierarchical (Manager-Worker)

         Orchestrator
        ┌────┼────┐
     Worker1 Worker2 Worker3

class HierarchicalMAS:
    def __init__(self):
        self.orchestrator = OrchestratorAgent(
            model="gpt-4o",
            workers={
                "data": DataWorker(model="gpt-4o-mini"),
                "research": ResearchWorker(model="gpt-4o"),
                "creative": CreativeWorker(model="claude-sonnet"),
            },
        )

    async def process(self, task):
        # Orchestrator decomposes task
        subtasks = await self.orchestrator.decompose(task)

        results = []
        for subtask in subtasks:
            # Assign to best worker
            worker = self.orchestrator.select_worker(subtask)
            result = await worker.handle(subtask)
            results.append(result)

        # Synthesize
        return await self.orchestrator.synthesize(results)

Pattern 4: Debate/Consensus

Agent A ←→ Agent B ←→ Agent C
       Vote/Consensus → Output

class DebateMAS:
    async def process(self, question):
        # Multiple agents generate independent answers
        answers = await asyncio.gather(
            self.agent_a.answer(question),
            self.agent_b.answer(question),
            self.agent_c.answer(question),
        )

        # Each agent critiques others
        critiques = await asyncio.gather(*[
            agent.critique(answers) for agent in [self.agent_a, self.agent_b, self.agent_c]
        ])

        # Revise based on critiques
        revised = await asyncio.gather(*[
            agent.revise(answer, critiques) for agent, answer in zip(
                [self.agent_a, self.agent_b, self.agent_c], answers)
        ])

        # Vote on best answer
        return await self.select_best(revised)

Communication Protocols

Direct Message Passing

class AgentCommunicator:
    async def send(self, from_agent, to_agent, message):
        """Direct agent-to-agent communication."""
        await self.message_queue.put({
            "from": from_agent.id,
            "to": to_agent.id,
            "message": message,
            "timestamp": datetime.utcnow(),
        })

    async def receive(self, agent_id):
        """Receive messages for an agent."""
        return await self.message_queue.get(filter={"to": agent_id})

Shared Blackboard

class Blackboard:
    """Shared state that all agents can read and write."""
    def __init__(self):
        self.state = {}
        self.subscribers = {}

    async def write(self, key, value, agent_id):
        self.state[key] = {
            "value": value,
            "written_by": agent_id,
            "timestamp": datetime.utcnow(),
        }
        # Notify subscribers
        for sub in self.subscribers.get(key, []):
            await sub.notify(key, value)

    async def read(self, key):
        return self.state.get(key, {}).get("value")

A2A Protocol

# Using A2A protocol for agent communication
from a2a import AgentClient

class A2ACommunication:
    def __init__(self):
        self.client = AgentClient(my_identity)

    async def delegate_to_agent(self, agent_id, task):
        """Delegate task to another agent via A2A."""
        response = await self.client.send_message(
            to=agent_id,
            message={
                "type": "task_delegation",
                "task": task,
                "budget": 10.00,
                "deadline": "24h",
            },
        )
        return response

Model Assignment Strategy

Different agents use different models based on task complexity:

class ModelAssigner:
    ASSIGNMENTS = {
        # Simple tasks → cheap model
        "classifier": {"model": "gpt-4o-mini", "reason": "Simple classification"},
        "router": {"model": "gpt-4o-mini", "reason": "Pattern matching"},
        "summarizer": {"model": "gpt-4o-mini", "reason": "Well-defined task"},

        # Medium tasks → mid model
        "researcher": {"model": "claude-haiku", "reason": "Needs reasoning"},
        "analyst": {"model": "gpt-4o", "reason": "Complex analysis"},
        "writer": {"model": "gpt-4o", "reason": "Quality writing"},

        # Hard tasks → best model
        "reasoner": {"model": "claude-sonnet", "reason": "Deep reasoning"},
        "creative": {"model": "claude-sonnet", "reason": "Creative tasks"},
        "orchestrator": {"model": "gpt-4o", "reason": "Strategic decisions"},
    }

    def get_model(self, agent_role):
        return self.ASSIGNMENTS.get(agent_role, {"model": "gpt-4o"})

Handling Failures

class ResilientMAS:
    async def execute_with_failover(self, task):
        """Execute task with automatic failover."""
        try:
            # Try primary agent
            return await self.primary_agent.process(task)
        except AgentError:
            # Fall back to secondary
            try:
                return await self.secondary_agent.process(task)
            except AgentError:
                # Last resort: simple agent
                return await self.simple_agent.process(task)

    async def execute_with_retry(self, task, max_retries=3):
        """Retry failed agent execution."""
        for attempt in range(max_retries):
            try:
                return await self.agent.process(task)
            except TimeoutError:
                if attempt == max_retries - 1:
                    return self.fallback_response(task)
                await asyncio.sleep(2 ** attempt)

Performance Optimization

Parallel Execution

async def parallel_execution(tasks, agents):
    """Execute tasks in parallel across multiple agents."""
    # Distribute tasks across agents
    results = await asyncio.gather(*[
        agent.process(task)
        for agent, task in zip(agents, tasks)
    ])
    return results

Caching Between Agents

class SharedCache:
    """Cache that agents share to avoid redundant work."""
    async def get_or_compute(self, key, compute_fn):
        cached = await self.redis.get(key)
        if cached:
            return cached

        result = await compute_fn()
        await self.redis.setex(key, 3600, result)
        return result

Monitoring Multi-Agent Systems

class MASMonitor:
    async def get_system_status(self):
        return {
            "active_agents": await self.count_active(),
            "messages_per_minute": await self.get_message_rate(),
            "avg_response_time": await self.get_avg_latency(),
            "error_rate": await self.get_error_rate(),
            "cost_today": await self.get_daily_cost(),
            "per_agent": await self.get_per_agent_stats(),
        }

    async def get_per_agent_stats(self):
        stats = {}
        for agent_name in self.agents:
            stats[agent_name] = {
                "requests": await self.count_requests(agent_name),
                "avg_duration": await self.get_avg_duration(agent_name),
                "success_rate": await self.get_success_rate(agent_name),
                "cost": await self.get_cost(agent_name),
            }
        return stats

Conclusion

Multi-agent systems unlock capabilities that single agents can't achieve — parallelism, specialization, and scalability. By choosing the right architecture pattern, communication protocol, and model assignment strategy, you can build systems that are more capable than any individual agent.

Learn More

• AI Agent Orchestration Patterns
• AI Agent Orchestration Tools
• A2A Protocol Tutorial
• Building AI Agents with TypeScript

Explore multi-agent tools on SkillExchange.