← Back to DevBytes
Developer Tutorial August 18, 2026 15 min read

Building a Multi-Tenant RAG Architecture with Pgvector

Introduction to Multi-Tenant RAG with Pgvector

Retrieval-Augmented Generation (RAG) has become the go-to pattern for building AI applications that ground large language models in private data. But when you're building a SaaS product that serves multiple customers, a new challenge emerges: how do you keep each tenant's data isolated while sharing the same infrastructure? This is where a multi-tenant RAG architecture comes in.

Pgvector, the PostgreSQL extension for vector similarity search, is an excellent foundation for this. Because it runs inside PostgreSQL, you get access to the full relational toolkit — schemas, row-level security, indexes, constraints, and transactions — alongside vector operations. This makes it uniquely suited for multi-tenant workloads where data isolation is non-negotiable.

In this tutorial, you'll build a complete multi-tenant RAG system from scratch. We'll cover schema design, tenant isolation strategies, document ingestion, embedding generation, and tenant-scoped retrieval. By the end, you'll have a production-ready architecture you can adapt to your own applications.

Why Multi-Tenancy Matters in RAG Systems

Imagine you're building a customer support assistant that lets companies query their own internal knowledge bases. Tenant A is a healthcare company. Tenant B is a financial services firm. If Tenant A's query accidentally retrieves documents from Tenant B's corpus, you have a catastrophic data breach on your hands.

A naive single-tenant approach — one database per customer — works for a handful of clients but breaks down at scale. You face operational overhead, connection pool exhaustion, and maintenance nightmares. A well-designed multi-tenant architecture lets you serve thousands of tenants from a single database while maintaining strict isolation guarantees.

Key Benefits

Prerequisites and Environment Setup

Before diving in, make sure you have the following:

Start by installing pgvector and creating your database:

-- Connect to PostgreSQL as a superuser
CREATE DATABASE rag_multitenant;

\c rag_multitenant

-- Install the pgvector extension
CREATE EXTENSION IF NOT EXISTS vector;

-- Verify installation
SELECT extversion FROM pg_extension WHERE extname = 'vector';

Now install the Python dependencies:

pip install psycopg[binary] openai numpy python-dotenv

Create a .env file to manage your secrets:

DATABASE_URL=postgresql://user:password@localhost:5432/rag_multitenant
OPENAI_API_KEY=sk-your-api-key-here

Database Schema Design for Multi-Tenancy

The schema is the backbone of your multi-tenant architecture. We'll use a shared-database, shared-schema approach where every table includes a tenant_id column. This is the most scalable pattern and works well with pgvector's indexing strategies.

Creating the Tables

-- Tenants table: stores metadata about each customer organization
CREATE TABLE tenants (
    id UUID PRIMARY KEY DEFAULT gen_random_uuid(),
    name VARCHAR(255) NOT NULL,
    slug VARCHAR(100) UNIQUE NOT NULL,
    plan VARCHAR(50) DEFAULT 'free',
    created_at TIMESTAMPTZ DEFAULT NOW(),
    updated_at TIMESTAMPTZ DEFAULT NOW()
);

-- Documents table: stores raw text and metadata per tenant
CREATE TABLE documents (
    id UUID PRIMARY KEY DEFAULT gen_random_uuid(),
    tenant_id UUID NOT NULL REFERENCES tenants(id) ON DELETE CASCADE,
    title VARCHAR(500) NOT NULL,
    content TEXT NOT NULL,
    source VARCHAR(255),
    metadata JSONB DEFAULT '{}'::jsonb,
    created_at TIMESTAMPTZ DEFAULT NOW(),
    updated_at TIMESTAMPTZ DEFAULT NOW()
);

-- Embeddings table: stores vector embeddings for each document chunk
CREATE TABLE document_chunks (
    id UUID PRIMARY KEY DEFAULT gen_random_uuid(),
    tenant_id UUID NOT NULL REFERENCES tenants(id) ON DELETE CASCADE,
    document_id UUID NOT NULL REFERENCES documents(id) ON DELETE CASCADE,
    chunk_index INTEGER NOT NULL,
    chunk_text TEXT NOT NULL,
    embedding vector(1536) NOT NULL,
    token_count INTEGER,
    created_at TIMESTAMPTZ DEFAULT NOW()
);

-- Indexes for efficient tenant-scoped queries
CREATE INDEX idx_documents_tenant ON documents(tenant_id);
CREATE INDEX idx_chunks_tenant ON document_chunks(tenant_id);
CREATE INDEX idx_chunks_document ON document_chunks(document_id);

-- Composite index for tenant + vector search
CREATE INDEX idx_chunks_tenant_embedding 
    ON document_chunks 
    USING hnsw (embedding vector_cosine_ops)
    WITH (m = 16, ef_construction = 64);

Notice the HNSW index on the embedding column. HNSW (Hierarchical Navigable Small World) graphs offer excellent recall-to-latency ratios for approximate nearest neighbor search. We'll discuss why this index choice matters in the best practices section.

Adding Row-Level Security

Row-Level Security (RLS) is your safety net. Even if application code has a bug that forgets to filter by tenant_id, the database itself will enforce isolation. This is defense in depth at its finest.

-- Enable RLS on all tenant-scoped tables
ALTER TABLE documents ENABLE ROW LEVEL SECURITY;
ALTER TABLE document_chunks ENABLE ROW LEVEL SECURITY;

-- Create a function to extract tenant_id from the current session
CREATE OR REPLACE FUNCTION current_tenant_id() 
    RETURNS UUID 
    AS $$
BEGIN
    RETURN NULLIF(current_setting('app.tenant_id', true), '')::UUID;
END;
$$ LANGUAGE plpgsql STABLE;

-- Policies: users can only see rows belonging to their session's tenant
CREATE POLICY tenant_isolation_documents ON documents
    USING (tenant_id = current_tenant_id())
    WITH CHECK (tenant_id = current_tenant_id());

CREATE POLICY tenant_isolation_chunks ON document_chunks
    USING (tenant_id = current_tenant_id())
    WITH CHECK (tenant_id = current_tenant_id());

-- Allow superusers to bypass RLS (for admin operations)
ALTER TABLE documents FORCE ROW LEVEL SECURITY;
ALTER TABLE document_chunks FORCE ROW LEVEL SECURITY;

Now, before any query, your application sets the tenant context:

SET LOCAL app.tenant_id = 'some-tenant-uuid-here';

This ensures every query within that transaction is automatically scoped to the correct tenant.

Building the RAG Pipeline

Now let's build the Python application layer. We'll create a clean, modular pipeline that handles tenant management, document ingestion, chunking, embedding, and retrieval.

Database Connection Manager

import os
import uuid
import psycopg
from contextlib import contextmanager
from dotenv import load_dotenv

load_dotenv()

DATABASE_URL = os.getenv("DATABASE_URL")

class DatabaseManager:
    def __init__(self, db_url: str):
        self.db_url = db_url

    @contextmanager
    def get_connection(self, tenant_id: str = None):
        """Get a database connection, optionally scoped to a tenant."""
        conn = psycopg.connect(self.db_url)
        try:
            if tenant_id:
                with conn.cursor() as cur:
                    cur.execute("SET LOCAL app.tenant_id = %s", (tenant_id,))
            yield conn
            conn.commit()
        except Exception:
            conn.rollback()
            raise
        finally:
            conn.close()

db = DatabaseManager(DATABASE_URL)

Tenant Management

class TenantManager:
    def __init__(self, database: DatabaseManager):
        self.db = database

    def create_tenant(self, name: str, slug: str, plan: str = "free") -> str:
        """Create a new tenant and return its UUID."""
        with self.db.get_connection() as conn:
            with conn.cursor() as cur:
                cur.execute(
                    """
                    INSERT INTO tenants (name, slug, plan)
                    VALUES (%s, %s, %s)
                    RETURNING id
                    """,
                    (name, slug, plan)
                )
                tenant_id = cur.fetchone()[0]
                return str(tenant_id)

    def get_tenant_by_slug(self, slug: str) -> dict:
        """Retrieve a tenant by its slug."""
        with self.db.get_connection() as conn:
            with conn.cursor() as cur:
                cur.execute(
                    "SELECT id, name, slug, plan FROM tenants WHERE slug = %s",
                    (slug,)
                )
                row = cur.fetchone()
                if row:
                    return {"id": str(row[0]), "name": row[1], "slug": row[2], "plan": row[3]}
                return None

tenant_manager = TenantManager(db)

Document Chunking Strategy

Chunking is critical for RAG quality. Chunks that are too large dilute relevance; chunks that are too small lose context. A common approach is token-based chunking with overlap.

import tiktoken

class TextChunker:
    def __init__(self, chunk_size: int = 512, chunk_overlap: int = 50):
        self.chunk_size = chunk_size
        self.chunk_overlap = chunk_overlap
        self.encoder = tiktoken.encoding_for_model("gpt-4")

    def chunk_text(self, text: str) -> list[dict]:
        """Split text into overlapping chunks based on token count."""
        tokens = self.encoder.encode(text)
        chunks = []
        
        start = 0
        chunk_index = 0
        
        while start < len(tokens):
            end = min(start + self.chunk_size, len(tokens))
            chunk_tokens = tokens[start:end]
            chunk_text = self.encoder.decode(chunk_tokens)
            
            chunks.append({
                "chunk_index": chunk_index,
                "chunk_text": chunk_text,
                "token_count": len(chunk_tokens)
            })
            
            chunk_index += 1
            start += (self.chunk_size - self.chunk_overlap)
        
        return chunks

chunker = TextChunker(chunk_size=512, chunk_overlap=50)

Embedding Generation

from openai import OpenAI
import os

client = OpenAI(api_key=os.getenv("OPENAI_API_KEY"))

class EmbeddingService:
    def __init__(self, model: str = "text-embedding-3-small"):
        self.model = model

    def embed_texts(self, texts: list[str]) -> list[list[float]]:
        """Generate embeddings for a list of texts."""
        response = client.embeddings.create(
            model=self.model,
            input=texts
        )
        return [item.embedding for item in response.data]

    def embed_query(self, query: str) -> list[float]:
        """Generate an embedding for a single query."""
        return self.embed_texts([query])[0]

embedding_service = EmbeddingService()

Document Ingestion Pipeline

Now let's tie it all together into an ingestion pipeline that takes a document, chunks it, embeds each chunk, and stores everything with proper tenant scoping.

class IngestionPipeline:
    def __init__(self, database, chunker, embedding_service):
        self.db = database
        self.chunker = chunker
        self.embedder = embedding_service

    def ingest_document(
        self, 
        tenant_id: str, 
        title: str, 
        content: str, 
        source: str = None,
        metadata: dict = None
    ) -> str:
        """Ingest a document: chunk it, embed chunks, and store everything."""
        
        # Step 1: Chunk the document
        chunks = self.chunker.chunk_text(content)
        
        # Step 2: Generate embeddings for all chunks in one batch
        chunk_texts = [c["chunk_text"] for c in chunks]
        embeddings = self.embedder.embed_texts(chunk_texts)
        
        # Step 3: Store document and chunks with tenant scoping
        with self.db.get_connection(tenant_id=tenant_id) as conn:
            with conn.cursor() as cur:
                # Insert the parent document
                cur.execute(
                    """
                    INSERT INTO documents (tenant_id, title, content, source, metadata)
                    VALUES (%s, %s, %s, %s, %s)
                    RETURNING id
                    """,
                    (tenant_id, title, content, source, 
                     psycopg.types.json.Json(metadata or {}))
                )
                document_id = cur.fetchone()[0]
                
                # Insert all chunks with their embeddings
                # Convert embeddings to pgvector format
                for chunk, embedding in zip(chunks, embeddings):
                    embedding_str = f"[{','.join(map(str, embedding))}]"
                    cur.execute(
                        """
                        INSERT INTO document_chunks 
                            (tenant_id, document_id, chunk_index, chunk_text, embedding, token_count)
                        VALUES (%s, %s, %s, %s, %s::vector, %s)
                        """,
                        (tenant_id, document_id, chunk["chunk_index"],
                         chunk["chunk_text"], embedding_str, chunk["token_count"])
                    )
        
        return str(document_id)

pipeline = IngestionPipeline(db, chunker, embedding_service)

Tenant-Scoped Retrieval

Retrieval is where multi-tenancy really matters. Every similarity search must be scoped to the requesting tenant. Thanks to RLS, even if you forget the WHERE tenant_id = clause, the database will filter results. But for performance, you should always include the tenant filter explicitly so the query planner can use the right indexes.

Building the Retrieval Service

class RetrievalService:
    def __init__(self, database, embedding_service):
        self.db = database
        self.embedder = embedding_service

    def retrieve(
        self,
        tenant_id: str,
        query: str,
        top_k: int = 5,
        similarity_threshold: float = 0.7
    ) -> list[dict]:
        """
        Retrieve the most relevant document chunks for a query,
        scoped to a specific tenant.
        """
        # Generate the query embedding
        query_embedding = self.embedder.embed_query(query)
        embedding_str = f"[{','.join(map(str, query_embedding))}]"
        
        with self.db.get_connection(tenant_id=tenant_id) as conn:
            with conn.cursor() as cur:
                cur.execute(
                    """
                    SELECT 
                        dc.id,
                        dc.chunk_text,
                        dc.chunk_index,
                        d.title,
                        d.source,
                        1 - (dc.embedding <=> %s::vector) AS similarity
                    FROM document_chunks dc
                    JOIN documents d ON dc.document_id = d.id
                    WHERE dc.tenant_id = %s
                      AND 1 - (dc.embedding <=> %s::vector) > %s
                    ORDER BY dc.embedding <=> %s::vector
                    LIMIT %s
                    """,
                    (embedding_str, tenant_id, embedding_str, 
                     similarity_threshold, top_k)
                )
                
                results = []
                for row in cur.fetchall():
                    results.append({
                        "chunk_id": str(row[0]),
                        "text": row[1],
                        "chunk_index": row[2],
                        "document_title": row[3],
                        "source": row[4],
                        "similarity": float(row[5])
                    })
                
                return results

retrieval = RetrievalService(db, embedding_service)

The <=> operator is pgvector's cosine distance operator. We convert it to a similarity score with 1 - distance. The query uses the HNSW index for fast approximate nearest neighbor search, and the tenant filter ensures we never cross tenant boundaries.

Putting It All Together: A Complete Example

def main():
    # 1. Create two tenants
    acme_id = tenant_manager.create_tenant("Acme Corp", "acme", "pro")
    globex_id = tenant_manager.create_tenant("Globex Inc", "globex", "enterprise")
    
    print(f"Created Acme Corp: {acme_id}")
    print(f"Created Globex Inc: {globex_id}")
    
    # 2. Ingest documents for Acme Corp
    pipeline.ingest_document(
        tenant_id=acme_id,
        title="Acme Employee Handbook",
        content="Acme Corp values transparency and collaboration. "
                "All employees receive 25 days of paid time off. "
                "Our headquarters is in Portland, Oregon. "
                "The dress code is business casual on weekdays.",
        source="hr_portal"
    )
    
    pipeline.ingest_document(
        tenant_id=acme_id,
        title="Acme Security Policy",
        content="All Acme employees must use two-factor authentication. "
                "Passwords must be at least 16 characters. "
                "Security training is mandatory quarterly.",
        source="security_wiki"
    )
    
    # 3. Ingest documents for Globex Inc
    pipeline.ingest_document(
        tenant_id=globex_id,
        title="Globex Employee Handbook",
        content="Globex Inc offers unlimited PTO and remote-first culture. "
                "Our headquarters is in Austin, Texas. "
                "Employees receive a $2,000 annual learning stipend.",
        source="hr_portal"
    )
    
    # 4. Query Acme's knowledge base
    print("\n--- Querying Acme Corp ---")
    acme_results = retrieval.retrieve(
        tenant_id=acme_id,
        query="How many vacation days do employees get?",
        top_k=3
    )
    for r in acme_results:
        print(f"  [{r['similarity']:.3f}] {r['document_title']}: {r['text'][:80]}...")
    
    # 5. Query Globex's knowledge base with the same question
    print("\n--- Querying Globex Inc ---")
    globex_results = retrieval.retrieve(
        tenant_id=globex_id,
        query="How many vacation days do employees get?",
        top_k=3
    )
    for r in globex_results:
        print(f"  [{r['similarity']:.3f}] {r['document_title']}: {r['text'][:80]}...")

if __name__ == "__main__":
    main()

When you run this, Acme's query returns Acme's handbook (25 days PTO), and Globex's query returns Globex's handbook (unlimited PTO). The same question, completely different answers, with zero cross-contamination.

Generating Context-Aware Responses

Retrieval is only half of RAG. The other half is feeding the retrieved context to an LLM to generate a grounded answer.

class RAGService:
    def __init__(self, retrieval_service, openai_client):
        self.retrieval = retrieval_service
        self.client = openai_client

    def answer(
        self,
        tenant_id: str,
        question: str,
        top_k: int = 5,
        model: str = "gpt-4o-mini"
    ) -> dict:
        """Generate a RAG answer scoped to a specific tenant."""
        
        # Step 1: Retrieve relevant chunks
        chunks = self.retrieval.retrieve(
            tenant_id=tenant_id,
            query=question,
            top_k=top_k
        )
        
        if not chunks:
            return {
                "answer": "I couldn't find any relevant information to answer your question.",
                "sources": []
            }
        
        # Step 2: Build the context from retrieved chunks
        context_parts = []
        sources = []
        for i, chunk in enumerate(chunks):
            context_parts.append(f"[Source {i+1}: {chunk['document_title']}]\n{chunk['text']}")
            sources.append({
                "title": chunk["document_title"],
                "source": chunk["source"],
                "similarity": chunk["similarity"]
            })
        
        context = "\n\n".join(context_parts)
        
        # Step 3: Generate the answer
        system_prompt = (
            "You are a helpful assistant. Answer the user's question based "
            "only on the provided context. If the context doesn't contain "
            "enough information, say so clearly. Always cite which source "
            "your answer comes from."
        )
        
        user_prompt = f"Context:\n{context}\n\nQuestion: {question}"
        
        response = self.client.chat.completions.create(
            model=model,
            messages=[
                {"role": "system", "content": system_prompt},
                {"role": "user", "content": user_prompt}
            ],
            temperature=0.1
        )
        
        return {
            "answer": response.choices[0].message.content,
            "sources": sources
        }

rag = RAGService(retrieval, client)

Usage is straightforward:

result = rag.answer(
    tenant_id=acme_id,
    question="What is the dress code at Acme?"
)

print(f"Answer: {result['answer']}")
print(f"Sources: {[s['title'] for s in result['sources']]}")

Best Practices for Production

1. Choose the Right Index Type

Pgvector offers several index types. For most multi-tenant RAG workloads, HNSW is the best default choice. It provides excellent recall with sub-millisecond query latency on datasets up to millions of vectors. However, if memory is constrained, consider IVFFlat, which uses less memory but requires careful tuning of the lists parameter.

-- HNSW index (recommended default)
CREATE INDEX idx_chunks_hnsw ON document_chunks
    USING hnsw (embedding vector_cosine_ops)
    WITH (m = 16, ef_construction = 64);

-- IVFFlat index (lower memory, needs tuning)
CREATE INDEX idx_chunks_ivf ON document_chunks
    USING ivfflat (embedding vector_cosine_ops)
    WITH (lists = 100);  -- Set to sqrt(row_count) as a starting point

2. Always Filter by Tenant_id in Queries

Even with RLS enabled, always include WHERE tenant_id = %s in your queries. RLS is a safety net, not a performance optimization. The query planner can use the tenant_id to prune index partitions and avoid scanning irrelevant data.

3. Use Connection Pooling with Tenant Context

In production, use a connection pooler like PgBouncer or psycopg's connection pool. Be careful with SET LOCAL — it only works within a transaction. If you're using connection pooling with transaction-level pooling mode, ensure each transaction sets the tenant context at the start.

from psycopg_pool import ConnectionPool

pool = ConnectionPool(
    DATABASE_URL,
    min_size=5,
    max_size=20,
    configure=pool_configure
)

def pool_configure(conn):
    """Configure each connection in the pool."""
    conn.execute("SET statement_timeout = '30s'")
    conn.execute("SET idle_in_transaction_session_timeout = '60s'")

4. Implement Tenant-Aware Rate Limiting

Different tenants may have different usage patterns and plan tiers. Implement rate limiting at the application layer to prevent any single tenant from monopolizing resources.

import time
from collections import defaultdict

class TenantRateLimiter:
    def __init__(self, max_requests_per_minute: int = 60):
        self.limits = defaultdict(list)
        self.max_per_minute = max_requests_per_minute

    def check(self, tenant_id: str) -> bool:
        now = time.time()
        window = 60  # 1 minute
        
        # Clean old entries
        self.limits[tenant_id] = [
            t for t in self.limits[tenant_id] if now - t < window
        ]
        
        if len(self.limits[tenant_id]) >= self.max_per_minute:
            return False
        
        self.limits[tenant_id].append(now)
        return True

rate_limiter = TenantRateLimiter(max_requests_per_minute=60)

5. Monitor Per-Tenant Metrics

Track query latency, retrieval quality, and token usage per tenant. This helps you identify noisy neighbors, plan capacity, and potentially offer tiered service levels.

-- Query to find tenants with the most documents
SELECT t.name, COUNT(d.id) as doc_count, 
       COUNT(dc.id) as chunk_count
FROM tenants t
LEFT JOIN documents d ON d.tenant_id = t.id
LEFT JOIN document_chunks dc ON dc.tenant_id = t.id
GROUP BY t.name
ORDER BY doc_count DESC;

-- Average similarity scores per tenant (retrieval quality indicator)
SELECT t.name, AVG(1 - (dc.embedding <=> dc.embedding)) as avg_sim
FROM tenants t
JOIN document_chunks dc ON dc.tenant_id = t.id
GROUP BY t.name;

6. Handle Tenant Data Deletion Gracefully

When a tenant churns, you need to delete their data completely. The ON DELETE CASCADE foreign keys we set up make this a single operation:

def delete_tenant(tenant_id: str):
    """Delete a tenant and all associated data."""
    with db.get_connection() as conn:
        with conn.cursor() as cur:
            # RLS might block this, so temporarily disable for admin ops
            cur.execute("SET LOCAL app.tenant_id = ''")
            cur.execute("DELETE FROM tenants WHERE id = %s", (tenant_id,))
            # CASCADE will automatically remove documents and chunks

7. Re-embed When Models Change

Embedding models improve over time. When you switch models, all existing embeddings become incompatible. Store the model name and version alongside embeddings so you can identify and re-embed stale vectors.

ALTER TABLE document_chunks 
    ADD COLUMN embedding_model VARCHAR(100) DEFAULT 'text-embedding-3-small';

-- Find chunks that need re-embedding
SELECT tenant_id, COUNT(*) as stale_count
FROM document_chunks
WHERE embedding_model != 'text-embedding-3-large'
GROUP BY tenant_id;

8. Use Partial Indexes for Large-Scale Deployments

If you have thousands of tenants and millions of chunks, consider partitioning the document_chunks table by tenant or using partial indexes for your most active tenants.

-- Partial index for a high-volume tenant
CREATE INDEX idx_chunks_acme ON document_chunks
    USING hnsw (embedding vector_cosine_ops)
    WHERE tenant_id = 'acme-corp-uuid-here';

Conclusion

Building a multi-tenant RAG architecture with pgvector gives you the best of both worlds: the proven reliability and rich feature set of PostgreSQL combined with high-performance vector similarity search. By designing your schema with tenant isolation from the start, leveraging row-level security as a safety net, and following the best practices outlined in this tutorial, you can build a system that scales to thousands of tenants while maintaining strict data boundaries. The key takeaways are to always scope queries by tenant_id for both security and performance, choose the right index type for your workload, implement proper connection pooling with tenant context, and monitor per-tenant metrics to catch issues early. As your system grows, consider table partitioning and partial indexes to keep query latency low. With this architecture in place, you're well-equipped to deliver AI-powered experiences to multiple customers from a single, well-governed database.

— Ad —

Google AdSense will appear here after approval

← Back to all articles