Vector DB, Part 3: Multi-Tenant Operations and Embedding Drift Monitoring

# 1: Row-Level Security ( — )
# pgvector + Supabase RLS
"""
ALTER TABLE documents ENABLE ROW LEVEL SECURITY;
CREATE POLICY tenant_isolation ON documents
    USING (tenant_id = current_setting('app.tenant_id')::uuid);
"""

# 2:
async def get_tenant_client(tenant_id: str):
    conn = await asyncpg.connect(DATABASE_URL)
    await conn.execute(f"SET search_path = tenant_{tenant_id}")
    return conn

# 3:
async def tenant_search(query_emb, tenant_id: str, k: int = 10):
    return await index.query(
        vector=query_emb,
        filter={"tenant_id": {"$eq": tenant_id}},
        top_k=k
    )

import asyncio
from datetime import datetime

class IncrementalIndexer:
    async def process_change_stream(self):
 """ when """
        async for event in self.db.change_stream("documents"):
            if event["operation"] == "INSERT":
                await self.add_document(event["document"])
            elif event["operation"] == "UPDATE":
                await self.update_document(event["document"])
            elif event["operation"] == "DELETE":
                await self.remove_document(event["document_id"])

    async def add_document(self, doc: dict):
        emb = await self.embed(doc["content"])
        await self.vector_db.upsert(
            id=doc["id"],
            vector=emb,
            metadata={
                "title": doc["title"],
                "tenant_id": doc["tenant_id"],
                "updated_at": datetime.utcnow().isoformat(),
            }
        )

    async def remove_document(self, doc_id: str):
        await self.vector_db.delete(ids=[doc_id])
 # : deleted=True

import numpy as np
from scipy.spatial.distance import cosine

class DriftMonitor:
    def __init__(self, baseline_sample: np.ndarray):
        self.baseline_centroid = baseline_sample.mean(axis=0)
        self.baseline_std = baseline_sample.std(axis=0)

    def check_drift(self, new_embeddings: np.ndarray) -> dict:
        new_centroid = new_embeddings.mean(axis=0)

        centroid_shift = cosine(self.baseline_centroid, new_centroid)
        std_ratio = new_embeddings.std(axis=0).mean() / self.baseline_std.mean()

        return {
            "centroid_cosine_shift": centroid_shift,
            "std_ratio": std_ratio,
            "drift_detected": centroid_shift > 0.05 or not (0.8 < std_ratio < 1.2),
            "recommendation": self._recommend(centroid_shift, std_ratio),
        }

    def _recommend(self, shift: float, ratio: float) -> str:
        if shift > 0.1:
 return " — "
        if shift > 0.05:
 return " after "
 return " "

# when
METRICS = {
 #
 "recall_at_10": "golden set 10 ",
 "mrr": " (Mean Reciprocal Rank)",

 #
 "p50_latency_ms": " ",
 "p99_latency_ms": " (SLA )",

 #
 "index_freshness_min": " after ",
 "embedding_cache_hit_rate": "when (: 40% )",
 "tenant_isolation_violations": " (0 )",
}

async def collect_metrics() -> dict:
    results = {}
    results["p50_latency_ms"] = await measure_percentile(queries=GOLDEN_QUERIES, p=50)
    results["p99_latency_ms"] = await measure_percentile(queries=GOLDEN_QUERIES, p=99)
    results["recall_at_10"] = await evaluate_recall(GOLDEN_SET, k=10)
    results["embedding_cache_hit_rate"] = cache.hit_rate()
    return results