0Pricing
AI Engineering Academy · Lektion

Gemeinsamer Speicher und Kommunikation zwischen Agents

Implementieren Sie eine gemeinsame Speicherschicht mit einem Key-Value-Store, aus dem Agents lesen und in den sie schreiben können. So wird asynchrone Zusammenarbeit ohne enge Kopplung zwischen den Agents möglich.

Gemeinsamer Speicher und Kommunikation zwischen Agents ist eine kostenlose AI Engineering Academy-Lektion auf CoddyKit. Dies ist Lektion 4 von 4. Du kannst die komplette Lektion unten kostenlos lesen – dann übst du sie direkt im Browser mit einem integrierten Code-Editor und einem KI-Tutor rund um die Uhr. Sie ist Teil des AI Engineering Academy-Lernpfads, und dein Fortschritt wird über Web und CoddyKit-App synchronisiert. Der AI Engineering Academy-Kurs umfasst insgesamt 4 Lektionen.

Teile dieser Lektion wurden noch nicht übersetzt und werden auf Englisch angezeigt.

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.

Häufig gestellte Fragen

Ist die Lektion „Gemeinsamer Speicher und Kommunikation zwischen Agents“ kostenlos?

Ja — der vollständige Text von „Gemeinsamer Speicher und Kommunikation zwischen Agents“ ist hier im Web kostenlos zu lesen. Um sie interaktiv zu üben (integrierter Code-Editor und 24/7 KI-Tutor) und den Rest des AI Engineering Academy-Kurses freizuschalten, upgrade auf CoddyKit PRO. Der AI Engineering Academy-Kurs umfasst insgesamt 4 Lektionen.

Was lerne ich in „Gemeinsamer Speicher und Kommunikation zwischen Agents“?

Implementieren Sie eine gemeinsame Speicherschicht mit einem Key-Value-Store, aus dem Agents lesen und in den sie schreiben können. So wird asynchrone Zusammenarbeit ohne enge Kopplung zwischen den A… Du übst AI Engineering Academy mit praktischem Code, den du direkt im Browser ausführst, und ein 24/7 KI-Tutor beantwortet deine Fragen während du die Lektion bearbeitest.

Brauche ich Erfahrung, um AI Engineering Academy zu starten?

Keine Vorkenntnisse erforderlich. AI Engineering Academy auf CoddyKit ist für Anfänger bis fortgeschrittene Lernende strukturiert, sodass du hier starten oder von Anfang an beginnen und in deinem eigenen Tempo voranschreiten kannst. Dies ist Lektion 4 von 4.

Wie lange dauert die Lektion „Gemeinsamer Speicher und Kommunikation zwischen Agents“?

Die meisten CoddyKit-Lektionen dauern etwa 5–10 Minuten. Jede ist kompakt und interaktiv, sodass du stetig Fortschritte machst und genau dort weitermachst, wo du aufgehört hast – im Web und in der App.

Kann ich in dieser AI Engineering Academy-Lektion Code schreiben und ausführen?

Ja. Jede AI Engineering Academy-Lektion enthält einen integrierten Code-Editor, sodass du echten Code direkt in deinem Browser schreibst und ausführst und sofort KI-Feedback erhältst — ohne lokale Einrichtung erforderlich.

Alle Lektionen in diesem Kurs

  1. Warum einzelne Agents an ihre Grenzen stoßen
  2. Das Orchestrator-Subagent-Muster
  3. Multi-Agent-Pipelines mit LangGraph entwickeln
  4. Gemeinsamer Speicher und Kommunikation zwischen Agents
← Zurück zu AI Engineering Academy