A RAG Architecture for Large Document Collections

Scale changes the system, not just the index

Most retrieval-augmented generation (RAG) tutorials index a handful of PDFs, ask a few questions and stop. That is a reasonable way to learn the idea, but it hides the problems that appear when a collection grows by two or three orders of magnitude.

At small scale, a flat vector search over naive fixed-size chunks is often good enough. As the collection grows, three parts of the system start to constrain each other: the ingestion pipeline (parsing, chunking, embedding and re-indexing), retrieval quality (whether the right passages reach the model) and generation quality (whether the model uses those passages faithfully). Optimizing one in isolation can quietly degrade another — larger chunks may help generation but blur retrieval; aggressive caching may cut cost but serve stale or unauthorized answers.

The number of files is also a poor capacity metric. Chunk count, embedding dimension, update frequency, filter selectivity and query volume determine memory, latency and cost. The architecture below is organized in four layers plus generation and evaluation, and each layer is designed so that it can be measured independently.

Layer 1: structure-aware ingestion

Blind fixed-size chunking cuts through headings, clauses and tables. A contract clause split in half, or a table separated from its caption, produces passages that are individually meaningless. Different document types also need different rules: legal documents should keep clause boundaries, manuals should keep procedures together and reports should keep section context.

The chunker below splits on detected headings first and only then falls back to token windows with overlap. Token counts use the same tokenizer family as the embedding or generation model, so budgets are exact rather than estimated from word counts. Every chunk keeps its source metadata, which later enables filtering, access control, citation and reprocessing when the parser improves.

import re
import tiktoken

HEADING = re.compile(r"^(#{1,6} .+|\d+(\.\d+)*[.)]? [A-ZÀ-Ỹ].{0,120}|[A-Z][A-Z ]{3,80})$", re.M)


class StructureAwareChunker:
    """Split on document headings first, then by tokens with overlap."""

    def __init__(self, max_tokens: int = 300, overlap: int = 50):
        if overlap >= max_tokens:
            raise ValueError("overlap must be smaller than max_tokens")
        self.max_tokens = max_tokens
        self.overlap = overlap
        self.encoder = tiktoken.get_encoding("cl100k_base")

    def sections(self, text: str) -> list[str]:
        starts = [m.start() for m in HEADING.finditer(text)]
        if not starts or starts[0] != 0:
            starts.insert(0, 0)
        bounds = starts + [len(text)]
        parts = (text[a:b].strip() for a, b in zip(bounds, bounds[1:]))
        return [p for p in parts if p]

    def split(self, section: str) -> list[str]:
        tokens = self.encoder.encode(section)
        if len(tokens) <= self.max_tokens:
            return [section]
        step = self.max_tokens - self.overlap
        return [
            self.encoder.decode(tokens[i : i + self.max_tokens])
            for i in range(0, len(tokens) - self.overlap, step)
        ]

    def chunk(self, text: str, metadata: dict) -> list[dict]:
        chunks = []
        for section_no, section in enumerate(self.sections(text)):
            for part in self.split(section):
                chunks.append({
                    "text": part,
                    "metadata": {
                        **metadata,
                        "section": section_no,
                        "tokens": len(self.encoder.encode(part)),
                    },
                })
        return chunks

The heading pattern is deliberately simple. Production parsers should use the structure that the source already provides — PDF outlines, DOCX styles, HTML headings — and route scanned or badly formatted files through OCR or a layout-aware extractor (for example pypdf or pdfplumber for digital PDFs and a managed OCR service for scans) before chunking. Poor text extraction is the most common reason retrieval fails, and no amount of downstream tuning recovers text that was never extracted correctly.

Layer 2: hybrid retrieval and reranking

Dense embeddings capture paraphrases and semantic similarity; lexical retrieval such as BM25 remains strong for identifiers, product codes, legal citations and rare terminology. Benchmarks such as BEIR show that neither approach dominates across domains, which is why a hybrid retriever is a sensible default rather than a controversial choice.

The retriever runs both searches, merges the ranked lists with Reciprocal Rank Fusion (RRF) and then reranks a small candidate set with a cross-encoder. RRF needs no score normalization: each list contributes 1 / (k + rank), with k = 60 as proposed by Cormack et al. The cross-encoder, which reads the query and passage together, then decides the final order. Mixing raw cross-encoder scores with RRF values is avoided because the two live on unrelated scales.

from collections.abc import Callable
from qdrant_client import QdrantClient, models
from rank_bm25 import BM25Okapi


def reciprocal_rank_fusion(rankings: list[list[str]], k: int = 60) -> list[tuple[str, float]]:
    """Cormack et al. (2009): sum 1 / (k + rank) across ranked lists."""
    scores: dict[str, float] = {}
    for ranking in rankings:
        for rank, doc_id in enumerate(ranking, start=1):
            scores[doc_id] = scores.get(doc_id, 0.0) + 1.0 / (k + rank)
    return sorted(scores.items(), key=lambda item: item[1], reverse=True)


class HybridRetriever:
    def __init__(
        self,
        client: QdrantClient,
        collection: str,
        corpus: dict[str, str],
        embed: Callable[[str], list[float]],
        rerank: Callable[[str, list[str]], list[float]],
    ):
        self.client = client
        self.collection = collection
        self.ids = list(corpus)
        self.texts = corpus
        self.bm25 = BM25Okapi([corpus[i].lower().split() for i in self.ids])
        self.embed = embed
        self.rerank = rerank

    def dense(self, query: str, limit: int, query_filter=None) -> list[str]:
        hits = self.client.query_points(
            self.collection,
            query=self.embed(query),
            query_filter=query_filter,
            limit=limit,
            with_payload=["doc_id"],
        ).points
        return [hit.payload["doc_id"] for hit in hits]

    def sparse(self, query: str, limit: int) -> list[str]:
        scores = self.bm25.get_scores(query.lower().split())
        order = sorted(range(len(self.ids)), key=lambda i: scores[i], reverse=True)
        return [self.ids[i] for i in order[:limit] if scores[i] > 0]

    def retrieve(self, query: str, top_k: int = 10, candidates: int = 20, query_filter=None) -> list[dict]:
        fused = reciprocal_rank_fusion([
            self.dense(query, candidates, query_filter),
            self.sparse(query, candidates),
        ])[:candidates]
        ids = [doc_id for doc_id, _ in fused]
        # Fusion selects candidates; the cross-encoder decides the final order.
        scores = self.rerank(query, [self.texts[i] for i in ids])
        ranked = sorted(zip(ids, scores), key=lambda item: item[1], reverse=True)
        return [{"id": i, "score": float(s), "text": self.texts[i]} for i, s in ranked[:top_k]]


def cross_encoder_reranker(model_name: str = "cross-encoder/ms-marco-MiniLM-L-6-v2"):
    from sentence_transformers import CrossEncoder

    model = CrossEncoder(model_name)  # load once, not per query
    return lambda query, texts: model.predict([(query, t) for t in texts]).tolist()

Two practical details matter. First, the reranker model should be loaded once at start-up; loading it per query dominates latency. Second, metadata filters (document type, date, tenant, permission scope) belong in the dense query itself so that vector scoring only considers eligible passages. In a large collection the BM25 side should likewise run on a proper search engine with the same filters rather than an in-process index.

Layer 3: choosing and configuring the vector store

Managed services, self-hosted engines and database extensions all work; the right choice depends less on raw query speed than on filtering behavior, update and re-indexing cost, backup and restore, observability, pricing model and who operates the system. Evaluate candidates on your own data, filters and update pattern, and record the date and configuration of any price or performance comparison because both change quickly.

The example uses Qdrant. Full-precision vectors are kept on disk while an int8 scalar-quantized copy stays in memory for search. Scalar quantization stores one byte instead of four per dimension, so the in-memory vector footprint shrinks by roughly a factor of four before index overhead. Payload indexes on filter fields let queries such as “contracts from 2023 that mention arbitration” narrow the candidate set before vector scoring.

import uuid
from qdrant_client import QdrantClient, models


def create_collection(client: QdrantClient, name: str, vector_size: int) -> None:
    client.create_collection(
        collection_name=name,
        vectors_config=models.VectorParams(
            size=vector_size,
            distance=models.Distance.COSINE,
            on_disk=True,  # full-precision vectors stay on disk
        ),
        quantization_config=models.ScalarQuantization(
            scalar=models.ScalarQuantizationConfig(
                type=models.ScalarType.INT8,  # 4 bytes -> 1 byte per dimension
                quantile=0.99,
                always_ram=True,  # quantized copy stays in memory for search
            )
        ),
    )
    # Index fields used for filtering before vector scoring.
    for field, schema in [
        ("doc_type", models.PayloadSchemaType.KEYWORD),
        ("year", models.PayloadSchemaType.INTEGER),
    ]:
        client.create_payload_index(name, field_name=field, field_schema=schema)


def index_chunks(client: QdrantClient, name: str, chunks: list[dict], batch_size: int = 256) -> int:
    batch: list[models.PointStruct] = []
    count = 0
    for chunk in chunks:
        batch.append(models.PointStruct(
            # Deterministic IDs make re-indexing idempotent.
            id=str(uuid.uuid5(uuid.NAMESPACE_URL, f"{chunk['doc_id']}#{chunk['chunk_no']}")),
            vector=chunk["embedding"],
            payload={
                "doc_id": chunk["doc_id"],
                "text": chunk["text"],
                "source": chunk["source"],
                "page": chunk["page"],
                "doc_type": chunk["doc_type"],
                "year": chunk["year"],
            },
        ))
        if len(batch) == batch_size:
            client.upsert(name, points=batch)
            count += len(batch)
            batch = []
    if batch:
        client.upsert(name, points=batch)
        count += len(batch)
    return count


def contracts_filter(year: int) -> models.Filter:
    return models.Filter(must=[
        models.FieldCondition(key="doc_type", match=models.MatchValue(value="contract")),
        models.FieldCondition(key="year", match=models.MatchValue(value=year)),
    ])

Deterministic point IDs derived from the document and chunk number make re-indexing idempotent: running the pipeline again updates points instead of duplicating them. Quantization can reduce recall slightly; measure it on your evaluation set and enable rescoring with the original vectors if the loss matters.

Layer 4: semantic caching

Many questions in an organization are near-duplicates. A semantic cache stores the embedding of a previous question together with its answer and sources, and returns that answer when a new question is close enough. This can save generation cost and latency, but it introduces three risks: stale answers, answers leaked across users with different permissions and false matches between questions that look similar but differ in an important detail.

The version below stores cache entries in a vector collection instead of scanning every key, filters by freshness and permission scope, and only accepts very close matches.

import time
import uuid
from qdrant_client import QdrantClient, models


class SemanticCache:
    """Reuse answers for near-duplicate questions via a vector index, not a key scan."""

    def __init__(self, client: QdrantClient, name: str, vector_size: int,
                 threshold: float = 0.95, ttl_seconds: int = 3600):
        self.client, self.name = client, name
        self.threshold, self.ttl = threshold, ttl_seconds
        if not client.collection_exists(name):
            client.create_collection(name, vectors_config=models.VectorParams(
                size=vector_size, distance=models.Distance.COSINE))
            client.create_payload_index(name, field_name="created_at",
                                        field_schema=models.PayloadSchemaType.FLOAT)

    def get(self, embedding: list[float], scope: str) -> dict | None:
        fresh = models.Filter(must=[
            models.FieldCondition(key="created_at", range=models.Range(gte=time.time() - self.ttl)),
            # Never share answers across users or permission scopes.
            models.FieldCondition(key="scope", match=models.MatchValue(value=scope)),
        ])
        hits = self.client.query_points(self.name, query=embedding, query_filter=fresh,
                                        score_threshold=self.threshold, limit=1).points
        return hits[0].payload if hits else None

    def set(self, embedding: list[float], scope: str, question: str, answer: str, sources: list[str]) -> None:
        self.client.upsert(self.name, points=[models.PointStruct(
            id=str(uuid.uuid4()), vector=embedding,
            payload={"scope": scope, "question": question, "answer": answer,
                     "sources": sources, "created_at": time.time()},
        )])

The similarity threshold should be tuned on labelled pairs of equivalent and non-equivalent questions, and cache entries must be invalidated when their source documents change. Report the cache hit rate together with the rate of incorrect cache hits; a high hit rate is not useful if a meaningful share of those answers is wrong.

Generation: packing context and requiring citations

Retrieving the right passages is only half of the task; the model must use them and show where each claim comes from. The generator packs the highest-ranked passages into an explicit token budget, numbers them, and instructs the model to answer only from that context and to cite document and page for each claim. A low temperature keeps answers close to the sources.

import tiktoken
from openai import OpenAI

SYSTEM_PROMPT = """Answer only from the numbered context documents.
Cite every claim as [Document N, page P].
If the context does not contain the answer, say so and do not guess."""


def pack_context(docs: list[dict], budget_tokens: int, encoder=None) -> list[dict]:
    encoder = encoder or tiktoken.get_encoding("cl100k_base")
    packed, used = [], 0
    for doc in sorted(docs, key=lambda d: d["score"], reverse=True):
        cost = len(encoder.encode(doc["text"]))
        if used + cost > budget_tokens:
            continue  # a shorter, lower-ranked chunk may still fit
        packed.append(doc)
        used += cost
    return packed


def build_messages(question: str, context: list[dict]) -> list[dict]:
    blocks = "\n\n".join(
        f"[Document {i}] source={d['source']} page={d['page']}\n{d['text']}"
        for i, d in enumerate(context, start=1)
    )
    return [
        {"role": "system", "content": SYSTEM_PROMPT},
        {"role": "user", "content": f"Context:\n{blocks}\n\nQuestion: {question}"},
    ]


def answer(client: OpenAI, model: str, question: str, docs: list[dict], budget_tokens: int = 6000) -> dict:
    context = pack_context(docs, budget_tokens)
    response = client.chat.completions.create(
        model=model,  # choose and pin a current model in configuration
        messages=build_messages(question, context),
        temperature=0.1,
        max_tokens=1000,
    )
    return {"answer": response.choices[0].message.content, "context": context}

Citations make answers verifiable by the user, but they do not guarantee faithfulness: a model can still cite a passage that does not support the claim. Treat citation support as a metric to measure, not as a property to assume. Keep prompts short and specific — long lists of rules are often followed less reliably than a few clear instructions with an example.

The metrics to measure, and how

A RAG system should be evaluated layer by layer, on a question set that reflects real use and has been reviewed by domain experts. Each labelled item needs the question, the passages that answer it and, ideally, the properties a good answer must have. Report every result with the dataset version, date, models, hardware and configuration so that it can be reproduced.

• Retrieval: Recall@k (did the relevant passages reach the candidate set?) and MRR or nDCG (were they ranked near the top?). Evaluate with and without filters and reranking to see what each component contributes.
• Generation: faithfulness (is every claim supported by the retrieved context?), answer relevance and citation correctness. Frameworks such as RAGAS provide reference implementations of these measures.
• Operations: latency percentiles (p50 and p95) for retrieval and end-to-end responses, cost per query, cache hit rate and incorrect-hit rate, indexing throughput and freshness lag after a document changes.
• User outcomes: task success and explicit feedback, collected with consent and reviewed alongside the logged retrieval results.

import json
import logging
import time
from contextlib import contextmanager

log = logging.getLogger("rag")


@contextmanager
def traced_query(query_id: str):
    record = {"query_id": query_id}
    started = time.perf_counter()
    try:
        yield record
    finally:
        record["latency_ms"] = round((time.perf_counter() - started) * 1000, 1)
        log.info(json.dumps(record))  # log IDs and scores, not private document text


def recall_at_k(retrieved: list[str], relevant: set[str], k: int) -> float:
    return len(set(retrieved[:k]) & relevant) / len(relevant) if relevant else 0.0


def reciprocal_rank(retrieved: list[str], relevant: set[str]) -> float:
    return next((1.0 / rank for rank, doc in enumerate(retrieved, 1) if doc in relevant), 0.0)


def evaluate(retrieve, labelled: list[dict], k: int = 10) -> dict:
    """labelled: [{"id", "question", "relevant": [...doc ids]}] reviewed by domain experts."""
    recalls, ranks = [], []
    for item in labelled:
        with traced_query(item["id"]) as record:
            ids = [hit["id"] for hit in retrieve(item["question"], top_k=k)]
            record["top_ids"] = ids
        relevant = set(item["relevant"])
        recalls.append(recall_at_k(ids, relevant, k))
        ranks.append(reciprocal_rank(ids, relevant))
    n = len(labelled)
    return {"questions": n, f"recall@{k}": sum(recalls) / n, "mrr": sum(ranks) / n}

Log query identifiers, retrieved IDs, scores and latency rather than private document text. When an answer is wrong, first check whether the right passage was retrieved: if it was not, generation quality is irrelevant and the fix belongs in ingestion or retrieval.

Lessons that apply to most RAG projects

• Invest in document preprocessing first. OCR errors, broken reading order and lost tables damage retrieval more than any choice of embedding model.
• Prefer short, precise prompts. Hundreds of tokens of instructions are harder for a model to follow consistently than a few clear rules.
• Monitor retrieval, not only the final answer. Most failures that look like hallucination start with the wrong or missing context.

Directions worth exploring

The practical path is incremental: start with a representative subset of documents, build the evaluation set early, measure each layer, then scale while watching the metrics that the previous step revealed. Users judge the system by whether it finds the right answer faster and more reliably than their current alternative; architecture choices matter only in how they serve that outcome.

• Multi-vector and late-interaction retrieval, which represent a passage with several embeddings instead of one.
• Learning from reviewed user feedback to fine-tune the retriever or reranker.
• Graph-based context that links related chunks — clauses that reference each other, versions of the same policy — so that retrieval can follow those relationships.

References

1. Lewis et al. (2020). Retrieval-Augmented Generation for Knowledge-Intensive NLP Tasks. arXiv:2005.11401
2. Gao et al. (2023). Retrieval-Augmented Generation for Large Language Models: A Survey. arXiv:2312.10997
3. Robertson & Zaragoza (2009). The Probabilistic Relevance Framework: BM25 and Beyond. Foundations and Trends in Information Retrieval
4. Karpukhin et al. (2020). Dense Passage Retrieval for Open-Domain Question Answering. arXiv:2004.04906
5. Reimers & Gurevych (2019). Sentence-BERT. arXiv:1908.10084
6. Thakur et al. (2021). BEIR: A Heterogeneous Benchmark for Zero-shot Evaluation of Information Retrieval Models. arXiv:2104.08663
7. Cormack, Clarke & Buettcher (2009). Reciprocal Rank Fusion Outperforms Condorcet and Individual Rank Learning Methods. SIGIR
8. Nogueira & Cho (2019). Passage Re-ranking with BERT. arXiv:1901.04085
9. Khattab & Zaharia (2020). ColBERT: Efficient and Effective Passage Search via Contextualized Late Interaction over BERT. arXiv:2004.12832
10. Es et al. (2023). Ragas: Automated Evaluation of Retrieval Augmented Generation. arXiv:2309.15217
11. Asai et al. (2023). Self-RAG: Learning to Retrieve, Generate, and Critique through Self-Reflection. arXiv:2310.11511
12. Qdrant documentation: Quantization