This article was originally published on AI Study Room. For the full version with working code examples and related articles, visit the original post.

Multi-Agent Systems: Coordination, Communication, Consensus

Introduction

A single LLM agent has limitations: finite context windows, single perspective bias, and vulnerability to cascading errors. Multi-agent systems address these by distributing work across specialized agents that communicate, coordinate, and reach consensus. This article covers the architectural patterns for building effective multi-agent systems.

Agent Roles and Specialization

Each agent should have a narrow, defined role:

@dataclass

class AgentSpec:

    name: str

    role: str

    system_prompt: str

    tools: list[dict]

    model: str

class Agent:

    def __init__(self, spec: AgentSpec, llm_fn):

        self.spec = spec

        self.llm = llm_fn

        self.message_history = [{"role": "system", "content": spec.system_prompt}]

    async def process(self, task: str, context: str = "") -> str:

        messages = self.message_history + [{"role": "user", "content": f"{context}\n\nTask: {task}"}]

        response = self.llm(messages, tools=self.spec.tools)

        self.message_history.append({"role": "assistant", "content": response})

        return response

# Define specialized agents

researcher = Agent(AgentSpec(

    name="Researcher",

    role="Information retrieval and fact-checking",

    system_prompt="You are a research specialist. Find accurate information and verify facts. Cite all sources.",

    tools=[search_tool, web_scrape_tool],

    model="claude-sonnet-4-20260512",

))

writer = Agent(AgentSpec(

    name="Writer",

    role="Content creation and editing",

    system_prompt="You are a technical writer. Produce clear, well-structured content. Adapt tone to the audience.",

    tools=[grammar_check_tool, style_guide_tool],

    model="claude-sonnet-4-20260512",

))

critic = Agent(AgentSpec(

    name="Critic",

    role="Quality assurance and review",

    system_prompt="You are a critical reviewer. Find errors, inconsistencies, and areas for improvement. Be thorough.",

    tools=[],

    model="claude-opus-4-20260512",  # Stronger model for evaluation

))

Communication Protocols

Agents communicate through structured messages:

from enum import Enum

from datetime import datetime

class MessageType(Enum):

    TASK = "task"

    RESULT = "result"

    QUERY = "query"

    RESPONSE = "response"

    FEEDBACK = "feedback"

    COORDINATION = "coordination"

@dataclass

class AgentMessage:

    id: str

    sender: str

    recipients: list[str]

    type: MessageType

    content: str

    metadata: dict = None

    timestamp: str = None

    def __post_init__(self):

        if self.timestamp is None:

            self.timestamp = datetime.now().isoformat()

class MessageBus:

    def __init__(self):

        self.queues: dict[str, list[AgentMessage]] = {}

        self.broadcast_log: list[AgentMessage] = []

    def send(self, message: AgentMessage):

        for recipient in message.recipients:

            if recipient not in self.queues:

                self.queues[recipient] = []

            self.queues[recipient].append(message)

        if not message.recipients:

            self.broadcast_log.append(message)

    def receive(self, agent_name: str) -> list[AgentMessage]:

        return self.queues.pop(agent_name, [])

    def broadcast(self, message: AgentMessage):

        for agent in self.queues:

            self.queues[agent].append(message)

        self.broadcast_log.append(message)

Orchestration Patterns

Sequential Handoff

One agent passes results to the next in a pipeline:

class SequentialOrchestrator:

    def __init__(self, agents: list[Agent], bus: MessageBus):

        self.agents = agents

        self.bus = bus

    async def run(self, initial_task: str) -> str:

        current_output = initial_task

        for i, agent in enumerate(self.agents):

            result = await agent.process(current_output)

            current_output = result

            if i < len(self.agents) - 1:

                self.bus.send(AgentMessage(

                    id=f"handoff_{i}",

                    sender=agent.spec.name,

                    recipients=[self.agents[i+1].spec.name],

                    type=MessageType.TASK,

                    content=result,

                ))

        return current_output

Debate and Consensus

Multiple agents work on the same problem and then converge:

class DebateOrchestrator:

    def __init__(self, agents: list[Agent], rounds: int = 3):

        self.agents = agents

        self.rounds = rounds

    async def debate

---

Read the full article on AI Study Room for complete code examples, comparison tables, and related resources.

Found this useful? Check out more developer guides and tool comparisons on AI Study Room.