Shared Knowledge Bases Across Agent Networks

In large enterprise agent networks, keeping each agent’s knowledge in its own private silo creates a painful consistency problem: ten agents may have ten slightly different understandings of the same pricing policy. A shared knowledge base solves this by making every agent draw from the same authoritative vector store: but it introduces new challenges around access control, versioning, and keeping embeddings fresh as source documents change.

Shared vs Per-Agent Memory

The instinct when building a multi-agent system is to give each agent its own memory: a dedicated vector store, its own embeddings, its own context window. This feels clean and isolated, but at enterprise scale it creates severe problems.

flowchart TD
    subgraph "Per-Agent Memory (Problematic)"
        SA["Sales Agent\n+ its own KB"] 
        FA["Finance Agent\n+ its own KB"]
        LA["Legal Agent\n+ its own KB"]
        SA -. "Different version\nof pricing policy" .-> W1["❌ Inconsistency"]
        FA -. "Different version\nof pricing policy" .-> W1
        LA -. "Different version\nof pricing policy" .-> W1
    end

    subgraph "Shared Knowledge Base (Recommended)"
        SKB[("Qdrant\nShared KB")]
        SA2["Sales Agent"]
        FA2["Finance Agent"]
        LA2["Legal Agent"]
        SA2 -->|"read: contracts\nread: pricing"| SKB
        FA2 -->|"read: pricing\nread: budgets"| SKB
        LA2 -->|"read: contracts\nread: compliance"| SKB
    end

    style SKB fill:#EEF0F7,stroke:#6366F1,color:#0F172A
    style W1 fill:#FEF3C7,stroke:#F59E0B,color:#0F172A
    style SA2 fill:#CCFBF1,stroke:#0D9488,color:#0F172A
    style FA2 fill:#CCFBF1,stroke:#0D9488,color:#0F172A
    style LA2 fill:#CCFBF1,stroke:#0D9488,color:#0F172A

With a shared store you embed each document exactly once, updates propagate immediately to all agents, and an agent’s new discovery is instantly available to its peers. The trade-off is that you must now enforce access control at query time.

Qdrant Collections with Agent-Namespaced Payloads

Every point in your Qdrant collection should carry a rich payload that encodes not just the content but who created it, which agents may read it, and when it was last verified.

The payload schema is the foundation of everything else: access control, versioning, and staleness detection all hang off these fields.

import asyncio
from datetime import datetime, timezone
from typing import Optional
from uuid import uuid4

from pydantic import BaseModel, Field
from qdrant_client import AsyncQdrantClient
from qdrant_client.models import (
    Distance,
    FieldCondition,
    Filter,
    MatchAny,
    PointStruct,
    VectorParams,
)

# Pydantic model for payload enforces schema at write time.
# Untyped dicts silently drop fields; this prevents that.
class KnowledgePayload(BaseModel):
    # The raw text chunk: stored alongside the vector so we can
    # return it directly without a separate document fetch.
    text: str

    # Source metadata lets agents cite where knowledge came from.
    source_doc_id: str
    source_url: Optional[str] = None
    source_type: str  # "confluence", "sharepoint", "salesforce", etc.

    # Which agent originally wrote this point. Used for audit trails.
    created_by_agent: str

    # Access control: list of agent IDs allowed to read this point.
    # An empty list means the point is readable by no one: use ["*"] for public.
    # Never allow agents to issue queries without filtering on this field.
    allowed_agents: list[str]

    # Versioning: monotonically increasing. When a source doc changes,
    # we bump this and set is_current=False on the old point.
    version: int = 1
    is_current: bool = True

    # ISO-8601 timestamps stored as strings for Qdrant payload filtering.
    # Qdrant supports range filters on numeric epoch timestamps too,
    # but strings are more human-readable in the dashboard.
    created_at: str = Field(
        default_factory=lambda: datetime.now(timezone.utc).isoformat()
    )
    last_verified_at: str = Field(
        default_factory=lambda: datetime.now(timezone.utc).isoformat()
    )

    # Optional: department or topic namespace for coarse filtering
    # before semantic search: dramatically reduces search space.
    namespace: Optional[str] = None


async def upsert_knowledge(
    client: AsyncQdrantClient,
    collection_name: str,
    embedding: list[float],
    payload: KnowledgePayload,
    point_id: Optional[str] = None,
) -> str:
    """Write one knowledge point. Returns the point ID."""
    pid = point_id or str(uuid4())

    # payload.model_dump() serializes to dict with Pydantic validation.
    # Using model_dump(mode="json") ensures datetime objects become strings.
    await client.upsert(
        collection_name=collection_name,
        points=[PointStruct(id=pid, vector=embedding, payload=payload.model_dump())],
    )
    return pid

Knowledge Access Control at Query Time

Access control in Qdrant is enforced through payload filters. Every query issued by an agent must include a must condition that checks allowed_agents. Without this filter, an agent can read any point in the collection.

async def agent_search(
    client: AsyncQdrantClient,
    collection_name: str,
    query_vector: list[float],
    agent_id: str,
    top_k: int = 5,
    namespace: Optional[str] = None,
) -> list[dict]:
    """
    Semantic search scoped to what this agent is allowed to read.

    The access filter is non-negotiable, it must be applied even if the
    agent requests an unfiltered search. Never expose a raw search endpoint.
    """
    must_conditions = [
        # Only retrieve points where allowed_agents contains this agent_id
        # OR the wildcard "*" (public knowledge).
        # MatchAny checks if the payload array contains any of the given values.
        FieldCondition(
            key="allowed_agents",
            match=MatchAny(any=[agent_id, "*"]),
        ),
        # Only retrieve current versions. Stale/superseded points are kept
        # for audit history but excluded from live queries.
        FieldCondition(key="is_current", match={"value": True}),
    ]

    # Optional namespace pre-filter: if the agent specifies a department,
    # we add an extra condition. This is a cheap exact-match filter that
    # dramatically reduces the vector search space before ANN kicks in.
    if namespace:
        must_conditions.append(
            FieldCondition(key="namespace", match={"value": namespace})
        )

    results = await client.search(
        collection_name=collection_name,
        query_vector=query_vector,
        query_filter=Filter(must=must_conditions),
        limit=top_k,
        with_payload=True,  # Return payload so we have the text, not just the ID
    )

    return [
        {
            "score": hit.score,
            "text": hit.payload["text"],
            "source": hit.payload.get("source_url"),
            "version": hit.payload.get("version"),
            "last_verified_at": hit.payload.get("last_verified_at"),
        }
        for hit in results
    ]

Knowledge Versioning and Staleness Detection

When source documents change, you must not simply overwrite the old embedding, you need an audit trail. The versioning pattern marks old points as non-current and inserts a new point with an incremented version.

flowchart LR
    SRC["Source Document\n(Confluence page)"]
    CDC["CDC Listener\n(Debezium)"]
    DIFF["Change\nDetector"]
    EMB["Embedding\nService"]
    QDRANT[("Qdrant")]
    OLD["Old Point\nis_current=False"]
    NEW["New Point\nis_current=True\nversion=N+1"]

    SRC -->|"page updated"| CDC
    CDC --> DIFF
    DIFF -->|"text diff > threshold"| EMB
    EMB -->|"new vector"| QDRANT
    QDRANT --> OLD
    QDRANT --> NEW

    style QDRANT fill:#EEF0F7,stroke:#6366F1,color:#0F172A
    style NEW fill:#CCFBF1,stroke:#0D9488,color:#0F172A
    style OLD fill:#FEF3C7,stroke:#F59E0B,color:#0F172A
    style EMB fill:#EEF0F7,stroke:#6366F1,color:#0F172A

import hashlib

async def update_knowledge_point(
    client: AsyncQdrantClient,
    collection_name: str,
    source_doc_id: str,
    new_text: str,
    new_embedding: list[float],
    agent_id: str,
    allowed_agents: list[str],
    namespace: Optional[str] = None,
) -> Optional[str]:
    """
    Version a knowledge update: mark old point stale, insert new point.
    Returns new point ID, or None if the text hasn't meaningfully changed.
    """
    # First, find all current points for this source document.
    existing, _ = await client.scroll(
        collection_name=collection_name,
        scroll_filter=Filter(
            must=[
                FieldCondition(key="source_doc_id", match={"value": source_doc_id}),
                FieldCondition(key="is_current", match={"value": True}),
            ]
        ),
        with_payload=True,
        limit=100,  # A document may be chunked into many points
    )

    if not existing:
        # Net-new document: just insert it
        return await upsert_knowledge(
            client, collection_name, new_embedding,
            KnowledgePayload(
                text=new_text,
                source_doc_id=source_doc_id,
                source_type="confluence",
                created_by_agent=agent_id,
                allowed_agents=allowed_agents,
                namespace=namespace,
            ),
        )

    # Hash-based staleness check: if the text hash matches, nothing changed.
    # This prevents re-embedding when a document is "saved" without real edits
    # embedding is expensive, so we skip it when unnecessary.
    new_hash = hashlib.sha256(new_text.encode()).hexdigest()
    old_text = existing[0].payload.get("text", "")
    old_hash = hashlib.sha256(old_text.encode()).hexdigest()

    if new_hash == old_hash:
        # Text unchanged: just bump last_verified_at so staleness timers reset
        await client.set_payload(
            collection_name=collection_name,
            payload={"last_verified_at": datetime.now(timezone.utc).isoformat()},
            points=[p.id for p in existing],
        )
        return None  # No new point needed

    # Mark all old points as non-current. We keep them for audit purposes
    # never hard-delete old versions unless you have a data retention policy
    # that explicitly requires it.
    old_version = existing[0].payload.get("version", 1)
    await client.set_payload(
        collection_name=collection_name,
        payload={"is_current": False},
        points=[p.id for p in existing],
    )

    # Insert new point with bumped version number
    return await upsert_knowledge(
        client, collection_name, new_embedding,
        KnowledgePayload(
            text=new_text,
            source_doc_id=source_doc_id,
            source_type="confluence",
            created_by_agent=agent_id,
            allowed_agents=allowed_agents,
            version=old_version + 1,
            namespace=namespace,
        ),
    )

Real-Time Updates via Change Data Capture

Polling source systems for changes is fragile and introduces latency. Change Data Capture (CDC) tools like Debezium stream database transaction logs and emit an event for every row insert, update, or delete in near real-time.

sequenceDiagram
    participant DB as Source DB (Postgres)
    participant DEB as Debezium CDC
    participant KAFKA as Kafka Topic
    participant WORKER as Embedding Worker
    participant QDRANT as Qdrant

    DB->>DEB: WAL log entry (row updated)
    DEB->>KAFKA: Emit change event {before, after, op}
    KAFKA->>WORKER: Consume event
    WORKER->>WORKER: Diff before vs after text
    alt text changed significantly
        WORKER->>WORKER: Embed new text
        WORKER->>QDRANT: Mark old point stale
        WORKER->>QDRANT: Insert new versioned point
    else no meaningful change
        WORKER->>QDRANT: Bump last_verified_at only
    end

    Note over WORKER,QDRANT: All agents immediately see\nthe updated knowledge

Knowledge Federation Across Departments

In large enterprises, different departments operate their own knowledge stores with different governance rules. A federation layer routes agent queries to the appropriate sub-store and merges results.

The key insight is that federation should be transparent to agents: they issue one query, the federation layer fans it out to the permitted sub-stores, and they receive a ranked merged result. Agents never need to know which sub-store answered.

Embedding Model Selection and Consistency

One of the most overlooked failure modes in shared knowledge bases is embedding model inconsistency: if Agent A writes a point using text-embedding-3-large and Agent B queries with voyage-4, the cosine similarity scores are meaningless because the vectors live in different geometric spaces.

The rule is simple: one collection, one embedding model, forever (or until you rebuild the entire index). Enforce this by:

1. Storing the model name and version in the collection’s metadata
2. Having your embedding service validate the model before any write
3. When you must migrate models, create a new collection, re-embed everything, and swap traffic atomically, never mix models in one collection

For most enterprise use cases, text-embedding-3-large (3072 dimensions, truncatable to 1536) from OpenAI or nomic-embed-text (self-hostable, 768 dimensions) covers the majority of semantic search quality requirements while remaining cost-effective at the scale of millions of chunks.