Building Multi-Agent AI Systems: Architectures, Communication & Coordination

Master the design and implementation of multi-agent AI systems — from agent societies to production deployment.

Introduction

Single agents have limitations. Multi-agent systems (MAS) enable specialization, parallelization, robustness, and emergent capabilities through agent collaboration.

Why Multi-Agent?

Single Agent	Multi-Agent
One model, all tasks	Specialized agents per domain
Sequential processing	Parallel execution
Context window limits	Distributed context
Single point of failure	Redundancy, resilience
Hard to debug	Isolated, testable components

Part 1: Multi-Agent Architectures

Architecture Patterns

# architectures.py
from enum import Enum
from dataclasses import dataclass
from typing import Literal

class ArchitecturePattern(Enum):
    SUPERVISOR = "supervisor"           # Central coordinator
    HIERARCHICAL = "hierarchical"       # Tree of supervisors
    SWARM = "swarm"                     # Peer-to-peer collaboration
    PIPELINE = "pipeline"               # Sequential stages
    MARKET = "market"                   # Auction-based task allocation
    BLACKBOARD = "blackboard"           # Shared knowledge space
    HYBRID = "hybrid"                   # Combination

@dataclass
class AgentSpec:
    name: str
    role: str
    capabilities: list[str]
    tools: list[str]
    model: str
    system_prompt: str

# Pattern 1: Supervisor (Most Common)
SUPERVISOR_SPEC = AgentSpec(
    name="supervisor",
    role="Coordinate specialists, decompose tasks, synthesize results",
    capabilities=["planning", "routing", "synthesis"],
    tools=["delegate", "aggregate"],
    model="gpt-4o",
    system_prompt="""You are a supervisor managing specialist agents.
    Break down user requests, delegate to appropriate specialists,
    and synthesize their outputs into a coherent response."""
)

SPECIALISTS = [
    AgentSpec("researcher", "Information gathering", ["search", "read", "summarize"], 
              ["web_search", "doc_lookup"], "gpt-4o-mini", "..."),
    AgentSpec("coder", "Code generation", ["write", "debug", "test"],
              ["code_exec", "lint"], "gpt-4o", "..."),
    AgentSpec("reviewer", "Quality assurance", ["review", "security", "performance"],
              ["static_analysis"], "gpt-4o", "..."),
    AgentSpec("writer", "Documentation", ["explain", "document", "summarize"],
              [], "gpt-4o-mini", "..."),
]

# Pattern 2: Hierarchical (For complex orgs)
HIERARCHICAL_SPECS = {
    "ceo": AgentSpec("ceo", "Strategic planning", ["strategy", "delegation"], [], "gpt-4o", "..."),
    "cto": AgentSpec("cto", "Technical leadership", ["architecture", "code_review"], [], "gpt-4o", "..."),
    "pm": AgentSpec("pm", "Project management", ["planning", "tracking"], [], "gpt-4o-mini", "..."),
    "dev_lead": AgentSpec("dev_lead", "Team lead", ["code_review", "mentoring"], [], "gpt-4o", "..."),
    "developers": [AgentSpec(f"dev_{i}", "Developer", ["coding", "testing"], [], "gpt-4o-mini", "...") for i in range(3)],
    "qa": AgentSpec("qa", "Quality assurance", ["testing", "validation"], [], "gpt-4o-mini", "..."),
}

# Pattern 3: Swarm (Peer-to-peer)
SWARM_PROMPT = """You are part of a swarm of agents solving: {task}

Other agents: {other_agents}
Your specialization: {my_specialization}

Communicate via shared messages. Propose actions, vote on decisions.
Reach consensus before taking final action.

Current state: {shared_state}
Your proposal:"""

Choosing an Architecture

TASK CHARACTERISTICS → ARCHITECTURE

Clear decomposition, known specialists → SUPERVISOR
Deep hierarchy, org-like structure → HIERARCHICAL
Exploratory, creative, consensus needed → SWARM
Sequential stages, clear handoffs → PIPELINE
Resource allocation, competing bids → MARKET
Complex reasoning, shared knowledge → BLACKBOARD
Real-world complexity → HYBRID

Part 2: Agent Communication Protocols

Message Passing

# communication.py
from dataclasses import dataclass, field
from datetime import datetime
from typing import Any, Optional
from enum import Enum
import uuid

class MessageType(Enum):
    TASK_REQUEST = "task_request"
    TASK_RESPONSE = "task_response"
    TASK_DELEGATE = "task_delegate"
    STATUS_UPDATE = "status_update"
    QUERY = "query"
    RESPONSE = "response"
    BROADCAST = "broadcast"
    CONSENSUS_PROPOSE = "consensus_propose"
    CONSENSUS_VOTE = "consensus_vote"
    CONSENSUS_RESULT = "consensus_result"

@dataclass
class AgentMessage:
    id: str = field(default_factory=lambda: str(uuid.uuid4()))
    type: MessageType = MessageType.TASK_REQUEST
    sender: str = ""
    recipient: str | list[str] = ""  # Empty = broadcast
    content: Any = None
    metadata: dict = field(default_factory=dict)
    timestamp: str = field(default_factory=lambda: datetime.utcnow().isoformat())
    correlation_id: str = ""  # For request-response pairing
    reply_to: str = ""  # Message ID being replied to

class MessageBus:
    """Central message bus for agent communication."""

    def __init__(self):
        self.subscribers: dict[str, list[callable]] = {}
        self.history: list[AgentMessage] = []

    def subscribe(self, agent_name: str, handler: callable):
        if agent_name not in self.subscribers:
            self.subscribers[agent_name] = []
        self.subscribers[agent_name].append(handler)

    def publish(self, message: AgentMessage):
        self.history.append(message)

        recipients = message.recipient if isinstance(message.recipient, list) else [message.recipient]

        if not recipients or recipients == ["*"]:  # Broadcast
            for agent, handlers in self.subscribers.items():
                for handler in handlers:
                    handler(message)
        else:
            for recipient in recipients:
                if recipient in self.subscribers:
                    for handler in self.subscribers[recipient]:
                        handler(message)

    def request_response(self, sender: str, recipient: str, content: Any, timeout: float = 30) -> Any:
        """Synchronous request-response pattern."""
        import asyncio
        correlation_id = str(uuid.uuid4())
        future = asyncio.Future()

        def response_handler(msg: AgentMessage):
            if msg.correlation_id == correlation_id and msg.sender == recipient:
                future.set_result(msg.content)

        self.subscribe(f"__response_{correlation_id}", response_handler)

        msg = AgentMessage(
            type=MessageType.TASK_REQUEST,
            sender=sender,
            recipient=recipient,
            content=content,
            correlation_id=correlation_id
        )
        self.publish(msg)

        try:
            return asyncio.wait_for(future, timeout=timeout)
        except asyncio.TimeoutError:
            raise TimeoutError(f"No response from {recipient}")

Structured Communication (ACL - Agent Communication Language)

# acl.py
from dataclasses import dataclass
from typing import Literal

@dataclass
class ACLMessage:
    """FIPA-ACL inspired message structure."""
    performative: Literal[
        "inform", "request", "query", "propose", "accept", "reject",
        "subscribe", "cancel", "confirm", "disconfirm", "failure"
    ]
    sender: str
    receiver: str | list[str]
    content: str
    language: str = "json"
    ontology: str = "default"
    protocol: str = "request-response"
    conversation_id: str = ""
    reply_with: str = ""
    in_reply_to: str = ""

# Example usage:
# request = ACLMessage(
#     performative="request",
#     sender="supervisor",
#     receiver="researcher",
#     content=json.dumps({"task": "Find papers on transformer architectures", "max_results": 10}),
#     conversation_id="conv-123",
#     reply_with="req-456"
# )

Part 3: Coordination Mechanisms

1. Task Decomposition & Delegation

# delegation.py
from typing import Literal
from pydantic import BaseModel

class TaskDecomposition(BaseModel):
    subtasks: list[dict]  # {"id": "", "description": "", "assigned_to": "", "dependencies": []}
    execution_order: list[str]  # Task IDs in order

DECOMPOSITION_PROMPT = """Decompose this task into subtasks for specialist agents.

Task: {task}
Available agents: {agents}

Consider:
- Dependencies between subtasks
- Which agent is best for each
- Parallel vs sequential execution
- Context passing requirements

Return JSON:
{
  "subtasks": [
    {"id": "1", "description": "...", "assigned_to": "researcher", "dependencies": []},
    {"id": "2", "description": "...", "assigned_to": "coder", "dependencies": ["1"]}
  ],
  "execution_order": ["1", "2"]
}"""

2. Consensus Building

# consensus.py
import asyncio
from dataclasses import dataclass
from typing import Literal

@dataclass
class ConsensusProposal:
    id: str
    proposer: str
    proposal: dict
    votes: dict[str, Literal["yes", "no", "abstain"]] = field(default_factory=dict)
    threshold: float = 0.66  # 2/3 majority
    deadline: float = 30.0  # seconds

class ConsensusManager:
    def __init__(self, message_bus: MessageBus):
        self.bus = message_bus
        self.active_proposals: dict[str, ConsensusProposal] = {}

    async def propose(self, proposer: str, proposal: dict, 
                      participants: list[str], threshold: float = 0.66) -> bool:
        proposal_id = str(uuid.uuid4())
        cp = ConsensusProposal(
            id=proposal_id,
            proposer=proposer,
            proposal=proposal,
            threshold=threshold
        )
        self.active_proposals[proposal_id] = cp

        # Broadcast proposal
        for participant in participants:
            msg = AgentMessage(
                type=MessageType.CONSENSUS_PROPOSE,
                sender=proposer,
                recipient=participant,
                content={"proposal_id": proposal_id, "proposal": proposal},
                correlation_id=proposal_id
            )
            self.bus.publish(msg)

        # Wait for votes
        await asyncio.sleep(cp.deadline)

        # Tally
        yes_votes = sum(1 for v in cp.votes.values() if v == "yes")
        total = len(participants)

        result = yes_votes / total >= threshold

        # Broadcast result
        result_msg = AgentMessage(
            type=MessageType.CONSENSUS_RESULT,
            sender="consensus",
            recipient="*",
            content={"proposal_id": proposal_id, "accepted": result, "votes": cp.votes}
        )
        self.bus.publish(result_msg)

        del self.active_proposals[proposal_id]
        return result

    def vote(self, proposal_id: str, voter: str, vote: Literal["yes", "no", "abstain"]):
        if proposal_id in self.active_proposals:
            self.active_proposals[proposal_id].votes[voter] = vote

3. Shared Blackboard

# blackboard.py
from typing import Any
import asyncio
from collections import defaultdict

class Blackboard:
    """Shared knowledge space for multi-agent collaboration."""

    def __init__(self):
        self.data: dict[str, Any] = {}
        self.subscribers: dict[str, list[callable]] = defaultdict(list)
        self.locks: dict[str, asyncio.Lock] = defaultdict(asyncio.Lock)
        self.version: dict[str, int] = defaultdict(int)

    async def write(self, key: str, value: Any, agent: str):
        async with self.locks[key]:
            self.data[key] = value
            self.version[key] += 1
            # Notify subscribers
            for callback in self.subscribers[key]:
                await callback(key, value, self.version[key], agent)

    async def read(self, key: str) -> tuple[Any, int]:
        return self.data.get(key), self.version.get(key, 0)

    def subscribe(self, key: str, callback: callable):
        self.subscribers[key].append(callback)

    async def atomic_update(self, key: str, updater: callable, agent: str):
        """Atomically update a value."""
        async with self.locks[key]:
            current = self.data.get(key)
            new_value = updater(current)
            self.data[key] = new_value
            self.version[key] += 1
            for callback in self.subscribers[key]:
                await callback(key, new_value, self.version[key], agent)
            return new_value

# Usage in agents:
# blackboard.subscribe("research_findings", on_new_research)
# await blackboard.write("research_findings", findings, "researcher")
# await blackboard.atomic_update("task_status", lambda s: {**s, "step_3": "done"}, "coder")

Part 4: Production Multi-Agent Frameworks

LangGraph Multi-Agent

# langgraph_multiagent.py
from langgraph.graph import StateGraph, START, END
from langgraph.checkpoint.sqlite import SqliteSaver
from typing import TypedDict, Annotated, Literal
from langgraph.graph.message import add_messages
from langchain_core.messages import BaseMessage

class MultiAgentState(TypedDict):
    messages: Annotated[list[BaseMessage], add_messages]
    current_agent: str
    task: str
    results: dict
    next_action: Literal["continue", "delegate", "complete"]

def supervisor_node(state: MultiAgentState):
    # Decide next agent or complete
    pass

def researcher_node(state: MultiAgentState):
    # Research subtask
    pass

def coder_node(state: MultiAgentState):
    # Code generation
    pass

def reviewer_node(state: MultiAgentState):
    # Review code
    pass

workflow = StateGraph(MultiAgentState)
workflow.add_node("supervisor", supervisor_node)
workflow.add_node("researcher", researcher_node)
workflow.add_node("coder", coder_node)
workflow.add_node("reviewer", reviewer_node)

workflow.add_edge(START, "supervisor")
workflow.add_conditional_edges("supervisor", lambda s: s["next_action"], {
    "delegate": "researcher",  # Dynamic based on state
    "continue": "coder",
    "complete": END
})
# Add edges between specialists and back to supervisor

AutoGen (Microsoft)

# autogen_example.py
import autogen

# Configure agents
config_list = [{"model": "gpt-4o", "api_key": "..."}]

assistant = autogen.AssistantAgent(
    name="assistant",
    llm_config={"config_list": config_list},
    system_message="You are a helpful AI assistant. Solve tasks using code when needed."
)

user_proxy = autogen.UserProxyAgent(
    name="user_proxy",
    human_input_mode="NEVER",
    max_consecutive_auto_reply=10,
    code_execution_config={"work_dir": "coding", "use_docker": False},
)

# Multi-agent: GroupChat
coder = autogen.AssistantAgent(
    name="coder",
    llm_config={"config_list": config_list},
    system_message="You are a Python expert. Write clean, tested code."
)

reviewer = autogen.AssistantAgent(
    name="reviewer",
    llm_config={"config_list": config_list},
    system_message="You are a code reviewer. Find bugs and suggest improvements."
)

manager = autogen.GroupChatManager(
    groupchat=autogen.GroupChat(
        agents=[user_proxy, coder, reviewer],
        messages=[],
        max_round=20
    ),
    llm_config={"config_list": config_list}
)

# Start
user_proxy.initiate_chat(manager, message="Build a REST API for a todo app with FastAPI")

CrewAI

# crewai_example.py
from crewai import Agent, Task, Crew, Process
from langchain_openai import ChatOpenAI

llm = ChatOpenAI(model="gpt-4o", temperature=0)

# Define agents
researcher = Agent(
    role="Senior Research Analyst",
    goal="Find cutting-edge information on {topic}",
    backstory="Expert at finding and synthesizing technical information",
    tools=[search_tool],
    llm=llm,
    verbose=True
)

writer = Agent(
    role="Technical Writer",
    goal="Create comprehensive, well-structured documentation",
    backstory="Expert at explaining complex topics clearly",
    llm=llm,
    verbose=True
)

reviewer = Agent(
    role="Quality Reviewer",
    goal="Ensure accuracy, completeness, and clarity",
    backstory="Meticulous reviewer with high standards",
    llm=llm,
    verbose=True
)

# Define tasks
research_task = Task(
    description="Research {topic} thoroughly. Find 10+ sources.",
    agent=researcher,
    expected_output="Research report with sources"
)

write_task = Task(
    description="Write a comprehensive guide based on research",
    agent=writer,
    expected_output="Complete technical guide",
    context=[research_task]
)

review_task = Task(
    description="Review the guide for accuracy and clarity",
    agent=reviewer,
    expected_output="Review report with suggested fixes",
    context=[write_task]
)

# Create crew
crew = Crew(
    agents=[researcher, writer, reviewer],
    tasks=[research_task, write_task, review_task],
    process=Process.sequential,  # or Process.hierarchical
    verbose=2
)

# Execute
result = crew.kickoff(inputs={"topic": "Multi-agent AI systems"})

Custom Framework (Lightweight)

# custom_framework.py
from abc import ABC, abstractmethod
from dataclasses import dataclass, field
from typing import Any
import asyncio

@dataclass
class Agent(ABC):
    name: str
    role: str
    tools: list = field(default_factory=list)
    memory: dict = field(default_factory=dict)

    @abstractmethod
    async def act(self, message: AgentMessage, bus: MessageBus) -> list[AgentMessage]:
        pass

    async def run(self, bus: MessageBus):
        bus.subscribe(self.name, self.handle_message)

    async def handle_message(self, message: AgentMessage):
        responses = await self.act(message, bus)
        for resp in responses:
            bus.publish(resp)

class SupervisorAgent(Agent):
    async def act(self, message: AgentMessage, bus: MessageBus) -> list[AgentMessage]:
        if message.type == MessageType.TASK_REQUEST:
            # Decompose and delegate
            plan = await self.decompose(message.content)

            responses = []
            for subtask in plan["subtasks"]:
                delegate_msg = AgentMessage(
                    type=MessageType.TASK_DELEGATE,
                    sender=self.name,
                    recipient=subtask["assigned_to"],
                    content=subtask,
                    correlation_id=message.id
                )
                responses.append(delegate_msg)

            return responses
        return []

    async def decompose(self, task: str) -> dict:
        # Use LLM to decompose
        pass

class SpecialistAgent(Agent):
    def __init__(self, name: str, specialty: str, **kwargs):
        super().__init__(name, specialty, **kwargs)
        self.specialty = specialty

    async def act(self, message: AgentMessage, bus: MessageBus) -> list[AgentMessage]:
        if message.type == MessageType.TASK_DELEGATE:
            # Execute subtask
            result = await self.execute_task(message.content)

            return [AgentMessage(
                type=MessageType.TASK_RESPONSE,
                sender=self.name,
                recipient=message.sender,
                content=result,
                correlation_id=message.correlation_id,
                reply_to=message.id
            )]
        return []

    async def execute_task(self, task: dict) -> dict:
        # Use LLM + tools
        pass

Part 5: Production Considerations

Observability

# observability.py
from dataclasses import dataclass
from datetime import datetime
import time

@dataclass
class AgentMetrics:
    agent_name: str
    timestamp: str
    event_type: str  # task_start, task_complete, task_failed, message_sent, message_received
    latency_ms: float = 0
    tokens_used: int = 0
    cost_usd: float = 0
    success: bool = True
    error: str = ""

class ObservabilityManager:
    def __init__(self):
        self.metrics: list[AgentMetrics] = []

    def record(self, metrics: AgentMetrics):
        self.metrics.append(metrics)

    def get_agent_stats(self, agent_name: str, window_min: int = 60) -> dict:
        # Filter and compute stats
        pass

    def get_system_health(self) -> dict:
        # Overall system health
        pass

# Distributed tracing
def trace_agent_call(agent_name: str):
    def decorator(func):
        async def wrapper(*args, **kwargs):
            start = time.time()
            trace_id = kwargs.get("trace_id", str(uuid.uuid4()))

            try:
                result = await func(*args, **kwargs)
                obs.record(AgentMetrics(
                    agent_name=agent_name,
                    timestamp=datetime.utcnow().isoformat(),
                    event_type="task_complete",
                    latency_ms=(time.time() - start) * 1000,
                    success=True,
                    trace_id=trace_id
                ))
                return result
            except Exception as e:
                obs.record(AgentMetrics(
                    agent_name=agent_name,
                    timestamp=datetime.utcnow().isoformat(),
                    event_type="task_failed",
                    latency_ms=(time.time() - start) * 1000,
                    success=False,
                    error=str(e),
                    trace_id=trace_id
                ))
                raise
        return wrapper
    return decorator

Error Handling & Resilience

# resilience.py
import asyncio
from functools import wraps

class CircuitBreaker:
    def __init__(self, failure_threshold: int = 5, timeout: float = 60):
        self.failure_threshold = failure_threshold
        self.timeout = timeout
        self.failures = 0
        self.last_failure = 0
        self.state = "closed"  # closed, open, half-open

    async def call(self, func, *args, **kwargs):
        if self.state == "open":
            if time.time() - self.last_failure > self.timeout:
                self.state = "half-open"
            else:
                raise CircuitBreakerOpen()

        try:
            result = await func(*args, **kwargs)
            if self.state == "half-open":
                self.state = "closed"
                self.failures = 0
            return result
        except Exception as e:
            self.failures += 1
            self.last_failure = time.time()
            if self.failures >= self.failure_threshold:
                self.state = "open"
            raise

def with_retry(max_retries: int = 3, backoff: float = 1.0):
    def decorator(func):
        @wraps(func)
        async def wrapper(*args, **kwargs):
            for attempt in range(max_retries):
                try:
                    return await func(*args, **kwargs)
                except Exception as e:
                    if attempt == max_retries - 1:
                        raise
                    await asyncio.sleep(backoff * (2 ** attempt))
            return None
        return wrapper
    return decorator

# Agent with resilience
class ResilientAgent(Agent):
    def __init__(self, *args, **kwargs):
        super().__init__(*args, **kwargs)
        self.circuit_breakers: dict[str, CircuitBreaker] = {}

    @with_retry(max_retries=3)
    async def act(self, message: AgentMessage, bus: MessageBus):
        # Tool calls with circuit breakers
        for tool_name in self.tools:
            if tool_name not in self.circuit_breakers:
                self.circuit_breakers[tool_name] = CircuitBreaker()

        # Execute with circuit breaker protection
        pass

Scaling & Deployment

# k8s/multi-agent.yaml
apiVersion: apps/v1
kind: Deployment
metadata:
  name: agent-supervisor
spec:
  replicas: 2
  selector:
    matchLabels:
      app: agent-supervisor
  template:
    spec:
      containers:
      - name: supervisor
        image: myorg/agent-supervisor:v1
        env:
        - name: MESSAGE_BUS_URL
          value: "redis://redis:6379"
        - name: AGENT_REGISTRY
          value: "consul://consul:8500"
        resources:
          requests:
            memory: "1Gi"
            cpu: "500m"
---
apiVersion: apps/v1
kind: Deployment
metadata:
  name: agent-specialists
spec:
  replicas: 3
  selector:
    matchLabels:
      app: agent-specialists
  template:
    spec:
      containers:
      - name: researcher
        image: myorg/agent-researcher:v1
      - name: coder
        image: myorg/agent-coder:v1
      - name: reviewer
        image: myorg/agent-reviewer:v1
        env:
        - name: MESSAGE_BUS_URL
          value: "redis://redis:6379"
        resources:
          requests:
            memory: "2Gi"
            cpu: "1000m"

Part 6: Evaluation & Testing

Multi-Agent Evaluation

# mas_evaluation.py
from dataclasses import dataclass

@dataclass
class MASMetrics:
    task_completion_rate: float
    coordination_efficiency: float  # Messages per successful task
    agent_utilization: dict[str, float]
    consensus_success_rate: float
    total_cost: float
    total_latency: float

async def evaluate_mas(system, test_cases: list) -> MASMetrics:
    results = []

    for tc in test_cases:
        start = time.time()
        try:
            result = await system.execute(tc.task)
            success = evaluate_result(tc, result)
            results.append({
                "success": success,
                "latency": time.time() - start,
                "messages": result.message_count,
                "cost": result.total_cost,
                "agent_contributions": result.agent_contributions
            })
        except Exception as e:
            results.append({"success": False, "error": str(e)})

    return MASMetrics(
        task_completion_rate=sum(r["success"] for r in results) / len(results),
        coordination_efficiency=sum(r.get("messages", 0) for r in results) / max(1, sum(r["success"] for r in results)),
        agent_utilization=compute_utilization(results),
        consensus_success_rate=compute_consensus_rate(results),
        total_cost=sum(r.get("cost", 0) for r in results),
        total_latency=sum(r.get("latency", 0) for r in results)
    )

Decision Framework

MULTI-AGENT DECISION TREE:

Need multiple specialized capabilities?
├─→ No → Single agent with tools
└─→ Yes
    ├─→ Clear task decomposition?
    │   ├─→ Yes → SUPERVISOR pattern
    │   └─→ No → SWARM or BLACKBOARD
    │
    ├─→ Need hierarchy/management?
    │   └─→ Yes → HIERARCHICAL
    │
    ├─→ Sequential pipeline?
    │   └─→ Yes → PIPELINE
    │
    ├─→ Resource competition?
    │   └─→ Yes → MARKET
    │
    └─→ Complex, evolving requirements → HYBRID

Resources

• LangGraph Multi-Agent
• AutoGen Documentation
• CrewAI Documentation
• FIPA ACL Standard
• Multi-Agent Systems (Wooldridge)
• Agent Communication Languages
• Swarm Intelligence

---

Multi-agent systems are the future of AI applications. Start with Supervisor pattern, add observability from day one, and scale to hybrid architectures as complexity grows. The key is clear interfaces, robust communication, and continuous evaluation.