0Pricing
AI Engineering Academy · Leçon

Mémoire partagée et communication inter-agents

Implémentez une couche de mémoire partagée à l’aide d’un magasin clé-valeur dans lequel les agents lisent et écrivent, afin de permettre une collaboration asynchrone sans couplage étroit entre eux.

Mémoire partagée et communication inter-agents est une leçon AI Engineering Academy gratuite sur CoddyKit. Ceci est la leçon 4 sur 4. Tu peux lire la leçon complète ci-dessous gratuitement — puis la pratiquer en direct dans le navigateur avec un éditeur de code intégré et un tuteur IA 24/7. Elle fait partie du parcours d'apprentissage AI Engineering Academy, et ta progression se synchronise sur le web et l'application CoddyKit. Le cours AI Engineering Academy comprend 4 leçons au total.

Certaines parties de cette leçon n'ont pas encore été traduites et s'affichent en anglais.

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.

Questions Fréquemment Posées

La leçon « Mémoire partagée et communication inter-agents » est-elle gratuite ?

Oui — le texte complet de « Mémoire partagée et communication inter-agents » est gratuit à lire ici sur le web. Pour la pratiquer de manière interactive (un éditeur de code intégré et un tuteur IA 24/7) et déverrouiller le reste du cours AI Engineering Academy, passe à CoddyKit PRO. Le cours AI Engineering Academy comprend 4 leçons au total.

Qu'est-ce que j'apprendrai dans « Mémoire partagée et communication inter-agents » ?

Implémentez une couche de mémoire partagée à l’aide d’un magasin clé-valeur dans lequel les agents lisent et écrivent, afin de permettre une collaboration asynchrone sans couplage étroit entre eux. Tu pratiques AI Engineering Academy avec du code pratique que tu exécutes directement dans le navigateur, et un tuteur IA 24/7 répond à tes questions au fur et à mesure que tu avances dans la leçon.

Dois-je avoir de l'expérience pour commencer AI Engineering Academy ?

Aucune expérience préalable n'est requise. AI Engineering Academy sur CoddyKit est structuré pour les débutants jusqu'aux apprenants avancés, donc tu peux commencer ici ou depuis le début et avancer à ton rythme. Ceci est la leçon 4 sur 4.

Combien de temps prend la leçon « Mémoire partagée et communication inter-agents » ?

La plupart des leçons CoddyKit prennent environ 5–10 minutes. Chacune est courte et interactive, tu progresses régulièrement et tu repiques exactement où tu t'es arrêté sur le web et l'app.

Peux-tu écrire et exécuter du code dans cette leçon AI Engineering Academy ?

Oui. Chaque leçon AI Engineering Academy inclut un éditeur de code intégré, tu écris et exécutes du vrai code directement dans ton navigateur et tu reçois des retours IA instantanés — aucune configuration locale requise.

Toutes les leçons de ce cours

  1. Pourquoi les agents uniques atteignent leurs limites
  2. Modèle orchestrateur-sous-agent
  3. Créer des pipelines multi-agents avec LangGraph
  4. Mémoire partagée et communication inter-agents
← Retour à AI Engineering Academy