Shared Memory Across Agent Teams: Building Collective Knowledge Bases

Why Individual Memory Is Not Enough

In multi-agent architectures, each agent typically maintains its own private memory. A research agent learns facts, a planning agent tracks goals, and a coding agent remembers solutions. But when these agents collaborate, they need to share knowledge. The research agent discovers that an API is deprecated — the coding agent needs to know this immediately, not after it generates code that fails.

Shared memory gives agent teams a collective knowledge base where any agent can read and contribute. Designing it well requires solving contribution tracking, conflict resolution, and access control.

Shared Memory Architecture

The architecture separates private agent memory from shared team memory. Each agent reads from both stores but writes to shared memory only when the information is relevant to the team.

from dataclasses import dataclass, field
from datetime import datetime
from enum import Enum
from typing import Optional
import threading


class AccessLevel(Enum):
    READ = "read"
    WRITE = "write"
    ADMIN = "admin"


@dataclass
class SharedMemoryEntry:
    content: str
    author_agent: str
    created_at: datetime
    category: str = "general"
    confidence: float = 0.8
    version: int = 1
    supersedes: Optional[str] = None
    tags: list[str] = field(default_factory=list)
    id: str = ""


class SharedMemoryStore:
    def __init__(self):
        self.entries: dict[str, SharedMemoryEntry] = {}
        self.access_control: dict[str, AccessLevel] = {}
        self._lock = threading.Lock()
        self._next_id = 0

    def register_agent(
        self, agent_id: str, level: AccessLevel = AccessLevel.WRITE
    ):
        self.access_control[agent_id] = level

    def _gen_id(self) -> str:
        self._next_id += 1
        return f"shared_{self._next_id:06d}"

    def contribute(
        self,
        agent_id: str,
        content: str,
        category: str = "general",
        confidence: float = 0.8,
        tags: list[str] | None = None,
    ) -> str | None:
        if self.access_control.get(agent_id) not in (
            AccessLevel.WRITE,
            AccessLevel.ADMIN,
        ):
            return None

        with self._lock:
            entry_id = self._gen_id()
            entry = SharedMemoryEntry(
                id=entry_id,
                content=content,
                author_agent=agent_id,
                created_at=datetime.now(),
                category=category,
                confidence=confidence,
                tags=tags or [],
            )
            self.entries[entry_id] = entry
        return entry_id

Contribution Tracking

Every shared memory entry records which agent contributed it, when, and with what confidence level. This provenance information is critical for debugging and for resolving conflicts when agents disagree.

def get_contributions_by_agent(
    self, agent_id: str
) -> list[SharedMemoryEntry]:
    return [
        e for e in self.entries.values()
        if e.author_agent == agent_id
    ]


def get_contributions_by_category(
    self, category: str
) -> list[SharedMemoryEntry]:
    return sorted(
        [
            e for e in self.entries.values()
            if e.category == category
        ],
        key=lambda e: e.created_at,
        reverse=True,
    )

Tracking contributions also enables accountability. If the coding agent generates incorrect code because the research agent contributed a wrong fact, the provenance trail makes the root cause traceable.

Conflict Resolution

When two agents contribute contradictory information to shared memory, the system needs a resolution strategy. Three common approaches work in practice.

Latest-wins — the most recent contribution supersedes older ones. Simple but fragile if a less reliable agent writes after a more reliable one.

Confidence-weighted — higher-confidence contributions take precedence. Each agent sets its confidence based on how certain it is about the fact.

Voting — when multiple agents contribute on the same topic, the majority view wins.

def resolve_conflict(
    self,
    existing_id: str,
    new_content: str,
    new_agent: str,
    new_confidence: float,
    strategy: str = "confidence",
) -> str | None:
    existing = self.entries.get(existing_id)
    if not existing:
        return None

    with self._lock:
        if strategy == "latest":
            new_id = self._gen_id()
            entry = SharedMemoryEntry(
                id=new_id,
                content=new_content,
                author_agent=new_agent,
                created_at=datetime.now(),
                confidence=new_confidence,
                supersedes=existing_id,
            )
            self.entries[new_id] = entry
            return new_id

        elif strategy == "confidence":
            if new_confidence > existing.confidence:
                new_id = self._gen_id()
                entry = SharedMemoryEntry(
                    id=new_id,
                    content=new_content,
                    author_agent=new_agent,
                    created_at=datetime.now(),
                    confidence=new_confidence,
                    supersedes=existing_id,
                )
                self.entries[new_id] = entry
                return new_id
            return None  # Existing entry has higher confidence

    return None

Access Control

Not every agent should read or write every category of shared memory. A security-sensitive agent may contribute API credentials that only the deployment agent should access. Category-based access control keeps sensitive information partitioned.

def query(
    self,
    agent_id: str,
    category: str | None = None,
    tags: list[str] | None = None,
    top_k: int = 10,
) -> list[SharedMemoryEntry]:
    if agent_id not in self.access_control:
        return []

    results = list(self.entries.values())
    # Filter superseded entries
    superseded = {
        e.supersedes for e in results if e.supersedes
    }
    results = [e for e in results if e.id not in superseded]

    if category:
        results = [
            e for e in results if e.category == category
        ]
    if tags:
        tag_set = set(tags)
        results = [
            e for e in results
            if tag_set & set(e.tags)
        ]

    results.sort(key=lambda e: e.created_at, reverse=True)
    return results[:top_k]

Practical Usage Pattern

In a typical multi-agent pipeline, the orchestrator sets up shared memory and passes it to each agent during execution.

shared = SharedMemoryStore()
shared.register_agent("researcher", AccessLevel.WRITE)
shared.register_agent("planner", AccessLevel.WRITE)
shared.register_agent("coder", AccessLevel.READ)

# Researcher discovers a fact
shared.contribute(
    "researcher",
    "The payments API v2 endpoint requires OAuth2 bearer tokens",
    category="api_facts",
    confidence=0.95,
    tags=["payments", "auth"],
)

# Coder queries shared memory before generating code
api_facts = shared.query("coder", category="api_facts")

FAQ

How do I prevent shared memory from growing unboundedly?

Apply the same consolidation and decay strategies as individual memory. Periodically summarize entries within each category and archive the originals. Set a maximum entry count per category and evict low-confidence, old entries when the limit is reached.

Should agents be able to delete other agents' contributions?

Generally no — only ADMIN-level agents should delete. Instead, use the supersedes mechanism where new entries replace old ones without deleting the history. This preserves the audit trail while keeping retrieval results current.

How do I handle concurrent writes from multiple agents?

The threading lock in the implementation prevents data corruption. For distributed agent teams running across multiple processes, replace the in-memory store with a database like PostgreSQL or Redis, which provides atomic operations natively.

---

#MultiAgent #SharedMemory #CollectiveKnowledge #Python #AgenticAI #LearnAI #AIEngineering