Multi-Agent Systems: Collaboration

Why Multi-Agent Systems?

A single LLM agent can handle straightforward tasks — answer a question, summarize a document, draft an email. But real-world workflows are rarely so simple. Consider an enterprise scenario: a customer support request arrives that needs intent classification, policy lookup, sentiment analysis, and a personalized response — all before escalation rules kick in. Asking one monolithic agent to juggle every sub-task leads to bloated system prompts, confused tool selection, and brittle failure modes.

Multi-agent systems (MAS) solve this by decomposing complex workflows into specialized agents that collaborate. Each agent owns a narrow domain, carries a focused system prompt, and has access only to the tools it needs. The result is a modular architecture where agents can be developed, tested, and scaled independently — much like microservices replaced monoliths in traditional software engineering.

Research from both academia and industry backs this up. Google's STORM system uses multiple agents to produce Wikipedia-quality articles. AutoGen from Microsoft demonstrated that agent-to-agent conversation can solve coding challenges more reliably than a single agent. OpenAI's Swarm framework showed that hand-off patterns between lightweight agents outperform heavyweight single-agent designs on customer service benchmarks.

Defining Agent Roles & Responsibilities

The first design decision in a multi-agent system is role decomposition. Each agent should have a clearly scoped responsibility, a system prompt that reinforces that scope, and a defined set of tools. Overlapping responsibilities create ambiguity; gaps between roles create dropped tasks.

Common role archetypes in production multi-agent systems include: Planner agents that break a high-level goal into sub-tasks, Executor agents that carry out individual steps (code generation, API calls, database queries), Critic/Reviewer agents that validate outputs against quality criteria, and Router agents that dispatch incoming requests to the appropriate specialist.

Communication Patterns

How agents talk to each other defines the system's behavior as much as the agents themselves. There are three fundamental communication patterns, each with distinct trade-offs in latency, complexity, and observability.

In practice, most production systems use a hybrid: structured messages for data hand-offs (tool results, retrieved documents) and conversational exchange for planning and error recovery. The key is to define a message schema — typically a typed dictionary or Pydantic model — so agents can parse each other's outputs deterministically rather than relying on free-text extraction.

# Message schema for inter-agent communication
from pydantic import BaseModel
from typing import Literal, Any
from datetime import datetime

class AgentMessage(BaseModel):
    sender: str                          # agent id
    receiver: str                        # target agent id or "broadcast"
    msg_type: Literal["task", "result", "error", "handoff"]
    payload: dict[str, Any]
    timestamp: datetime
    correlation_id: str                  # traces a full workflow

# Example: Planner hands off a sub-task to Researcher
msg = AgentMessage(
    sender="planner-01",
    receiver="researcher-01",
    msg_type="task",
    payload={"query": "Find latest benchmarks for GPT-4o vs Claude 3.5"},
    timestamp=datetime.now(),
    correlation_id="wf-8a3f"
)

Coordination Strategies

Communication is the mechanism; coordination is the policy. It answers: who runs when, in what order, and what happens when things go wrong? Three dominant coordination strategies have emerged in the LLM agent ecosystem.

Sequential Pipeline

Agents execute in a fixed order — A → B → C → D. Each agent's output becomes the next agent's input. This is the simplest pattern and easiest to debug, but it offers no parallelism and a single failure stalls the entire pipeline. Use it when tasks are inherently serial (e.g., plan → research → draft → review).

Parallel Fan-Out / Fan-In

A coordinator agent distributes independent sub-tasks to multiple worker agents simultaneously, then collects and merges results. This dramatically reduces wall-clock time for embarrassingly parallel work like searching multiple data sources or running the same analysis with different prompts. The challenge is designing a robust aggregation step that handles partial failures and conflicting results.

Dynamic Orchestration

An orchestrator agent decides at runtime which agent to invoke next, based on the current state. This is the most flexible pattern — the orchestrator can loop, branch, retry, or escalate — but also the most complex to reason about. OpenAI's Swarm and LangGraph both support this via state-machine or graph-based orchestration.

Hierarchical vs Flat Architectures

The structural topology of your multi-agent system determines how control flows through it. The two primary architectures — hierarchical and flat — each carry distinct strengths depending on task complexity and required autonomy.

A pragmatic middle ground is the hierarchical with lateral channels pattern: a manager handles top-level orchestration, but specialist agents can communicate directly for tightly-coupled sub-workflows (e.g., a Coder and Tester pair that iterate without going back to the manager each time). This reduces latency for inner loops while preserving top-level oversight.

Implementation with Python

Let's build a minimal multi-agent system from scratch — no framework dependencies. This gives you a clear mental model before graduating to tools like LangGraph, AutoGen, or CrewAI. We'll implement the hierarchical pattern with a Planner, Researcher, and Writer agent.

Step 1 — Base Agent Class

from abc import ABC, abstractmethod
from dataclasses import dataclass, field
from typing import Any

# Every agent has a name, role prompt, and an execute method
@dataclass
class BaseAgent(ABC):
    name: str
    system_prompt: str
    tools: list[str] = field(default_factory=list)

    @abstractmethod
    def execute(self, message: dict[str, Any]) -> dict[str, Any]:
        """Process a message and return a result."""
        ...

    def __repr__(self) -> str:
        return f"Agent({self.name})"

Step 2 — Concrete Agent Implementations

import openai

class LLMAgent(BaseAgent):
    """Agent backed by an LLM with optional tool calling."""

    def execute(self, message: dict[str, Any]) -> dict[str, Any]:
        client = openai.OpenAI()
        response = client.chat.completions.create(
            model="gpt-4o",
            messages=[
                {"role": "system", "content": self.system_prompt},
                {"role": "user", "content": message["query"]}
            ],
            temperature=0.3
        )
        return {
            "agent": self.name,
            "result": response.choices[0].message.content,
            "status": "success"
        }

# Instantiate specialist agents
planner = LLMAgent(
    name="planner",
    system_prompt="You are a planning agent. Break the user goal into 2-4 research sub-tasks. Return JSON list."
)
researcher = LLMAgent(
    name="researcher",
    system_prompt="You are a research agent. Given a sub-task, return factual findings with sources.",
    tools=["web_search", "arxiv_search"]
)
writer = LLMAgent(
    name="writer",
    system_prompt="You are a writing agent. Synthesize research findings into a coherent report."
)

Step 3 — Orchestrator

import json, asyncio
from typing import Any

class Orchestrator:
    """Hierarchical coordinator: plan → research (parallel) → write."""

    def __init__(self, planner: BaseAgent, workers: list[BaseAgent], writer: BaseAgent):
        self.planner = planner
        self.workers = {w.name: w for w in workers}
        self.writer = writer

    async def run(self, goal: str) -> str:
        # Step 1: Plan
        plan = self.planner.execute({"query": goal})
        sub_tasks = json.loads(plan["result"])

        # Step 2: Fan-out research tasks in parallel
        research_coros = [
            asyncio.to_thread(
                self.workers["researcher"].execute,
                {"query": task}
            )
            for task in sub_tasks
        ]
        results = await asyncio.gather(*research_coros)

        # Step 3: Synthesize
        combined = "\n\n".join(r["result"] for r in results)
        report = self.writer.execute({"query": combined})
        return report["result"]

# Usage
orch = Orchestrator(planner, [researcher], writer)
output = asyncio.run(orch.run("Compare multi-agent frameworks for production use"))
print(output)

Step 4 — Adding Error Recovery

import time, logging

logger = logging.getLogger("multi_agent")

def safe_execute(agent: BaseAgent, message: dict, retries: int = 3) -> dict:
    """Execute with exponential back-off and fallback."""
    for attempt in range(retries):
        try:
            result = agent.execute(message)
            if result["status"] == "success":
                return result
        except Exception as e:
            wait = 2 ** attempt
            logger.warning(f"{agent.name} failed (attempt {attempt+1}): {e}. Retrying in {wait}s")
            time.sleep(wait)
    # Fallback: return a structured error instead of crashing
    return {"agent": agent.name, "result": "", "status": "failed"}

Challenges & Pitfalls

Multi-agent systems introduce a class of problems that don't exist in single-agent designs. Being aware of these pitfalls early saves significant debugging time in production.

1. Infinite Loops & Circular Delegation

Without explicit termination conditions, agents can bounce tasks back and forth indefinitely. A Coder agent sends code to a Reviewer, the Reviewer requests changes, the Coder makes changes, the Reviewer finds new issues — and the loop never converges. Always set a max iteration count and a convergence criterion (e.g., "if the Reviewer scores > 8/10, stop").

2. Context Window Exhaustion

As agents pass messages back and forth, conversation histories grow. A chain of five agents, each adding 500 tokens of output, quickly consumes 2,500+ tokens of context before the final agent even starts reasoning. Mitigation strategies include: summarizing intermediate results, passing only diffs between iterations, and using separate conversation threads per agent instead of a single shared history.

3. Cost Amplification

Every inter-agent message that routes through an LLM is a billable API call. A system with 4 agents and 3 iterations per task generates 12 LLM calls per user request. At $0.01 per call, that's $0.12 per request — 12× the cost of a single-agent approach. Budget controls, caching, and selective use of smaller models for routing decisions help manage costs.

4. Debugging & Observability

When a multi-agent system produces a wrong answer, which agent is at fault? Was the plan bad? Did research miss critical information? Did the writer hallucinate despite good inputs? You need per-agent logging with correlation IDs that trace a single user request through every agent interaction. Tools like LangSmith, Langfuse, and Phoenix provide this out of the box.

5. State Management at Scale

Shared state between agents — whether a blackboard, a database, or a message queue — must handle concurrent writes, partial failures, and rollbacks. If the Researcher agent updates the shared context while the Coder agent is reading a stale version, you get inconsistent outputs. Use versioned state objects or event-sourcing patterns to maintain consistency.

# Versioned shared state for multi-agent coordination
from dataclasses import dataclass, field
from threading import Lock
from copy import deepcopy

@dataclass
class SharedState:
    data: dict = field(default_factory=dict)
    version: int = 0
    _lock: Lock = field(default_factory=Lock)

    def read(self) -> tuple[dict, int]:
        with self._lock:
            return deepcopy(self.data), self.version

    def write(self, updates: dict, expected_version: int) -> bool:
        """Optimistic concurrency — reject stale writes."""
        with self._lock:
            if self.version != expected_version:
                return False  # Caller must re-read and retry
            self.data.update(updates)
            self.version += 1
            return True