0Pricing
AI Engineering Academy · Lección

Memoria compartida y comunicación entre agentes

Implemente una capa de memoria compartida mediante un almacén de clave-valor del que los agentes lean y en el que escriban, permitiendo la colaboración asíncrona sin un acoplamiento estrecho entre agentes.

Memoria compartida y comunicación entre agentes es una lección gratuita de AI Engineering Academy en CoddyKit. Esta es la lección 4 de 4. Puedes leer la lección completa abajo gratuitamente — luego la practicas en el navegador con un editor de código integrado y un tutor de IA 24/7. Forma parte de la ruta de aprendizaje de AI Engineering Academy, y tu progreso se sincroniza en la web y la app de CoddyKit. El curso de AI Engineering Academy incluye 4 lecciones en total.

Partes de esta lección aún no han sido traducidas y se muestran en inglés.

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.

Preguntas frecuentes

¿La lección «Memoria compartida y comunicación entre agentes» es gratis?

Sí — el texto completo de «Memoria compartida y comunicación entre agentes» es gratis para leer aquí en la web. Para practicarla de forma interactiva (editor de código integrado y tutor de IA 24/7) y desbloquear el resto del curso de AI Engineering Academy, actualiza a CoddyKit PRO. El curso de AI Engineering Academy incluye 4 lecciones en total.

¿Qué aprenderé en «Memoria compartida y comunicación entre agentes»?

Implemente una capa de memoria compartida mediante un almacén de clave-valor del que los agentes lean y en el que escriban, permitiendo la colaboración asíncrona sin un acoplamiento estrecho entre ag… Practicas AI Engineering Academy con código real que ejecutas directamente en el navegador, y un tutor de IA 24/7 responde tus preguntas mientras trabajas en la lección.

¿Necesito experiencia previa para empezar AI Engineering Academy?

No se requiere experiencia previa. AI Engineering Academy en CoddyKit está estructurado para principiantes hasta estudiantes avanzados, así que puedes empezar aquí o desde el inicio y avanzar a tu ritmo. Esta es la lección 4 de 4.

¿Cuánto tiempo toma la lección «Memoria compartida y comunicación entre agentes»?

La mayoría de las lecciones de CoddyKit toman alrededor de 5–10 minutos. Cada una es compacta e interactiva, así que avanzas constantemente y retomas exactamente por donde dejaste en la web y la app.

¿Puedo escribir y ejecutar código en esta lección de AI Engineering Academy?

Sí. Cada lección de AI Engineering Academy incluye un editor de código integrado, así que escribes y ejecutas código real directamente en tu navegador y obtienes retroalimentación instantánea de IA — sin configuración local necesaria.

Todas las lecciones de este curso

  1. Por qué los agentes individuales llegan a un límite
  2. Patrón de orquestador y subagentes
  3. Creación de pipelines multiagente con LangGraph
  4. Memoria compartida y comunicación entre agentes
← Volver a AI Engineering Academy