0Pricing
AI Engineering Academy · Lesson

Shared Memory and Inter-Agent Communication

Implement a shared memory layer using a key-value store that agents read from and write to, enabling asynchronous collaboration without tight coupling between agents.

Shared Memory and Inter-Agent Communication is a free AI Engineering Academy lesson on CoddyKit — lesson 4 of 4. You can read the complete lesson below for free — then practise it hands-on in the browser with a built-in code editor and a 24/7 AI tutor. It is part of the AI Engineering Academy learning path, one of 4 lessons in the course, and your progress syncs across the web and the CoddyKit app.

The Problem of Agent Isolation

In a multi-agent system, each agent runs in its own context and has no awareness of what other agents are doing or have done. If a researcher agent discovers a key fact, how does the writer agent know about it? How does a monitoring agent know when the coder agent encounters an error? Without a shared communication mechanism, agents are isolated silos that cannot collaborate effectively.

Shared Memory: The Blackboard Model

The classic solution for multi-agent communication is the blackboard model: a shared data store (the blackboard) that any agent can read from or write to. Agents post their findings, read others' contributions, and coordinate implicitly through the shared state. This model decouples agents from each other — they do not need to know about each other's existence, only about the structure of the shared memory.

# Simple in-memory blackboard using a dictionary
from threading import Lock

class Blackboard:
    def __init__(self):
        self._data = {}
        self._lock = Lock()  # thread-safe for parallel agents

    def write(self, key: str, value, agent_id: str):
        with self._lock:
            self._data[key] = {'value': value, 'written_by': agent_id}
            print(f'[{agent_id}] wrote: {key}')

    def read(self, key: str):
        with self._lock:
            return self._data.get(key, {}).get('value')

    def keys(self):
        with self._lock:
            return list(self._data.keys())

blackboard = Blackboard()

Persistent Shared Memory with Redis

For production multi-agent systems where agents run as separate processes or services, in-memory dictionaries are not sufficient. Redis is the most popular choice for shared agent memory: it is fast, supports rich data types (strings, hashes, lists, sorted sets), has built-in TTL for expiration, and handles concurrent reads and writes safely with atomic operations.

import redis
import json

class RedisSharedMemory:
    def __init__(self, prefix='agent:'):
        self.redis = redis.Redis(host='localhost', port=6379, decode_responses=True)
        self.prefix = prefix

    def set(self, key: str, value, ttl_seconds=3600):
        full_key = self.prefix + key
        self.redis.setex(full_key, ttl_seconds, json.dumps(value))

    def get(self, key: str):
        full_key = self.prefix + key
        raw = self.redis.get(full_key)
        return json.loads(raw) if raw else None

    def append_to_list(self, key: str, item):
        full_key = self.prefix + key
        self.redis.rpush(full_key, json.dumps(item))

    def get_list(self, key: str):
        full_key = self.prefix + key
        return [json.loads(x) for x in self.redis.lrange(full_key, 0, -1)]

memory = RedisSharedMemory(prefix='research_project:')

Namespacing Shared Memory

In complex multi-agent systems, agents write many different types of data and a flat key namespace quickly becomes chaotic. Use hierarchical namespacing to organize shared memory clearly. A common pattern is project_id:agent_role:data_type, for example proj_123:researcher:findings or proj_123:coder:error_log. This makes it easy to query all data for a project or all output from a specific agent.

class NamespacedMemory:
    def __init__(self, project_id: str, agent_id: str, redis_client):
        self.base = f'{project_id}:{agent_id}'
        self.redis = redis_client

    def write_finding(self, topic: str, content: str):
        key = f'{self.base}:findings:{topic}'
        self.redis.set(key, content)

    def read_all_findings(self, project_id: str):
        # Read findings from ALL agents in this project
        pattern = f'{project_id}:*:findings:*'
        keys = self.redis.keys(pattern)
        return {k: self.redis.get(k) for k in keys}

# Usage
researcher_memory = NamespacedMemory('proj_123', 'researcher', redis_client)
researcher_memory.write_finding('competitors', 'OpenAI, Anthropic, Google...')

writer_memory = NamespacedMemory('proj_123', 'writer', redis_client)
all_findings = writer_memory.read_all_findings('proj_123')

Structured vs Unstructured Memory

Shared memory content can be unstructured (raw text blobs the LLM will read) or structured (JSON/Python objects with typed fields). Structured memory is preferable because it enables programmatic queries, validation, and merging. Always define a schema for what each agent writes to shared memory, document it, and validate writes against it to prevent one buggy agent from corrupting the memory store.

from pydantic import BaseModel
from typing import Optional, list
from datetime import datetime

class ResearchFinding(BaseModel):
    topic: str
    summary: str
    sources: list[str]
    confidence: float  # 0.0 to 1.0
    written_by: str
    timestamp: datetime

# Validated write - bad data is caught before it enters shared memory
def write_finding(memory, finding_dict: dict):
    finding = ResearchFinding(**finding_dict)  # validates on creation
    memory.set(f'findings:{finding.topic}', finding.model_dump())
    print(f'Validated finding written for topic: {finding.topic}')

Event-Driven Communication with Pub/Sub

Instead of polling shared memory to check for updates, agents can use publish/subscribe (pub/sub) communication to notify other agents when they complete a task. Agent A publishes an event ('research_complete'), and Agent B, which is subscribed to that event, wakes up and starts processing. Redis pub/sub and message queues like RabbitMQ or Kafka support this pattern.

import redis

# Publisher (researcher agent)
def researcher_agent(topic, redis_client):
    findings = do_research(topic)
    redis_client.set(f'findings:{topic}', findings)
    
    # Notify all subscribers that research is done
    redis_client.publish('agent_events', f'research_complete:{topic}')
    print(f'Research complete, published event for topic: {topic}')

# Subscriber (writer agent) - runs in separate process
def writer_agent_listener(redis_client):
    pubsub = redis_client.pubsub()
    pubsub.subscribe('agent_events')
    
    for message in pubsub.listen():
        if message['type'] == 'message':
            event = message['data']
            if event.startswith('research_complete:'):
                topic = event.split(':')[1]
                findings = redis_client.get(f'findings:{topic}')
                write_draft(findings)  # start writing immediately

Shared Memory in LangGraph

In LangGraph, shared memory between agents is the graph state object itself. Each node reads from and writes to the same typed state dictionary. LangGraph handles the read/write coordination automatically. For more complex scenarios, you can also inject an external memory client (Redis, database) into each node function via dependency injection.

from langgraph.graph import StateGraph
from typing import TypedDict

class SharedState(TypedDict):
    # All shared data lives here - every node can read any field
    query: str
    research_findings: str   # written by researcher, read by writer
    written_draft: str       # written by writer, read by reviewer
    review_notes: str        # written by reviewer, read by writer (loop)
    final_output: str        # written by synthesizer

# Researcher writes to 'research_findings'
def researcher(state: SharedState) -> dict:
    findings = search_and_summarize(state['query'])
    return {'research_findings': findings}  # partial state update

# Writer reads 'research_findings', writes 'written_draft'
def writer(state: SharedState) -> dict:
    draft = write_from_findings(state['research_findings'])  # reads researcher output
    return {'written_draft': draft}

Memory Conflicts and Consistency

When multiple agents write to shared memory concurrently, write conflicts can occur. Two agents might overwrite each other's work or read stale data between a read and a write. Handle this with: optimistic locking (check version before writing), atomic compare-and-swap operations in Redis, or serializing writes through an orchestrator agent that is the sole writer to critical memory fields.

# Optimistic locking with Redis
def safe_write(redis_client, key, new_value, expected_version):
    with redis_client.pipeline() as pipe:
        try:
            pipe.watch(key + ':version')  # watch for concurrent modification
            current_version = int(pipe.get(key + ':version') or 0)
            
            if current_version != expected_version:
                raise ValueError(f'Version conflict: expected {expected_version}, got {current_version}')
            
            pipe.multi()  # start transaction
            pipe.set(key, new_value)
            pipe.set(key + ':version', current_version + 1)
            pipe.execute()  # atomic commit
            print('Write successful')
        except redis.WatchError:
            print('Conflict detected, retry write')

Memory TTL and Cleanup

Shared agent memory accumulates over time and can grow unbounded if not managed. Always set time-to-live (TTL) on memory entries so they expire automatically. For project-scoped memory, clean up all entries when the project completes. Use TTL values that match the expected duration of a workflow: short TTLs (minutes) for transient data, longer TTLs (hours to days) for results that may be reused.

def cleanup_project_memory(redis_client, project_id: str):
    pattern = f'{project_id}:*'
    keys = redis_client.keys(pattern)
    if keys:
        redis_client.delete(*keys)
        print(f'Cleaned up {len(keys)} memory entries for project {project_id}')

# Set TTL when writing
def write_with_ttl(redis_client, key, value, ttl_hours=2):
    redis_client.setex(
        key,
        ttl_hours * 3600,  # convert to seconds
        json.dumps(value)
    )

# Register cleanup callback when workflow completes
def on_workflow_complete(project_id):
    cleanup_project_memory(redis_client, project_id)
    print(f'Workflow {project_id} complete, memory cleaned up')

Memory as Agent History

Shared memory can also store the history of agent actions, not just their outputs. Recording which agent did what, when, and why creates an audit trail that is invaluable for debugging failures, understanding how a final output was produced, and resuming interrupted workflows. This action log is the memory-based equivalent of LangSmith tracing.

import time
from dataclasses import dataclass

@dataclass
class AgentAction:
    agent_id: str
    action_type: str     # 'research', 'write', 'review', 'tool_call'
    input_summary: str
    output_summary: str
    timestamp: float
    success: bool

def log_action(memory, action: AgentAction):
    key = f'action_log:{action.agent_id}:{action.timestamp}'
    memory.set(key, vars(action))

# Usage in an agent
def researcher_with_logging(state, memory):
    start = time.time()
    findings = do_research(state['query'])
    log_action(memory, AgentAction(
        agent_id='researcher',
        action_type='research',
        input_summary=state['query'][:100],
        output_summary=findings[:100],
        timestamp=start,
        success=True
    ))
    return findings

Choosing a Memory Architecture

The right shared memory architecture depends on your deployment model. For single-process LangGraph workflows, the graph state is sufficient. For multi-process or distributed agents, use Redis. For long-lived projects that need durability across restarts, use a relational database with proper indexing. For event-driven coordination between services, add pub/sub on top of whatever storage you choose.

Quick Check

Test your understanding of shared memory and inter-agent communication from this lesson.

Lesson Recap

In this lesson you learned: the blackboard model uses a shared data store where any agent can read and write, enabling implicit coordination without direct agent-to-agent coupling, Redis is the go-to choice for persistent shared memory in distributed multi-agent systems, and pub/sub enables event-driven communication so agents react immediately when dependencies complete. Next up we explore the code execution loop for agent tasks.

Frequently asked questions

Is the “Shared Memory and Inter-Agent Communication” lesson free?

Yes — the full text of “Shared Memory and Inter-Agent Communication” is free to read here on the web, and the AI Engineering Academy course includes 4 lessons in total. To practise it interactively (a built-in code editor and a 24/7 AI tutor) and unlock the rest of the AI Engineering Academy course, upgrade to CoddyKit PRO.

What will I learn in “Shared Memory and Inter-Agent Communication”?

Implement a shared memory layer using a key-value store that agents read from and write to, enabling asynchronous collaboration without tight coupling between agents. You practise AI Engineering Academy with hands-on code you run directly in the browser, and a 24/7 AI tutor answers your questions as you work through the lesson.

Do I need any experience to start AI Engineering Academy?

No prior experience is required. AI Engineering Academy on CoddyKit is structured for beginners through advanced learners; this is — lesson 4 of 4, so you can start here or from the beginning and move at your own pace.

How long does the “Shared Memory and Inter-Agent Communication” lesson take?

Most CoddyKit lessons take about 5–10 minutes. Each one is bite-sized and interactive, so you make steady progress and pick up exactly where you left off across the web and the app.

Can I write and run code in this AI Engineering Academy lesson?

Yes. Every AI Engineering Academy lesson includes a built-in code editor, so you write and run real code right in your browser and get instant AI feedback — no local setup required.

All lessons in this course

  1. Why Single Agents Hit a Wall
  2. Orchestrator-Subagent Pattern
  3. Building Multi-Agent Pipelines with LangGraph
  4. Shared Memory and Inter-Agent Communication
← Back to AI Engineering Academy