Enterprise RAG: Building Knowledge Systems with Vector Databases

Enterprise RAG (Retrieval Augmented Generation) is fundamentally different from the tutorial-grade RAG demos you see online. Enterprise data is messy, unstructured, spread across dozens of systems, and governed by strict access controls. Building a RAG system that handles this complexity while remaining fast and accurate requires careful architectural decisions.

This guide focuses on the vector database and knowledge system layer of enterprise RAG — the foundation that determines whether your RAG system actually works.

Choosing a Vector Database

The vector database is the most critical infrastructure decision for your RAG system. Here's what matters:

Qdrant

Our most recommended choice for production RAG:

from qdrant_client import QdrantClient, models

client = QdrantClient(host="localhost", port=6333)

# Create collection with optimized settings
client.create_collection(
    collection_name="enterprise_docs",
    vectors_config=models.VectorParams(
        size=1024,
        distance=models.Distance.COSINE,
        on_disk=True,           # For large collections
    ),
    optimizers_config=models.OptimizersConfigDiff(
        indexing_threshold=20000,
    ),
    hnsw_config=models.HnswConfigDiff(
        m=16,
        ef_construct=100,
    )
)

Why Qdrant:

• Excellent filtering performance (critical for enterprise access controls)
• Rust-based, memory-efficient
• Native hybrid search (dense + sparse vectors)
• Payload indexing for metadata filtering
• Snapshots and replication for production reliability

Weaviate

Best for teams that need multi-modal search and a GraphQL API:

{
  Get {
    Document(
      nearText: { concepts: ["quarterly revenue analysis"] }
      where: {
        operator: And
        operands: [
          { path: ["department"], operator: Equal, valueText: "finance" }
          { path: ["year"], operator: GreaterThan, valueInt: 2024 }
        ]
      }
    ) {
      title
      content
      _additional { distance certainty }
    }
  }
}

Why Weaviate:

• Multi-modal (text, images, video embeddings)
• Built-in vectorization modules
• GraphQL API for complex queries
• Strong enterprise features (multi-tenancy, RBAC)

pgvector (PostgreSQL Extension)

Best for teams already invested in PostgreSQL:

-- Create table with vector column
CREATE TABLE documents (
    id SERIAL PRIMARY KEY,
    content TEXT,
    embedding vector(1024),
    metadata JSONB,
    department TEXT,
    created_at TIMESTAMP DEFAULT NOW()
);

-- Create HNSW index for fast similarity search
CREATE INDEX ON documents
USING hnsw (embedding vector_cosine_ops)
WITH (m = 16, ef_construction = 100);

-- Hybrid query with metadata filtering
SELECT content, metadata,
       1 - (embedding <=> $1::vector) as similarity
FROM documents
WHERE department = 'engineering'
  AND created_at > '2025-01-01'
ORDER BY embedding <=> $1::vector
LIMIT 10;

Why pgvector:

• No new infrastructure — runs in your existing PostgreSQL
• ACID transactions for data consistency
• Full SQL for complex filtering and joins
• Mature ecosystem (backups, monitoring, replication)

Decision Matrix

Enterprise Chunking Strategies

Enterprise documents require specialized chunking approaches:

Hierarchical Chunking

Store documents at multiple granularity levels:

class HierarchicalChunker:
    def chunk(self, document):
        chunks = []

        # Level 1: Full sections (for broad context)
        for section in document.sections:
            chunks.append(Chunk(
                text=section.full_text,
                level="section",
                parent_id=document.id,
                metadata=section.metadata
            ))

            # Level 2: Paragraphs (for precise retrieval)
            for para in section.paragraphs:
                chunks.append(Chunk(
                    text=para.text,
                    level="paragraph",
                    parent_id=section.id,
                    metadata=para.metadata
                ))

        return chunks

Retrieve at the paragraph level, but return the full section for context. This gives you the best of both worlds: precise retrieval and comprehensive answers.

Table-Aware Chunking

Enterprise documents are full of tables. Most chunkers destroy table structure:

class TableAwareChunker:
    def chunk(self, document):
        chunks = []
        for element in document.elements:
            if element.type == "table":
                # Keep table intact as one chunk
                chunks.append(Chunk(
                    text=element.to_markdown(),
                    type="table",
                    metadata={"headers": element.headers}
                ))
            elif element.type == "text":
                # Normal text chunking
                chunks.extend(self.text_chunker.chunk(element.text))
        return chunks

Access Control-Aware Chunking

Enterprise RAG must respect document permissions:

class SecureChunk:
    text: str
    embedding: list[float]
    metadata: dict
    access_control: AccessControl  # Who can see this chunk

class AccessControl:
    departments: list[str]         # e.g., ["engineering", "finance"]
    roles: list[str]               # e.g., ["manager", "admin"]
    classification: str            # e.g., "confidential", "internal", "public"

class SecureRetriever:
    async def retrieve(self, query, user):
        # Filter by user's access permissions BEFORE vector search
        results = await self.vector_store.search(
            query=query,
            filter={
                "should": [
                    {"key": "departments", "match": {"any": user.departments}},
                    {"key": "classification", "match": {"value": "public"}}
                ]
            }
        )
        return results

Embedding Model Strategy

Multi-Model Embedding

Different content types benefit from different embedding models:

class MultiModelEmbedder:
    def __init__(self):
        self.text_model = SentenceTransformer("BAAI/bge-large-en-v1.5")
        self.code_model = SentenceTransformer("jinaai/jina-embeddings-v2-base-code")
        self.multilingual_model = SentenceTransformer("intfloat/multilingual-e5-large")

    def embed(self, chunk):
        if chunk.type == "code":
            return self.code_model.encode(chunk.text)
        elif chunk.language != "en":
            return self.multilingual_model.encode(chunk.text)
        else:
            return self.text_model.encode(chunk.text)

Embedding Pipeline

class EmbeddingPipeline:
    async def process(self, documents):
        chunks = self.chunker.chunk_all(documents)

        # Batch embed for efficiency
        batch_size = 64
        for i in range(0, len(chunks), batch_size):
            batch = chunks[i:i+batch_size]
            embeddings = self.embedder.encode_batch(
                [c.text for c in batch]
            )

            # Upsert to vector store
            await self.vector_store.upsert(
                points=[
                    Point(
                        id=chunk.id,
                        vector=embedding,
                        payload=chunk.metadata
                    )
                    for chunk, embedding in zip(batch, embeddings)
                ]
            )

Production Patterns

Incremental Indexing

Don't re-embed your entire corpus when one document changes:

class IncrementalIndexer:
    async def sync(self, data_source):
        changes = await data_source.get_changes_since(self.last_sync)

        for change in changes:
            if change.type == "created" or change.type == "modified":
                chunks = self.chunker.chunk(change.document)
                await self.index_chunks(chunks)
            elif change.type == "deleted":
                await self.vector_store.delete(
                    filter={"document_id": change.document_id}
                )

        self.last_sync = datetime.utcnow()

Query Routing

Not every query needs vector search. Route simple queries to simpler systems:

class QueryRouter:
    async def route(self, query):
        query_type = await self.classify_query(query)

        if query_type == "exact_lookup":
            return await self.keyword_search(query)
        elif query_type == "semantic":
            return await self.hybrid_search(query)
        elif query_type == "analytical":
            return await self.sql_query(query)
        elif query_type == "multi_hop":
            return await self.iterative_retrieval(query)

Caching

Cache frequent queries and their results:

class SemanticCache:
    def __init__(self, vector_store, similarity_threshold=0.95):
        self.cache_store = vector_store
        self.threshold = similarity_threshold

    async def get(self, query):
        results = await self.cache_store.search(query, top_k=1)
        if results and results[0].score > self.threshold:
            return results[0].payload["response"]
        return None

    async def set(self, query, response):
        await self.cache_store.upsert(query, {"response": response})

Getting Started

Enterprise RAG is a complex but high-value investment. The key is starting with a clear use case, choosing the right vector database for your requirements, and building robust data pipelines that handle the messiness of real enterprise data.

Contact Cloudpush to discuss your enterprise RAG architecture and get expert guidance on implementation.