---
title: "Shared Memory Across Agent Teams: Building Collective Knowledge Bases"
description: "Design shared memory architectures for multi-agent teams that enable collective knowledge building, with contribution tracking, conflict resolution, and access control."
canonical: https://callsphere.ai/blog/shared-memory-agent-teams-collective-knowledge-bases
category: "Learn Agentic AI"
tags: ["Multi-Agent", "Shared Memory", "Collective Knowledge", "Python", "Agentic AI"]
author: "CallSphere Team"
published: 2026-03-17T00:00:00.000Z
updated: 2026-05-06T01:02:44.522Z
---

# Shared Memory Across Agent Teams: Building Collective Knowledge Bases

> Design shared memory architectures for multi-agent teams that enable collective knowledge building, with contribution tracking, conflict resolution, and access control.

## 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.

```mermaid
flowchart TD
    MSG(["New message"])
    WORKING["Working memory
rolling window"]
    EPISODIC[("Episodic memory
past sessions")]
    SEMANTIC[("Semantic memory
facts and preferences")]
    SUM["Summarizer
compresses old turns"]
    ROUTER{"Retrieve
needed memories"}
    PROMPT["Assembled context"]
    LLM["LLM"]
    UPD["Memory updater
writes new facts"]
    MSG --> WORKING --> ROUTER
    ROUTER -->|Past sessions| EPISODIC
    ROUTER -->|User facts| SEMANTIC
    EPISODIC --> SUM --> PROMPT
    SEMANTIC --> PROMPT
    WORKING --> PROMPT --> LLM --> UPD
    UPD --> EPISODIC
    UPD --> SEMANTIC
    style ROUTER fill:#4f46e5,stroke:#4338ca,color:#fff
    style LLM fill:#f59e0b,stroke:#d97706,color:#1f2937
    style EPISODIC fill:#ede9fe,stroke:#7c3aed,color:#1e1b4b
    style SEMANTIC fill:#ede9fe,stroke:#7c3aed,color:#1e1b4b
```

```python
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.

```python
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.

```python
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.

```python
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.

```python
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

---

Source: https://callsphere.ai/blog/shared-memory-agent-teams-collective-knowledge-bases