Implementing Agent-to-Agent Communication Protocols: Building Structured Message Passing, Shared State, and Coordination Layers for Collaborative Multi-Agent RAG Systems

Introduction

Imagine you're building a RAG system to support a financial research firm. Users ask complex questions like: "Compare Nvidia's R&D spending trajectory against AMD's over the last three years, flag any discrepancies with analyst consensus estimates, and summarize the regulatory risk exposure for both companies." No single agent can handle this well. One agent needs to retrieve financial filings, another needs to pull analyst reports, a third needs to reason over regulatory documents, and a fourth needs to synthesize everything into a coherent comparison. The question isn't whether to use multiple agents — it's how to make them work together without descending into a mess of race conditions, hallucinated hand-offs, and lost context.

This is the core challenge of multi-agent RAG systems: coordination. Individual agents are relatively tractable. You can prompt them, give them tools, and evaluate their outputs in isolation. But the moment you introduce inter-agent communication, you're building a distributed system, and distributed systems have a famously unforgiving set of failure modes. Messages get dropped. State diverges. Agents step on each other's work. One slow retrieval agent creates a bottleneck that cascades across the entire pipeline. Getting this right requires thinking carefully about message schemas, state management, and the protocols that govern how agents hand work to one another.

By the end of this lesson, you'll be able to design and implement production-grade communication protocols for collaborative multi-agent RAG systems. You'll build structured message passing layers with typed schemas, implement shared state stores that handle concurrent access safely, design coordination layers that route tasks and handle failure, and understand the architectural trade-offs that determine when these patterns make sense versus when a simpler single-agent approach is better.

What you'll learn:

• How to design typed message schemas and envelope structures for reliable agent-to-agent communication
• How to implement shared state using in-memory stores, Redis, and database-backed persistence with proper concurrency control
• How to build coordination layers including orchestrator patterns, blackboard architectures, and event-driven pipelines
• How to handle failure, retries, and partial completion in multi-agent workflows
• How to trace, debug, and monitor inter-agent communication in production

---

Prerequisites

You should be comfortable with:

• Python async programming (asyncio, async/await)
• Building individual RAG agents with tool use — if you need a refresher, start with AI Agents: From Concept to Implementation
• Basic vector retrieval concepts from Building Your First RAG Pipeline
• Multi-agent orchestration concepts from Multi-Agent Orchestration: Building RAG Systems Where Specialized Agents Collaborate Across Data Sources
• Working knowledge of Pydantic, Redis or similar key-value stores, and basic concurrency primitives

---

Why Communication Protocols Matter More Than You Think

Before writing any code, let's establish why this problem is hard in a way that catches most teams off guard. When developers first build multi-agent systems, they typically connect agents by passing Python dicts or raw strings between function calls. This works fine in a notebook. It falls apart the moment you have more than two agents, any concurrency, or any requirement to resume a failed workflow.

The problem isn't the data itself — it's the implicit contract. When agent A hands a result to agent B as a raw dict, agent B must know exactly what keys to expect, what types they'll have, and what state the upstream work is in. That implicit contract lives in your head and in comments. It doesn't get validated. When agent A changes its output format or adds a new field, agent B silently misinterprets the result. You won't find out until a user complains about a strange response.

Structured communication protocols solve this by making the contract explicit and machine-verifiable. They also unlock capabilities you can't get from function-to-function calls: async message queues, persistent task tracking, retry logic, observability, and the ability to add new agents to an existing pipeline without rewriting the coordination logic.

There's also a subtler benefit. When agents communicate through a well-defined protocol, you can inspect every message that flows through the system. This makes debugging dramatically easier. Instead of trying to reconstruct what happened from logs and print statements, you have a complete record of every piece of work that was requested, assigned, completed, or failed.

---

Designing Your Message Schema

The foundation of any multi-agent communication protocol is the message schema. Every piece of work that moves between agents — task assignments, results, errors, status updates — should be a typed, versioned message with a consistent envelope.

Here's the mental model: think of each message like an HTTP request, not a Python function call. It has a header (metadata about the message itself) and a body (the actual content). The header tells the system where the message came from, where it's going, what kind of message it is, and when it was created. The body carries the domain-specific payload.

Let's build this using Pydantic:

from enum import Enum
from typing import Any, Dict, List, Optional, Union
from uuid import UUID, uuid4
from datetime import datetime, timezone
from pydantic import BaseModel, Field

class MessageType(str, Enum):
    TASK_REQUEST = "task_request"
    TASK_RESULT = "task_result"
    TASK_ERROR = "task_error"
    STATUS_UPDATE = "status_update"
    RETRIEVAL_REQUEST = "retrieval_request"
    RETRIEVAL_RESULT = "retrieval_result"
    SYNTHESIS_REQUEST = "synthesis_request"
    SYNTHESIS_RESULT = "synthesis_result"

class MessagePriority(str, Enum):
    LOW = "low"
    NORMAL = "normal"
    HIGH = "high"
    CRITICAL = "critical"

class MessageStatus(str, Enum):
    PENDING = "pending"
    IN_PROGRESS = "in_progress"
    COMPLETED = "completed"
    FAILED = "failed"
    CANCELLED = "cancelled"

class MessageEnvelope(BaseModel):
    """The outer wrapper for all inter-agent messages."""
    message_id: UUID = Field(default_factory=uuid4)
    correlation_id: UUID  # Links related messages in a workflow
    session_id: UUID      # Top-level user session
    message_type: MessageType
    sender_id: str        # e.g., "retrieval_agent_1"
    recipient_id: str     # e.g., "synthesis_agent" or "orchestrator"
    priority: MessagePriority = MessagePriority.NORMAL
    created_at: datetime = Field(default_factory=lambda: datetime.now(timezone.utc))
    schema_version: str = "1.0"
    payload: Dict[str, Any]
    metadata: Dict[str, Any] = Field(default_factory=dict)

    class Config:
        json_encoders = {UUID: str, datetime: lambda v: v.isoformat()}

The correlation_id field deserves special attention. When an orchestrator decomposes a user query into five sub-tasks, all five task messages and their eventual results should share the same correlation_id. This is what lets you reconstruct the complete trace of a workflow — which requests spawned which results, in what order, with what latencies. Without it, your logs are a soup of unrelated events.

Now let's define the domain-specific payloads that go inside the envelope:

class RetrievalRequest(BaseModel):
    query: str
    data_source: str           # e.g., "sec_filings", "analyst_reports"
    top_k: int = 10
    filters: Dict[str, Any] = Field(default_factory=dict)
    strategy: str = "semantic"  # "semantic", "hybrid", "keyword"
    
class RetrievedChunk(BaseModel):
    chunk_id: str
    content: str
    source_document: str
    score: float
    metadata: Dict[str, Any]

class RetrievalResult(BaseModel):
    query: str
    data_source: str
    chunks: List[RetrievedChunk]
    total_found: int
    latency_ms: float
    retrieval_strategy_used: str

class SynthesisRequest(BaseModel):
    user_query: str
    retrieved_contexts: List[RetrievalResult]
    synthesis_instructions: str
    output_format: str = "markdown"  # "markdown", "json", "structured"

class SynthesisResult(BaseModel):
    user_query: str
    response: str
    sources_cited: List[str]
    confidence_score: float
    token_count: int

class TaskError(BaseModel):
    agent_id: str
    error_type: str
    error_message: str
    recoverable: bool
    suggested_retry_delay_seconds: Optional[int] = None
    context: Dict[str, Any] = Field(default_factory=dict)

A complete message for a retrieval request looks like this:

def create_retrieval_request(
    sender: str,
    recipient: str,
    query: str,
    data_source: str,
    correlation_id: UUID,
    session_id: UUID,
    **kwargs
) -> MessageEnvelope:
    payload = RetrievalRequest(
        query=query,
        data_source=data_source,
        **kwargs
    )
    return MessageEnvelope(
        correlation_id=correlation_id,
        session_id=session_id,
        message_type=MessageType.RETRIEVAL_REQUEST,
        sender_id=sender,
        recipient_id=recipient,
        payload=payload.dict()
    )

# Usage
import asyncio
from uuid import uuid4

session = uuid4()
correlation = uuid4()

msg = create_retrieval_request(
    sender="orchestrator",
    recipient="sec_filings_agent",
    query="Nvidia R&D expenses 2021-2024",
    data_source="sec_filings",
    correlation_id=correlation,
    session_id=session,
    top_k=15,
    filters={"ticker": "NVDA", "doc_type": "10-K"}
)

print(msg.json(indent=2))

---

Building a Shared State Store

Structured messages solve half the problem. The other half is shared state — the working memory of the collaborative system. When a retrieval agent completes its work, where does the result live? When the synthesis agent needs to know whether all retrieval tasks are done, how does it check? When a workflow fails partway through and needs to resume, what state does it read?

For development and single-machine deployments, an in-memory store with async locks is sufficient. For production multi-machine deployments, you need a persistent store like Redis. Let's build both, with a shared interface.

In-Memory Shared State (Development)

import asyncio
from typing import Optional
from collections import defaultdict

class WorkflowState(BaseModel):
    workflow_id: UUID
    session_id: UUID
    user_query: str
    status: MessageStatus = MessageStatus.PENDING
    created_at: datetime = Field(default_factory=lambda: datetime.now(timezone.utc))
    updated_at: datetime = Field(default_factory=lambda: datetime.now(timezone.utc))
    task_count: int = 0
    completed_tasks: int = 0
    failed_tasks: int = 0
    results: Dict[str, Any] = Field(default_factory=dict)
    errors: List[Dict[str, Any]] = Field(default_factory=list)
    agent_assignments: Dict[str, str] = Field(default_factory=dict)
    
    class Config:
        json_encoders = {UUID: str, datetime: lambda v: v.isoformat()}

class InMemoryStateStore:
    """Thread-safe shared state store for single-process deployments."""
    
    def __init__(self):
        self._workflows: Dict[str, WorkflowState] = {}
        self._messages: Dict[str, MessageEnvelope] = {}
        self._agent_queues: Dict[str, asyncio.Queue] = defaultdict(asyncio.Queue)
        self._lock = asyncio.Lock()
    
    async def create_workflow(self, workflow: WorkflowState) -> None:
        async with self._lock:
            key = str(workflow.workflow_id)
            if key in self._workflows:
                raise ValueError(f"Workflow {key} already exists")
            self._workflows[key] = workflow
    
    async def get_workflow(self, workflow_id: UUID) -> Optional[WorkflowState]:
        async with self._lock:
            return self._workflows.get(str(workflow_id))
    
    async def update_workflow(
        self, 
        workflow_id: UUID, 
        updates: Dict[str, Any]
    ) -> WorkflowState:
        async with self._lock:
            key = str(workflow_id)
            if key not in self._workflows:
                raise KeyError(f"Workflow {key} not found")
            
            workflow = self._workflows[key]
            workflow_dict = workflow.dict()
            workflow_dict.update(updates)
            workflow_dict["updated_at"] = datetime.now(timezone.utc)
            
            updated = WorkflowState(**workflow_dict)
            self._workflows[key] = updated
            return updated
    
    async def store_message(self, message: MessageEnvelope) -> None:
        async with self._lock:
            self._messages[str(message.message_id)] = message
    
    async def enqueue_for_agent(
        self, 
        agent_id: str, 
        message: MessageEnvelope
    ) -> None:
        # No lock needed — asyncio.Queue is already thread-safe
        await self._agent_queues[agent_id].put(message)
    
    async def dequeue_for_agent(
        self, 
        agent_id: str,
        timeout: float = 30.0
    ) -> Optional[MessageEnvelope]:
        try:
            return await asyncio.wait_for(
                self._agent_queues[agent_id].get(),
                timeout=timeout
            )
        except asyncio.TimeoutError:
            return None
    
    async def get_workflow_messages(
        self, 
        correlation_id: UUID
    ) -> List[MessageEnvelope]:
        async with self._lock:
            cid = str(correlation_id)
            return [
                msg for msg in self._messages.values()
                if str(msg.correlation_id) == cid
            ]

Redis-Backed State Store (Production)

For production, you need persistence across restarts, atomic operations for concurrent updates, and the ability to run multiple orchestrator processes:

import json
import redis.asyncio as aioredis
from contextlib import asynccontextmanager

class RedisStateStore:
    """Production-grade shared state store backed by Redis."""
    
    WORKFLOW_PREFIX = "workflow:"
    MESSAGE_PREFIX = "message:"
    CORRELATION_PREFIX = "correlation:"
    AGENT_QUEUE_PREFIX = "queue:"
    WORKFLOW_TTL_SECONDS = 86400 * 7  # 7 days
    
    def __init__(self, redis_url: str = "redis://localhost:6379"):
        self._redis_url = redis_url
        self._client: Optional[aioredis.Redis] = None
    
    async def connect(self) -> None:
        self._client = await aioredis.from_url(
            self._redis_url,
            encoding="utf-8",
            decode_responses=True
        )
    
    async def disconnect(self) -> None:
        if self._client:
            await self._client.aclose()
    
    async def create_workflow(self, workflow: WorkflowState) -> None:
        key = f"{self.WORKFLOW_PREFIX}{workflow.workflow_id}"
        serialized = workflow.json()
        
        # Use SET NX (set if not exists) for idempotent creation
        result = await self._client.set(
            key, 
            serialized, 
            nx=True,
            ex=self.WORKFLOW_TTL_SECONDS
        )
        if not result:
            raise ValueError(f"Workflow {workflow.workflow_id} already exists")
    
    async def get_workflow(self, workflow_id: UUID) -> Optional[WorkflowState]:
        key = f"{self.WORKFLOW_PREFIX}{workflow_id}"
        data = await self._client.get(key)
        if data is None:
            return None
        return WorkflowState.parse_raw(data)
    
    async def update_workflow(
        self,
        workflow_id: UUID,
        updates: Dict[str, Any]
    ) -> WorkflowState:
        """Atomic update using a Lua script to prevent race conditions."""
        key = f"{self.WORKFLOW_PREFIX}{workflow_id}"
        
        # Lua script ensures read-modify-write is atomic
        lua_script = """
        local current = redis.call('GET', KEYS[1])
        if current == false then
            return redis.error_reply('Workflow not found')
        end
        local data = cjson.decode(current)
        local updates = cjson.decode(ARGV[1])
        for k, v in pairs(updates) do
            data[k] = v
        end
        data['updated_at'] = ARGV[2]
        local serialized = cjson.encode(data)
        redis.call('SET', KEYS[1], serialized, 'KEEPTTL')
        return serialized
        """
        
        updates["updated_at"] = datetime.now(timezone.utc).isoformat()
        result = await self._client.eval(
            lua_script,
            1,
            key,
            json.dumps(updates),
            updates["updated_at"]
        )
        return WorkflowState.parse_raw(result)
    
    async def store_message(self, message: MessageEnvelope) -> None:
        pipe = self._client.pipeline()
        
        # Store the message itself
        msg_key = f"{self.MESSAGE_PREFIX}{message.message_id}"
        pipe.set(msg_key, message.json(), ex=self.WORKFLOW_TTL_SECONDS)
        
        # Index by correlation_id for efficient lookup
        corr_key = f"{self.CORRELATION_PREFIX}{message.correlation_id}"
        pipe.sadd(corr_key, str(message.message_id))
        pipe.expire(corr_key, self.WORKFLOW_TTL_SECONDS)
        
        await pipe.execute()
    
    async def enqueue_for_agent(
        self, 
        agent_id: str, 
        message: MessageEnvelope
    ) -> None:
        queue_key = f"{self.AGENT_QUEUE_PREFIX}{agent_id}"
        await self._client.rpush(queue_key, message.json())
    
    async def dequeue_for_agent(
        self,
        agent_id: str,
        timeout: float = 30.0
    ) -> Optional[MessageEnvelope]:
        queue_key = f"{self.AGENT_QUEUE_PREFIX}{agent_id}"
        # BLPOP blocks until an item is available or timeout
        result = await self._client.blpop(queue_key, timeout=int(timeout))
        if result is None:
            return None
        _, message_json = result
        return MessageEnvelope.parse_raw(message_json)

---

Designing the Coordination Layer

With structured messages and shared state in place, you need a coordination layer that decides how work flows between agents. There are three primary patterns, each with distinct trade-offs.

Pattern 1: Orchestrator-Worker

The orchestrator receives the user query, decomposes it into sub-tasks, assigns each task to a specialized agent, collects results, and coordinates the final synthesis. This is the most common pattern and the easiest to reason about.

from abc import ABC, abstractmethod
import logging

logger = logging.getLogger(__name__)

class BaseAgent(ABC):
    """Base class for all agents in the system."""
    
    def __init__(self, agent_id: str, state_store: Union[InMemoryStateStore, RedisStateStore]):
        self.agent_id = agent_id
        self.state_store = state_store
        self._running = False
    
    async def start(self) -> None:
        self._running = True
        logger.info(f"Agent {self.agent_id} started")
        await self._run_loop()
    
    async def stop(self) -> None:
        self._running = False
    
    async def _run_loop(self) -> None:
        while self._running:
            message = await self.state_store.dequeue_for_agent(
                self.agent_id, 
                timeout=5.0
            )
            if message is None:
                continue
            
            try:
                await self.state_store.store_message(message)
                result = await self.handle_message(message)
                
                if result is not None:
                    await self.state_store.store_message(result)
                    await self.state_store.enqueue_for_agent(
                        result.recipient_id,
                        result
                    )
            except Exception as e:
                logger.exception(f"Agent {self.agent_id} failed to handle message {message.message_id}")
                error_msg = self._create_error_message(message, e)
                await self.state_store.enqueue_for_agent(
                    message.sender_id,
                    error_msg
                )
    
    @abstractmethod
    async def handle_message(self, message: MessageEnvelope) -> Optional[MessageEnvelope]:
        ...
    
    def _create_error_message(
        self, 
        original: MessageEnvelope, 
        error: Exception
    ) -> MessageEnvelope:
        error_payload = TaskError(
            agent_id=self.agent_id,
            error_type=type(error).__name__,
            error_message=str(error),
            recoverable=isinstance(error, (TimeoutError, ConnectionError)),
            context={"original_message_id": str(original.message_id)}
        )
        return MessageEnvelope(
            correlation_id=original.correlation_id,
            session_id=original.session_id,
            message_type=MessageType.TASK_ERROR,
            sender_id=self.agent_id,
            recipient_id=original.sender_id,
            payload=error_payload.dict()
        )

Now let's implement the orchestrator. The key insight is that the orchestrator doesn't just fire off tasks and wait — it manages a completion map that tracks which tasks are done and triggers synthesis only when all required results are available:

class OrchestratorAgent(BaseAgent):
    """Coordinates task decomposition, assignment, and result aggregation."""
    
    # Maps data source names to the agent IDs that handle them
    AGENT_ROUTING = {
        "sec_filings": "sec_agent",
        "analyst_reports": "analyst_agent",
        "regulatory_docs": "regulatory_agent",
    }
    
    async def handle_message(self, message: MessageEnvelope) -> Optional[MessageEnvelope]:
        if message.message_type == MessageType.TASK_REQUEST:
            return await self._handle_task_request(message)
        elif message.message_type == MessageType.RETRIEVAL_RESULT:
            return await self._handle_retrieval_result(message)
        elif message.message_type == MessageType.TASK_ERROR:
            return await self._handle_error(message)
        return None
    
    async def _handle_task_request(self, message: MessageEnvelope) -> None:
        user_query = message.payload["query"]
        data_sources = message.payload.get("data_sources", list(self.AGENT_ROUTING.keys()))
        
        # Create a workflow to track this job
        workflow = WorkflowState(
            workflow_id=message.correlation_id,
            session_id=message.session_id,
            user_query=user_query,
            status=MessageStatus.IN_PROGRESS,
            task_count=len(data_sources),
            agent_assignments={
                src: self.AGENT_ROUTING[src] for src in data_sources
            }
        )
        await self.state_store.create_workflow(workflow)
        
        # Fan out retrieval tasks to specialized agents
        for data_source in data_sources:
            agent_id = self.AGENT_ROUTING[data_source]
            retrieval_msg = create_retrieval_request(
                sender=self.agent_id,
                recipient=agent_id,
                query=user_query,
                data_source=data_source,
                correlation_id=message.correlation_id,
                session_id=message.session_id
            )
            await self.state_store.enqueue_for_agent(agent_id, retrieval_msg)
            logger.info(f"Dispatched retrieval to {agent_id} for source '{data_source}'")
    
    async def _handle_retrieval_result(self, message: MessageEnvelope) -> Optional[MessageEnvelope]:
        correlation_id = message.correlation_id
        result_payload = RetrievalResult(**message.payload)
        
        # Store this result in the workflow state
        workflow = await self.state_store.update_workflow(
            correlation_id,
            {
                "completed_tasks": None,  # We'll compute this atomically
                f"results.{result_payload.data_source}": result_payload.dict()
            }
        )
        
        # Check if all tasks are complete
        # Re-fetch to get the authoritative count after atomic update
        workflow = await self._increment_completed(correlation_id)
        
        if workflow.completed_tasks + workflow.failed_tasks >= workflow.task_count:
            return await self._trigger_synthesis(workflow)
        
        logger.info(
            f"Workflow {correlation_id}: {workflow.completed_tasks}/{workflow.task_count} tasks done"
        )
        return None
    
    async def _increment_completed(self, workflow_id: UUID) -> WorkflowState:
        """Atomically increment the completed task counter."""
        workflow = await self.state_store.get_workflow(workflow_id)
        return await self.state_store.update_workflow(
            workflow_id,
            {"completed_tasks": workflow.completed_tasks + 1}
        )
    
    async def _trigger_synthesis(self, workflow: WorkflowState) -> MessageEnvelope:
        retrieval_results = [
            RetrievalResult(**v) 
            for v in workflow.results.values()
        ]
        
        synthesis_payload = SynthesisRequest(
            user_query=workflow.user_query,
            retrieved_contexts=retrieval_results,
            synthesis_instructions=(
                "Synthesize the retrieved information into a comprehensive, "
                "well-structured response. Cite sources explicitly."
            )
        )
        
        return MessageEnvelope(
            correlation_id=workflow.workflow_id,
            session_id=workflow.session_id,
            message_type=MessageType.SYNTHESIS_REQUEST,
            sender_id=self.agent_id,
            recipient_id="synthesis_agent",
            payload=synthesis_payload.dict()
        )
    
    async def _handle_error(self, message: MessageEnvelope) -> None:
        error = TaskError(**message.payload)
        workflow = await self.state_store.get_workflow(message.correlation_id)
        
        if workflow is None:
            return
        
        await self.state_store.update_workflow(
            message.correlation_id,
            {
                "failed_tasks": workflow.failed_tasks + 1,
                "errors": workflow.errors + [error.dict()]
            }
        )
        
        if error.recoverable:
            logger.warning(f"Recoverable error from {error.agent_id}, will retry")
            # Implement retry logic here
        else:
            logger.error(f"Unrecoverable error in workflow {message.correlation_id}")

Pattern 2: Blackboard Architecture

The orchestrator-worker pattern works well for predictable workflows, but it has a limitation: the orchestrator must know in advance which agents exist and what they can do. For more dynamic systems — where agents self-register their capabilities and claim tasks that match their skills — a blackboard architecture is more appropriate.

In a blackboard system, shared state (the "blackboard") is the central artifact. Agents watch the blackboard for tasks they can handle, claim them, execute them, and post results back. No central orchestrator routes work; agents pull rather than being pushed to.

class BlackboardEntry(BaseModel):
    entry_id: UUID = Field(default_factory=uuid4)
    workflow_id: UUID
    entry_type: str           # "task", "result", "hypothesis"
    capability_required: str  # What kind of agent can handle this
    status: MessageStatus = MessageStatus.PENDING
    claimed_by: Optional[str] = None
    claimed_at: Optional[datetime] = None
    content: Dict[str, Any]
    created_at: datetime = Field(default_factory=lambda: datetime.now(timezone.utc))

class Blackboard:
    """Shared workspace where agents post tasks and results."""
    
    def __init__(self, state_store: RedisStateStore):
        self.state_store = state_store
        self._prefix = "blackboard:"
    
    async def post(self, entry: BlackboardEntry) -> None:
        key = f"{self._prefix}{entry.entry_id}"
        await self.state_store._client.set(
            key,
            entry.json(),
            ex=RedisStateStore.WORKFLOW_TTL_SECONDS
        )
        # Publish notification for waiting agents
        await self.state_store._client.publish(
            f"blackboard:new:{entry.capability_required}",
            str(entry.entry_id)
        )
    
    async def claim_task(
        self, 
        agent_id: str, 
        capability: str
    ) -> Optional[BlackboardEntry]:
        """Atomically claim the oldest unclaimed task for this capability."""
        
        # Scan for matching entries (in production, use a sorted set for efficiency)
        pattern = f"{self._prefix}*"
        cursor = 0
        candidates = []
        
        while True:
            cursor, keys = await self.state_store._client.scan(
                cursor, match=pattern, count=100
            )
            for key in keys:
                data = await self.state_store._client.get(key)
                if data:
                    entry = BlackboardEntry.parse_raw(data)
                    if (entry.capability_required == capability 
                            and entry.status == MessageStatus.PENDING):
                        candidates.append(entry)
            if cursor == 0:
                break
        
        if not candidates:
            return None
        
        # Sort by creation time (FIFO within priority)
        candidates.sort(key=lambda e: e.created_at)
        target = candidates[0]
        
        # Atomic claim using Lua
        claim_script = """
        local key = KEYS[1]
        local current = redis.call('GET', key)
        if current == false then return nil end
        local entry = cjson.decode(current)
        if entry['status'] ~= 'pending' then return nil end
        entry['status'] = 'in_progress'
        entry['claimed_by'] = ARGV[1]
        entry['claimed_at'] = ARGV[2]
        redis.call('SET', key, cjson.encode(entry), 'KEEPTTL')
        return cjson.encode(entry)
        """
        
        result = await self.state_store._client.eval(
            claim_script,
            1,
            f"{self._prefix}{target.entry_id}",
            agent_id,
            datetime.now(timezone.utc).isoformat()
        )
        
        if result is None:
            return None  # Someone else claimed it first
        
        return BlackboardEntry.parse_raw(result)

Pattern 3: Event-Driven Pipeline

For high-throughput systems where you need maximum decoupling, consider an event-driven approach using a proper message broker like Kafka or RabbitMQ. Each agent publishes domain events and subscribes to the events it cares about. This gives you durable message delivery, replay, and natural backpressure handling.

The implementation varies by broker, but the key is that your MessageEnvelope schema should map cleanly to broker messages. For Kafka, the correlation_id makes an excellent partition key, ensuring all messages for a given workflow land on the same partition and are processed in order.

---

Handling Failure, Retries, and Partial Completion

Production multi-agent systems fail constantly. Networks time out. LLM APIs return 429s. Retrieval agents hit vector DB connection limits. Your protocol needs to handle these gracefully.

The critical design question is: when part of a workflow fails, what happens to the rest?

class RetryPolicy(BaseModel):
    max_attempts: int = 3
    base_delay_seconds: float = 1.0
    max_delay_seconds: float = 60.0
    exponential_base: float = 2.0
    jitter: bool = True

class RetryManager:
    def __init__(self, state_store: Union[InMemoryStateStore, RedisStateStore]):
        self.state_store = state_store
    
    async def schedule_retry(
        self,
        original_message: MessageEnvelope,
        attempt_number: int,
        policy: RetryPolicy
    ) -> Optional[MessageEnvelope]:
        if attempt_number >= policy.max_attempts:
            logger.error(
                f"Message {original_message.message_id} exhausted retries "
                f"after {attempt_number} attempts"
            )
            return None
        
        delay = min(
            policy.base_delay_seconds * (policy.exponential_base ** attempt_number),
            policy.max_delay_seconds
        )
        
        if policy.jitter:
            import random
            delay *= (0.5 + random.random() * 0.5)
        
        logger.info(f"Scheduling retry {attempt_number + 1} in {delay:.1f}s")
        await asyncio.sleep(delay)
        
        # Re-enqueue with incremented attempt counter
        retry_message = original_message.copy(deep=True)
        retry_message.metadata["attempt_number"] = attempt_number + 1
        retry_message.metadata["is_retry"] = True
        
        await self.state_store.enqueue_for_agent(
            original_message.recipient_id,
            retry_message
        )
        return retry_message

For partial completion — where some retrieval agents succeed and others fail — you have two options. The strict approach fails the whole workflow and retries from scratch. The lenient approach synthesizes from whatever results are available, clearly signaling to the user that some sources couldn't be retrieved. Which is correct depends on your domain. For a financial research system, partial results are dangerous and should be clearly flagged. For a general knowledge assistant, partial results are usually fine.

async def _trigger_synthesis_with_partial_results(
    self, 
    workflow: WorkflowState
) -> MessageEnvelope:
    """Build a synthesis request even when some retrievals failed."""
    
    available_results = [
        RetrievalResult(**v) for v in workflow.results.values()
    ]
    failed_sources = [e["context"].get("data_source") for e in workflow.errors]
    
    instructions = "Synthesize the retrieved information comprehensively."
    if failed_sources:
        instructions += (
            f" NOTE: Retrieval from {failed_sources} failed. "
            "Clearly indicate in your response that these sources could not be consulted "
            "and that your answer may be incomplete."
        )
    
    synthesis_payload = SynthesisRequest(
        user_query=workflow.user_query,
        retrieved_contexts=available_results,
        synthesis_instructions=instructions
    )
    
    return MessageEnvelope(
        correlation_id=workflow.workflow_id,
        session_id=workflow.session_id,
        message_type=MessageType.SYNTHESIS_REQUEST,
        sender_id=self.agent_id,
        recipient_id="synthesis_agent",
        payload=synthesis_payload.dict()
    )

This connects to broader concerns about Structured Output Contracts for AI Agents: Enforcing JSON Schemas, Retry Logic, and Validation Pipelines in Production — your synthesis agent should validate that its output conforms to the agreed schema before returning it, especially when working with partial inputs that might cause the LLM to generate unusual response structures.

---

Tracing and Observability

A multi-agent system without observability is a black box. You have no idea which agent is slow, which message is stuck, or where errors are originating. The message schema you built already includes the key fields needed for distributed tracing — you just need to hook them up to a tracing backend.

from opentelemetry import trace
from opentelemetry.trace.propagation.tracecontext import TraceContextTextMapPropagator

tracer = trace.get_tracer("multi_agent_rag")

class TracingMixin:
    """Mixin that adds OpenTelemetry tracing to any agent."""
    
    def _extract_trace_context(self, message: MessageEnvelope) -> trace.Context:
        carrier = message.metadata.get("trace_context", {})
        return TraceContextTextMapPropagator().extract(carrier)
    
    def _inject_trace_context(self, message: MessageEnvelope) -> None:
        carrier = {}
        TraceContextTextMapPropagator().inject(carrier)
        message.metadata["trace_context"] = carrier
    
    async def handle_message_traced(self, message: MessageEnvelope) -> Optional[MessageEnvelope]:
        ctx = self._extract_trace_context(message)
        
        with tracer.start_as_current_span(
            f"{self.agent_id}.handle_{message.message_type.value}",
            context=ctx,
            attributes={
                "agent.id": self.agent_id,
                "message.id": str(message.message_id),
                "correlation.id": str(message.correlation_id),
                "session.id": str(message.session_id),
                "message.type": message.message_type.value,
            }
        ) as span:
            try:
                result = await self.handle_message(message)
                if result is not None:
                    self._inject_trace_context(result)
                    span.set_attribute("result.message_type", result.message_type.value)
                return result
            except Exception as e:
                span.record_exception(e)
                span.set_status(trace.StatusCode.ERROR)
                raise

Beyond individual spans, track workflow-level metrics. How long does a full workflow take end-to-end? What's the p95 latency per agent? What's the retry rate? These metrics tell you where to invest optimization effort. A retrieval agent with 20% retry rate needs attention. An orchestrator with p95 latency of 50ms but a synthesis agent with p95 of 8000ms tells you where your bottleneck is.

---

Hands-On Exercise

Build a complete three-agent RAG system for answering research questions about public companies. Your system should include:

Agent 1: Financial Data Retrieval Agent Uses the message protocol you've built to receive RETRIEVAL_REQUEST messages for SEC filing data. Implement it to query a vector store (use pgvector or Chroma locally) with filters for ticker symbol and document type. Return a properly structured RETRIEVAL_RESULT message.

Agent 2: News Retrieval Agent Similar structure, but retrieves recent news articles. Simulate failures on 20% of requests by randomly raising a TimeoutError to verify your retry logic works correctly.

Agent 3: Synthesis Agent Receives SYNTHESIS_REQUEST messages containing results from both retrieval agents. Use GPT-4 or Claude to generate a structured response. The response should explicitly cite which chunks informed each claim. Return the result as a SYNTHESIS_RESULT message with a populated sources_cited field.

Orchestrator Wire these together with the orchestrator pattern. For the query "Summarize Apple's most recent earnings performance and any significant news in the last 30 days," the orchestrator should:

1. Fan out retrieval tasks to both specialized agents concurrently
2. Handle the simulated 20% failure rate from the news agent using exponential backoff retry
3. Trigger synthesis once both results are available (or after two failed retry attempts)
4. Store the complete workflow state and all messages in Redis
5. Print a workflow trace showing message sequence, latencies, and retry events

Stretch goal: Add a fourth agent that validates the synthesis output by checking each cited source actually supports the claim made. If it detects an unsupported claim, it should post a correction request back to the synthesis agent. This implements the self-correction loop described in Agentic RAG: Building Self-Correcting Retrieval Pipelines That Query, Reflect, and Retry.

---

Common Mistakes & Troubleshooting

Mistake 1: Shared State Without Concurrency Control

The most common bug in multi-agent systems is a read-modify-write race condition on shared state. Two agents both read completed_tasks: 2, both increment it to 3, and both write back completed_tasks: 3 — but the true value should be 4. The system thinks one fewer task is done than actually is.

The fix: use atomic operations. Redis INCR and HINCRBY are atomic. Lua scripts execute atomically. Postgres UPDATE ... SET count = count + 1 WHERE id = ... RETURNING * is atomic. Never read-modify-write without a transaction or atomic primitive.

Mistake 2: Unbounded Message Queues

If a synthesis agent processes messages slowly and retrieval agents fill its queue faster than it drains, you'll run out of memory. Always configure queue bounds and implement backpressure. In Redis, use LLEN to check queue depth before enqueuing. If it exceeds your threshold, either block the producer or route to a dead-letter queue for later processing.

Mistake 3: Missing Correlation IDs in Error Messages

When an agent posts an error back to the orchestrator, it's critical that the error message carries the same correlation_id as the original task. Without it, the orchestrator can't associate the error with the right workflow. Many teams get this right for success paths and forget it on error paths. Always use your _create_error_message helper or equivalent.

Mistake 4: Not Handling Idempotency

What happens if a message is delivered twice? In at-least-once delivery systems (most real message brokers), this is expected. Your agents must be idempotent — processing the same message twice should produce the same result, not double-count or double-execute. Use the message_id as an idempotency key. Before processing, check if you've already processed this message ID. If yes, return the cached result.

async def _is_already_processed(self, message_id: UUID) -> bool:
    key = f"processed:{message_id}"
    return await self._client.exists(key) > 0

async def _mark_processed(self, message_id: UUID, result: Any) -> None:
    key = f"processed:{message_id}"
    await self._client.set(key, json.dumps(result), ex=3600)

Mistake 5: Synchronous Agents in Async Systems

If any agent in your system makes synchronous blocking calls (synchronous HTTP requests, blocking database queries, time.sleep()), it will block the entire event loop and stall every other agent running in the same process. Use asyncio.to_thread() to run blocking code in a thread pool, or switch to async clients (aiohttp, asyncpg, etc.).

Troubleshooting: Workflow Stuck in IN_PROGRESS

If a workflow stays IN_PROGRESS indefinitely, it usually means a message was lost. Check:

1. Did the retrieval agent receive the task? (Check its queue or processing log)
2. Did it complete? (Check the stored messages for correlation_id)
3. Did its result message reach the orchestrator? (Check the orchestrator's queue depth)
4. Is the orchestrator's completion counter correct? (Compare completed_tasks + failed_tasks against task_count)

Adding a workflow timeout that fires after a configurable deadline and checks these conditions is worth the implementation cost. A workflow that never completes is worse than one that fails fast.

---

Architecture Trade-offs: When to Reach for This Complexity

Multi-agent communication protocols add real complexity. Before implementing everything in this lesson, ask yourself whether you actually need it.

Single-agent RAG with tool use is the right choice when your query types are predictable, your data sources are few, and latency requirements are lenient. If one well-prompted agent with access to multiple retrieval tools can handle 90% of your queries, don't build a distributed system. As you explore Building Multi-Step AI Agents with Planning and Memory, you'll see how far you can push a single agent before coordination becomes necessary.

Multi-agent with simple function calls (agents as Python functions, no message passing) is appropriate when workflows are sequential (not parallel), agents don't need to run in separate processes, and you don't need persistence across restarts. Many "multi-agent" systems in production are really just well-organized function call chains.

Full structured message passing with shared state — what we've built in this lesson — is the right choice when:

• You need parallel retrieval across multiple data sources and latency matters
• Agents may be deployed in separate processes or machines
• Workflows need to survive process restarts or agent failures
• You need full observability of inter-agent communication
• The system needs to support dynamic agent registration

The memory model for your agents also shapes this decision significantly. The interaction between tool memory and retrieval memory across agents requires careful thought — workflow state is a form of shared episodic memory, and your shared state store is essentially the multi-agent memory substrate.

Finally, consider that latency compounds. If your orchestrator takes 20ms to process, each retrieval agent takes 800ms, and synthesis takes 1500ms, your total latency is ~2300ms with proper parallelism — but it becomes 4100ms if your orchestrator blocks between stages. Always profile parallel versus sequential execution paths. Retrieval latency optimization strategies covered in Retrieval Latency Optimization become even more critical when retrieval is on the critical path for a multi-agent workflow.

---

Summary & Next Steps

You've built a complete agent-to-agent communication protocol from the ground up. Let's recap what you've implemented and why each piece matters:

Typed message schemas with versioned envelopes give you machine-verifiable contracts between agents. The correlation_id links messages across a workflow, enabling tracing and debugging. Schema versioning lets you evolve the protocol without breaking running workflows.

Shared state stores with atomic operations prevent the race conditions that silently corrupt workflow state in concurrent systems. The in-memory store serves development; the Redis-backed store handles production multi-process deployments.

Coordination patterns — orchestrator-worker, blackboard, and event-driven — each suit different problem shapes. Orchestrator-worker is the right default. Blackboard shines when you need dynamic agent discovery. Event-driven pipelines handle high throughput and maximum decoupling.

Failure handling with typed errors, retry policies with exponential backoff and jitter, and partial completion logic makes the system resilient to the constant low-level failures that characterize real production environments.

Observability through distributed tracing and workflow-level metrics transforms the system from a black box into something you can reason about, optimize, and debug.

Where to go next:

If the retrieval quality across your multiple specialized agents is inconsistent, explore Query Routing in RAG: How to Direct Questions to the Right Data Source or Retrieval Strategy — intelligent routing can improve both quality and latency by getting queries to the right agent faster.

If your synthesis agent is receiving too much context from multiple retrieval results, Contextual Compression in RAG: Filtering and Compressing Retrieved Chunks Before Passing to the LLM shows how to reduce payload size before synthesis without losing critical information.

If you're encountering hallucination in synthesized outputs that span multiple retrieved sources, Corrective RAG with Hallucination Detection provides the validation and self-correction patterns you need to plug into the synthesis stage of your multi-agent pipeline.

The architecture you've built is production-ready in structure. Fill in the domain-specific retrieval logic, connect it to your actual vector stores, and wire it to your LLM provider. The protocol handles the rest.