How to design a RAG pipeline for enterprise documents

Building a retrieval-augmented generation system that works on enterprise documents requires more than stitching together a vector store and an LLM. You need to handle messy PDFs, respect access controls, optimize for latency, and measure whether answers are actually correct. This guide walks through how to design a RAG pipeline from ingestion to evaluation, with code you can run and verification checkpoints at each stage.

Step 1: Define your document corpus and access model

Before writing ingestion code, enumerate what you’re indexing. Enterprise corpora typically include PDFs (contracts, reports), Office files, Confluence/Notion pages, email archives, and code repositories. Each source has different structure, update frequency, and permission semantics.

Create a manifest that captures source, refresh cadence, and access tier:

# corpus_manifest.yaml
sources:
  - id: legal-contracts
    type: sharepoint
    path: /sites/legal/contracts
    refresh: daily
    access_tier: restricted  # requires legal group membership
    parser: pdf_plumber
  - id: engineering-rfcs
    type: confluence
    space: ENG
    refresh: hourly
    access_tier: internal
    parser: confluence_html
  - id: customer-emails
    type: gmail
    query: "label:support after:2023/01/01"
    refresh: 15min
    access_tier: pii_restricted
    parser: mime

Verify: Run a dry-run inventory script that counts documents per source and confirms your service account can read each location. Log any permission errors before they surface in production.

Step 2: Build idempotent ingestion with content hashing

Enterprise documents change. Re-ingesting everything on every run wastes compute and breaks citation stability. Use content-addressable storage: hash each document, store the hash, and only process deltas.

# ingestion/hasher.py
import hashlib
import xxhash  # faster than hashlib for large files

def content_hash(bytes_: bytes) -> str:
    return xxhash.xxh64(bytes_).hexdigest()

def file_hash(path: str) -> str:
    with open(path, "rb") as f:
        return content_hash(f.read())

Pair this with a lightweight metadata store (SQLite, Postgres, or DynamoDB) tracking doc_id, source_id, content_hash, last_modified, access_tier, and chunk_ids.

# ingestion/state.py
import sqlite3
from dataclasses import dataclass
from typing import Optional

@dataclass
class DocRecord:
    doc_id: str
    source_id: str
    content_hash: str
    last_modified: float
    access_tier: str
    chunk_ids: list[str]

class IngestionState:
    def __init__(self, db_path: str = "ingestion.db"):
        self.conn = sqlite3.connect(db_path)
        self._init_schema()

    def _init_schema(self):
        self.conn.execute("""
            CREATE TABLE IF NOT EXISTS docs (
                doc_id TEXT PRIMARY KEY,
                source_id TEXT NOT NULL,
                content_hash TEXT NOT NULL,
                last_modified REAL NOT NULL,
                access_tier TEXT NOT NULL,
                chunk_ids TEXT NOT NULL  -- JSON array
            )
        """)
        self.conn.execute("CREATE INDEX IF NOT EXISTS idx_source ON docs(source_id)")

    def get_existing(self, doc_id: str) -> Optional[DocRecord]:
        cur = self.conn.execute(
            "SELECT * FROM docs WHERE doc_id = ?", (doc_id,)
        )
        row = cur.fetchone()
        if not row:
            return None
        return DocRecord(
            doc_id=row[0], source_id=row[1], content_hash=row[2],
            last_modified=row[3], access_tier=row[4],
            chunk_ids=json.loads(row[5])
        )

    def upsert(self, record: DocRecord):
        self.conn.execute("""
            INSERT INTO docs VALUES (?, ?, ?, ?, ?, ?)
            ON CONFLICT(doc_id) DO UPDATE SET
                content_hash=excluded.content_hash,
                last_modified=excluded.last_modified,
                access_tier=excluded.access_tier,
                chunk_ids=excluded.chunk_ids
        """, (record.doc_id, record.source_id, record.content_hash,
              record.last_modified, record.access_tier,
              json.dumps(record.chunk_ids)))
        self.conn.commit()

Verify: Ingest a test corpus twice. The second run should process zero new chunks. Confirm chunk_ids remain stable for unchanged documents.

Step 3: Choose a chunking strategy that preserves semantic boundaries

Naive fixed-size chunking splits tables, code blocks, and legal clauses mid-sentence. For enterprise docs, use structure-aware chunking:

• PDFs: Extract with pdfplumber or marker, preserve table markdown, headings, and page numbers
• HTML/Confluence: Parse DOM, chunk by heading hierarchy (h1 → h2 → p)
• Code: Use tree-sitter to split by function/class definitions
• Email: Separate headers, quoted threads, and signatures

# chunking/strategies.py
from abc import ABC, abstractmethod
from dataclasses import dataclass
from typing import Iterator
import re

@dataclass
class Chunk:
    text: str
    metadata: dict  # doc_id, page, section, heading_path, char_start, char_end

class Chunker(ABC):
    @abstractmethod
    def chunk(self, doc_id: str, content: str, metadata: dict) -> Iterator[Chunk]:
        pass

class HeadingAwareChunker(Chunker):
    """Chunk by markdown heading hierarchy, targeting ~512 tokens."""
    def __init__(self, target_tokens: int = 512, overlap_tokens: int = 50):
        self.target_tokens = target_tokens
        self.overlap_tokens = overlap_tokens

    def chunk(self, doc_id: str, content: str, metadata: dict) -> Iterator[Chunk]:
        # Split on headings, keep heading path in metadata
        sections = re.split(r'\n(#{1,6}\s+.+)', content)
        # sections alternates: text, heading, text, heading...
        heading_path = []
        current_text = ""
        
        for i, segment in enumerate(sections):
            if re.match(r'^#{1,6}\s+', segment):
                # Emit previous section
                if current_text.strip():
                    yield from self._emit_chunks(doc_id, current_text, metadata, heading_path)
                heading_path.append(segment.strip())
                current_text = ""
            else:
                current_text += segment
        
        if current_text.strip():
            yield from self._emit_chunks(doc_id, current_text, metadata, heading_path)

    def _emit_chunks(self, doc_id: str, text: str, metadata: dict, heading_path: list) -> Iterator[Chunk]:
        # Simple token approximation: ~4 chars per token
        target_chars = self.target_tokens * 4
        overlap_chars = self.overlap_tokens * 4
        
        for i in range(0, len(text), target_chars - overlap_chars):
            chunk_text = text[i:i + target_chars]
            yield Chunk(
                text=chunk_text,
                metadata={
                    **metadata,
                    "doc_id": doc_id,
                    "heading_path": " > ".join(heading_path),
                    "char_start": i,
                    "char_end": min(i + target_chars, len(text)),
                }
            )

Verify: Spot-check 20 chunks from each document type. Confirm tables stay intact, headings propagate as metadata, and no chunk exceeds your embedding model’s context window.

Step 4: Generate embeddings with batching and retry logic

Embedding throughput matters at scale. Batch requests, handle rate limits, and store vectors with the same chunk_id used in your ingestion state.

# embeddings/client.py
import asyncio
import httpx
from tenacity import retry, stop_after_attempt, wait_exponential_jitter
from typing import List
import numpy as np

class EmbeddingClient:
    def __init__(
        self,
        base_url: str = "https://api.openai.com/v1",
        api_key: str = None,
        model: str = "text-embedding-3-large",
        batch_size: int = 128,
        max_concurrent: int = 4,
    ):
        self.base_url = base_url
        self.headers = {"Authorization": f"Bearer {api_key}"}
        self.model = model
        self.batch_size = batch_size
        self.semaphore = asyncio.Semaphore(max_concurrent)
        self.client = httpx.AsyncClient(timeout=60.0)

    @retry(
        wait=wait_exponential_jitter(initial=1, max=30),
        stop=stop_after_attempt(3),
    )
    async def _embed_batch(self, texts: List[str]) -> List[List[float]]:
        async with self.semaphore:
            resp = await self.client.post(
                f"{self.base_url}/embeddings",
                headers=self.headers,
                json={"model": self.model, "input": texts},
            )
            resp.raise_for_status()
            data = resp.json()["data"]
            # API returns in order, but sort by index to be safe
            data.sort(key=lambda x: x["index"])
            return [d["embedding"] for d in data]

    async def embed_all(self, texts: List[str]) -> np.ndarray:
        """Embed texts in batches, return (n, dim) float32 array."""
        all_embeddings = []
        for i in range(0, len(texts), self.batch_size):
            batch = texts[i:i + self.batch_size]
            embeddings = await self._embed_batch(batch)
            all_embeddings.extend(embeddings)
        return np.array(all_embeddings, dtype=np.float32)

Verify: Embed a known test set (e.g., 100 chunks from your corpus). Confirm output shape matches (n_chunks, 3072) for text-embedding-3-large. Compute cosine similarity between duplicate chunks — it should be > 0.99.

Step 5: Index into a vector store with metadata filtering

Enterprise RAG requires filtering by access_tier, source_id, and date ranges at query time. Choose a vector store that supports metadata filtering natively (Pinecone, Weaviate, Qdrant, pgvector). Avoid post-filtering — it breaks top-k guarantees.

# vectorstore/qdrant_client.py
from qdrant_client import QdrantClient, models
from qdrant_client.http.models import Filter, FieldCondition, MatchValue, Range
import numpy as np
from typing import List, Optional
from dataclasses import dataclass

@dataclass
class SearchResult:
    chunk_id: str
    score: float
    text: str
    metadata: dict

class QdrantVectorStore:
    def __init__(
        self,
        url: str,
        api_key: str,
        collection_name: str = "enterprise_rag",
        vector_size: int = 3072,
    ):
        self.client = QdrantClient(url=url, api_key=api_key)
        self.collection = collection_name
        self._ensure_collection(vector_size)

    def _ensure_collection(self, vector_size: int):
        if not self.client.collection_exists(self.collection):
            self.client.create_collection(
                collection_name=self.collection,
                vectors_config=models.VectorParams(
                    size=vector_size,
                    distance=models.Distance.COSINE,
                ),
            )
            # Create payload indexes for filtering
            for field in ["access_tier", "source_id", "doc_id", "heading_path"]:
                self.client.create_payload_index(
                    collection_name=self.collection,
                    field_name=field,
                    field_schema=models.PayloadSchemaType.KEYWORD,
                )
            self.client.create_payload_index(
                collection_name=self.collection,
                field_name="last_modified",
                field_schema=models.PayloadSchemaType.FLOAT,
            )

    def upsert_chunks(self, chunks: List[Chunk], embeddings: np.ndarray):
        points = []
        for chunk, vec in zip(chunks, embeddings):
            points.append(models.PointStruct(
                id=chunk.metadata["chunk_id"],
                vector=vec.tolist(),
                payload={
                    "text": chunk.text,
                    **chunk.metadata,
                },
            ))
        self.client.upsert(collection_name=self.collection, points=points)

    def search(
        self,
        query_vector: np.ndarray,
        top_k: int = 10,
        access_tiers: Optional[List[str]] = None,
        source_ids: Optional[List[str]] = None,
        date_range: Optional[tuple] = None,  # (start_ts, end_ts)
    ) -> List[SearchResult]:
        must = []
        if access_tiers:
            must.append(FieldCondition(key="access_tier", match=MatchValue(any=access_tiers)))
        if source_ids:
            must.append(FieldCondition(key="source_id", match=MatchValue(any=source_ids)))
        if date_range:
            must.append(FieldCondition(
                key="last_modified",
                range=Range(gte=date_range[0], lte=date_range[1]),
            ))

        query_filter = Filter(must=must) if must else None

        hits = self.client.search(
            collection_name=self.collection,
            query_vector=query_vector.tolist(),
            query_filter=query_filter,
            limit=top_k,
            with_payload=True,
        )
        return [
            SearchResult(
                chunk_id=hit.id,
                score=hit.score,
                text=hit.payload.pop("text"),
                metadata=hit.payload,
            )
            for hit in hits
        ]

Verify: Insert 1,000 test chunks with varied access_tier values. Query with a filter for one tier — confirm zero results from other tiers. Measure p99 latency at your target QPS.

Step 6: Implement hybrid retrieval (vector + keyword)

Pure vector search misses exact matches (product codes, error numbers, proper names). Combine with BM25 or a sparse vector model like SPLADE. Reciprocal rank fusion (RRF) merges results without score calibration.

# retrieval/hybrid.py
from rank_bm25 import BM25Okapi
from typing import List, Dict
import numpy as np

class HybridRetriever:
    def __init__(
        self,
        vector_store: QdrantVectorStore,
        embedding_client: EmbeddingClient,
        bm25_corpus: List[str],  # chunk texts for BM25
        chunk_id_to_text: Dict[str, str],
        k: int = 60,  # RRF parameter
    ):
        self.vector_store = vector_store
        self.embedding_client = embedding_client
        self.bm25 = BM25Okapi([doc.split() for doc in bm25_corpus])
        self.chunk_id_to_text = chunk_id_to_text
        self.k = k

    def retrieve(
        self,
        query: str,
        top_k: int = 10,
        access_tiers: List[str] = None,
        source_ids: List[str] = None,
        vector_weight: float = 0.7,
    ) -> List[SearchResult]:
        # Vector search
        query_vec = await self.embedding_client.embed_all([query])
        vector_results = self.vector_store.search(
            query_vec[0], top_k=top_k * 2,  # fetch more for fusion
            access_tiers=access_tiers, source_ids=source_ids,
        )

        # BM25 search (filter post-retrieval since BM25 doesn't support metadata filters)
        bm25_scores = self.bm25.get_scores(query.split())
        top_bm25_indices = np.argsort(bm25_scores)[::-1][:top_k * 2]
        bm25_results = [
            SearchResult(
                chunk_id=list(self.chunk_id_to_text.keys())[i],
                score=float(bm25_scores[i]),
                text=self.chunk_id_to_text[list(self.chunk_id_to_text.keys())[i]],
                metadata={},  # would need separate metadata lookup
            )
            for i in top_bm25_indices if bm25_scores[i] > 0
        ]

        # Reciprocal rank fusion
        return self._rrf_fuse(vector_results, bm25_results, top_k)

    def _rrf_fuse(
        self,
        vector_results: List[SearchResult],
        bm25_results: List[SearchResult],
        top_k: int,
    ) -> List[SearchResult]:
        scores = {}
        for rank, result in enumerate(vector_results):
            scores.setdefault(result.chunk_id, 0.0)
            scores[result.chunk_id] += 1.0 / (self.k + rank + 1)
        for rank, result in enumerate(bm25_results):
            scores.setdefault(result.chunk_id, 0.0)
            scores[result.chunk_id] += 1.0 / (self.k + rank + 1)

        # Reconstruct full results
        all_chunks = {r.chunk_id: r for r in vector_results + bm25_results}
        fused = sorted(
            [(chunk_id, score) for chunk_id, score in scores.items()],
            key=lambda x: x[1],
            reverse=True,
        )[:top_k]

        return [all_chunks[chunk_id] for chunk_id, _ in fused]

Verify: Create a test set of 50 queries with known relevant chunks (including exact-match keywords). Measure recall@10 for vector-only, BM25-only, and hybrid. Hybrid should win on both semantic and keyword queries.

Step 7: Build a reranker stage for precision

Cross-encoder rerankers (e.g., bge-reranker-v2-m3, cohere-rerank-3) score query-chunk pairs with full attention. Run on the top 50-100 hybrid results, keep top 5-10 for the LLM.

# retrieval/reranker.py
import httpx
from typing import List
import asyncio

class CohereReranker:
    def __init__(self, api_key: str, model: str = "rerank-english-v3.0", top_n: int = 10):
        self.api_key = api_key
        self.model = model
        self.top_n = top_n
        self.client = httpx.AsyncClient(timeout=30.0)

    async def rerank(
        self,
        query: str,
        documents: List[str],
        chunk_ids: List[str],
    ) -> List[tuple[str, float]]:  # (chunk_id, relevance_score)
        resp = await self.client.post(
            "https://api.cohere.ai/v1/rerank",
            headers={"Authorization": f"Bearer {self.api_key}"},
            json={
                "model": self.model,
                "query": query,
                "documents": documents,
                "top_n": self.top_n,
                "return_documents": False,
            },
        )
        resp.raise_for_status()
        results = resp.json()["results"]
        return [(chunk_ids[r["index"]], r["relevance_score"]) for r in results]

Verify: On your labeled test set, compare nDCG@10 before and after reranking. Expect 10-20% relative improvement. Measure added latency (typically 100-300ms) and confirm it fits your SLA.

Step 8: Construct the generation prompt with citations

The final prompt must include retrieved chunks with stable citation IDs, enforce grounding, and handle context window limits. Use a structured format the LLM can parse.

# generation/prompt.py
from string import Template
from typing import List
from retrieval.hybrid import SearchResult

PROMPT_TEMPLATE = Template("""You are an enterprise knowledge assistant. Answer the user's question using ONLY the provided context. 

Rules:
- Cite sources inline using [doc:N] where N is the doc_id from the context
- If the context doesn't contain the answer, say "I don't have enough information to answer"
- Do not use external knowledge
- Be concise

Context:
$context

Question: $question

Answer:""")

def build_context(chunks: List[SearchResult], max_tokens: int = 8000) -> str:
    """Pack chunks into context, respecting token budget."""
    # Rough token estimate: 4 chars per token
    max_chars = max_tokens * 4
    context_parts = []
    current_chars = 0
    
    for i, chunk in enumerate(chunks):
        citation = f"[doc:{chunk.metadata.get('doc_id', f'chunk_{i}')}]"
        chunk_text = f"{citation} {chunk.text}\n"
        if current_chars + len(chunk_text) > max_chars:
            break
        context_parts.append(chunk_text)
        current_chars += len(chunk_text)
    
    return "\n".join(context_parts)

def build_prompt(question: str, chunks: List[SearchResult]) -> str:
    context = build_context(chunks)
    return PROMPT_TEMPLATE.substitute(context=context, question=question)

Verify: Feed the prompt to your LLM with a question that has no answer in context. Confirm it refuses. Feed a question with multiple relevant chunks — confirm citations reference correct doc_id values.

Step 9: Implement access control at query time

Enterprise RAG must enforce the same permissions as the source systems. Never rely on post-filtering the LLM output. Filter at retrieval time using the access_tier metadata you indexed.

# auth/access.py
from typing import List, Set
from dataclasses import dataclass

@dataclass
class UserContext:
    user_id: str
    groups: List[str]  # e.g., ["legal", "engineering", "support"]
    email: str

# Map access_tier -> required groups
TIER_PERMISSIONS = {
    "public": [],
    "internal": ["employee"],
    "restricted": ["legal", "finance", "hr"],
    "pii_restricted": ["support", "compliance"],
}

def resolve_access_tiers(user: UserContext) -> Set[str]:
    """Return all access tiers the user can read."""
    allowed = {"public", "internal"}  # baseline
    user_groups = set(user.groups)
    
    for tier, required_groups in TIER_PERMISSIONS.items():
        if not required_groups or user_groups & set(required_groups):
            allowed.add(tier)
    return allowed

Wire this into your search call:

# In your query handler
user = get_current_user(request)
allowed_tiers = resolve_access_tiers(user)
results = retriever.retrieve(
    query=question,
    access_tiers=list(allowed_tiers),
    top_k=10,
)

Verify: Create test users with different group memberships. Query each with a known restricted document. Confirm users without the required group receive zero results for that tier.

Step 10: Evaluate end-to-end with a golden dataset

Component-level metrics (recall@k, nDCG) don’t measure answer quality. Build a golden dataset of (question, expected_answer, required_citations) tuples and evaluate the full pipeline.

# eval/golden.py
from dataclasses import dataclass
from typing import List
import json

@dataclass
class GoldenCase:
    question: str
    expected_answer: str
    required_doc_ids: List[str]  # at least one must be cited
    access_tier: str  # minimum tier needed to answer

GOLDEN_CASES = [
    GoldenCase(
        question="What is the termination clause in the Acme Corp MSA?",
        expected_answer="Either party may terminate with 30 days written notice.",
        required_doc_ids=["acme-msa-2024"],
        access_tier="restricted",
    ),
    GoldenCase(
        question="How do I restart the payment service?",
        expected_answer="Run `kubectl rollout restart deployment/payment-service -n prod`.",
        required_doc_ids=["runbook-payment-service"],
        access_tier="internal",
    ),
]

def evaluate_pipeline(pipeline, cases: List[GoldenCase], user: UserContext) -> dict:
    """Run cases, score with LLM-as-judge."""
    from eval.judge import judge_answer  # separate LLM call for scoring
    
    results = []
    for case in cases:
        if case.access_tier not in resolve_access_tiers(user):
            results.append({"case": case.question, "skipped": True, "reason": "insufficient_access"})
            continue
        
        answer, citations = pipeline.run(case.question, user)
        score = judge_answer(case.question, case.expected_answer, answer, citations, case.required_doc_ids)
        results.append({
            "question": case.question,
            "answer": answer,
            "citations": citations,
            "score": score,
            "passed": score >= 0.7,
        })
    
    passed = sum(1 for r in results if r.get("passed", False))
    total = len([r for r in results if not r.get("skipped")])
    return {
        "pass_rate": passed / total if total else 0,
        "details": results,
    }

Verify: Run evaluation weekly. Track pass rate over time. Alert if it drops > 5% week-over-week. Use failures to expand your golden set and improve retrieval.

Step 11: Monitor production metrics

Ship observability from day one. Log every query with: latency breakdown (embedding, vector search, rerank, generation), token counts, retrieval scores, citation coverage, and user feedback.

# observability/logging.py
import structlog
import time
from contextvars import ContextVar
from dataclasses import dataclass, asdict

logger = structlog.get_logger()

current_request: ContextVar[dict] = ContextVar("current_request", default={})

@dataclass
class QueryLog:
    request_id: str
    user_id: str
    question: str
    latency_ms: dict  # {"embedding": 45, "vector_search": 120, "rerank": 180, "generation": 2100}
    tokens: dict      # {"prompt": 4500, "completion": 320}
    retrieval: dict   # {"num_candidates": 50, "num_reranked": 10, "top_score": 0.87}
    citations: list   # [{"doc_id": "x", "score": 0.9}, ...]
    answer: str
    feedback: str = None  # "thumbs_up" | "thumbs_down" | None

def log_query(log: QueryLog):
    logger.info("rag_query", **asdict(log))

Verify: Build a dashboard showing p50/p95/p99 latency per stage, citation rate (fraction of answers with ≥1 citation), and thumbs-down rate. Set alerts on p99 > 5s or thumbs-down > 10%.

Putting it together

The pipeline order matters: ingest → hash → chunk → embed → index → hybrid retrieve → rerank → generate → log. Each stage has a verification step you can automate in CI. Start with a 1,000-document subset, run the golden eval, then scale.

When you need to swap the embedding model or add a new document source, the content hash and chunk ID stability mean you only re-process what changed. The metadata filtering in the vector store keeps authorization simple. The reranker adds ~200ms but catches the hallucinations that pure vector search misses.

If you’re running this at scale across multiple model providers, an inference gateway like n4n.ai can simplify the embedding and generation calls — one endpoint, automatic fallback, and usage metering without rewriting your pipeline code.