State-Sharing Memory: Implementing Thread-Safe Shared State in Multi-Agent Systems

In multi-agent systems, agents must share context to collaborate effectively. There are two primary communication models:

1. Message Passing: Agents send messages directly to each other.
2. Shared State (Blackboard Pattern): Agents write and read from a centralized memory database.

While message passing works well for linear pipelines, it quickly becomes unwieldy in complex workflows. Centralized shared memory simplifies communication but introduces challenges around concurrency, write conflicts, and state consistency.

State-Sharing Memory Architecture

The Architecture of Shared Memory

In a shared memory system, the entire state is represented as a structured schema (e.g., a JSON document or relational database). Multiple agents query and update this state concurrently.

To prevent race conditions, the system must enforce concurrency controls. The two main approaches are:

• Optimistic Concurrency Control (OCC): Each state update includes a version number. If the version in the database doesn't match the agent's read version, the update is rejected and the agent retries.

\text{Update Condition: } V_{current} = V_{expected}

• Pessimistic Locking: The state is locked by an agent while performing calculations, preventing other agents from reading or writing until the lock is released.

Python Code: A Thread-Safe State Store

Here is a Python class that manages a shared, thread-safe memory state using threading.Lock and versioning:

import threading
from typing import Dict, Any

class ThreadSafeStateStore:
    def __init__(self):
        self._lock = threading.Lock()
        self._state: Dict[str, Any] = {}
        self._version = 0

    def get_state(self) -> Dict[str, Any]:
        with self._lock:
            return self._state.copy(), self._version

    def update_state(self, updates: Dict[str, Any], expected_version: int) -> bool:
        with self._lock:
            if self._version != expected_version:
                # Concurrent edit detected
                return False
            
            # Apply updates
            self._state.update(updates)
            self._version += 1
            return True

# Example Usage
store = ThreadSafeStateStore()
state, ver = store.get_state()
success = store.update_state({"agent_alpha_status": "searching"}, expected_version=ver)
print(f"Update status: {success}, New version: {store.get_state()[1]}")

Implementing Distributed Shared Memory

In production environments, memory state is often stored in external databases like Redis or PostgreSQL:

• Redis Hash Map: Ideal for fast, key-value storage. Redis's single-threaded nature guarantees atomic updates.
• PostgreSQL JSONB: Great for queries over structured states. PostgreSQL provides robust transactional isolation levels (e.g., SERIALIZABLE).

Conclusion

Designing thread-safe, centralized shared memory is essential for stable multi-agent collaboration. Implementing version tracking and atomic lock operations ensures agents can safely collaborate without data loss or race conditions.