Production RAG Pipelines, Evaluation & Topic Modeling

A chunk is not a deliverable. A maintained, evaluated, topic-aware index is.

Vec, Pipeline-Minded AI Agent

31.7.1 Production RAG ETL Pipelines

A production ingestion pipeline must handle document updates, deletions, and versioning in addition to initial loading. The key engineering challenges include:

Incremental Indexing

When documents are updated, you must re-chunk and re-embed only the changed documents, not the entire corpus. This requires tracking document versions (typically via content hashes or timestamps) and maintaining a mapping between source documents and their chunks in the vector database.

# Incremental indexing with content hashing
import hashlib
import json
from typing import Dict, List, Optional
from pathlib import Path


class IncrementalIndexer:
    """
    Tracks document versions to enable incremental re-indexing.
    Only processes documents that have changed since the last run.
    """

    def __init__(self, state_file: str = "indexer_state.json"):
        self.state_file = Path(state_file)
        self.state: Dict[str, str] = {}
        if self.state_file.exists():
            self.state = json.loads(self.state_file.read_text())

    def content_hash(self, content: str) -> str:
        return hashlib.sha256(content.encode()).hexdigest()

    def get_changes(
        self, documents: Dict[str, str]
    ) -> Dict[str, List[str]]:
        """
        Compare current documents against stored state.

        Args:
            documents: dict of {doc_id: content}

        Returns:
            {"added": [...], "modified": [...], "deleted": [...]}
        """
        current_ids = set(documents.keys())
        stored_ids = set(self.state.keys())
        added = current_ids - stored_ids
        deleted = stored_ids - current_ids
        modified = set()
        for doc_id in current_ids & stored_ids:
            new_hash = self.content_hash(documents[doc_id])
            if new_hash != self.state[doc_id]:
                modified.add(doc_id)
        return {
            "added": list(added),
            "modified": list(modified),
            "deleted": list(deleted),
        }

    def update_state(self, documents: Dict[str, str]):
        """Update stored hashes after successful indexing."""
        for doc_id, content in documents.items():
            self.state[doc_id] = self.content_hash(content)
        self.state_file.write_text(json.dumps(self.state, indent=2))

    def process_changes(self, documents: Dict[str, str]):
        """Main entry point for incremental processing."""
        changes = self.get_changes(documents)
        print(f"Added: {len(changes['added'])} documents")
        print(f"Modified: {len(changes['modified'])} documents")
        print(f"Deleted: {len(changes['deleted'])} documents")
        # For added/modified: chunk, embed, upsert
        to_process = changes["added"] + changes["modified"]
        if to_process:
            print(f"Processing {len(to_process)} documents...")
            # chunk_and_embed(to_process)
            # vector_db.upsert(chunks)
        # For deleted: remove from vector DB
        if changes["deleted"]:
            print(f"Removing {len(changes['deleted'])} documents...")
            # vector_db.delete(filter={"doc_id": {"$in": changes["deleted"]}})
        # For modified: also remove old chunks before upserting new ones
        if changes["modified"]:
            print(f"Replacing chunks for {len(changes['modified'])} documents...")
            # vector_db.delete(filter={"doc_id": {"$in": changes["modified"]}})
            # vector_db.upsert(new_chunks)
        self.update_state(documents)


# Usage
indexer = IncrementalIndexer()
docs = {
    "report_2024.pdf": "Full text of the 2024 report...",
    "manual_v3.pdf": "Updated product manual content...",
    "faq.md": "Frequently asked questions...",
}
indexer.process_changes(docs)

Metadata Enrichment

Every chunk should carry metadata that enables effective filtering and attribution. Essential metadata fields include:

• Source: The original file name or URL for citation and deduplication.
• Page/section: Location within the source document for precise references.
• Title hierarchy: Section and subsection headings for contextual understanding.
• Date: Creation or last-modified date for recency filtering.
• Document type: Category labels (policy, FAQ, report, transcript) for scoped search.
• Access permissions: User or group identifiers for access-controlled retrieval.

31.7.2 Evaluation and Iteration

Chunking is not a one-time configuration; it requires ongoing evaluation and tuning. The most effective approach is to build a small evaluation set of 50 to 100 representative queries with known relevant passages, then measure retrieval metrics (recall@k, MRR, NDCG) across different chunking configurations. Systematic A/B testing of chunking strategies often reveals that the optimal configuration depends heavily on the document type and query patterns specific to your application. Figure 31.7.1a shows the iterative evaluation loop for chunking quality.

Figure 31.7.1b: Chunking quality requires iterative evaluation against representative queries with ground-truth relevance judgments.

31.7.3 Topic Modeling with LLM Embeddings

Topic modeling discovers the latent themes in a collection of documents without requiring labeled data. Classical approaches like LDA (Latent Dirichlet Allocation) and NMF (Non-negative Matrix Factorization) operate on bag-of-words representations, which discard word order and semantic nuance. BERTopic (Grootendorst, 2022) replaces this with a pipeline built on the same embedding models used for retrieval, producing topics that are semantically coherent and interpretable. Understanding BERTopic is valuable because the same embeddings you create for RAG (as covered in Section 31.1) can power topic discovery, clustering, and content organization without additional model training.

31.7.3.1 The BERTopic Pipeline

BERTopic operates in four sequential stages, each handled by a separate, swappable component:

1. Embed: Convert each document to a dense vector using a sentence softmax model (the same models from Section 31.1).
2. Reduce: Project the high-dimensional embeddings into a lower-dimensional space using UMAP, preserving local neighborhood structure while making clustering feasible.
3. Cluster: Group similar documents using HDBSCAN, a density-based clustering algorithm that automatically determines the number of clusters and identifies outliers (documents that do not fit any topic).
4. Represent: Label each cluster with descriptive terms using c-TF-IDF (class-based TF-IDF) or, optionally, an LLM that generates human-readable topic labels from the cluster's representative documents.

Figure 31.7.4: The BERTopic pipeline. Each stage is a swappable component, so the same architecture supports any embedding model, any reducer (UMAP, PCA), any clustering algorithm (HDBSCAN, k-means), and any labeling strategy (c-TF-IDF, KeyBERT, LLM-generated names).

The labeling stage uses a class-based variant of TF-IDF (Grootendorst, 2022) that treats each cluster as a single super-document. For class $c$ and term $t$, the c-TF-IDF score is:

where $\mathrm{tf}_{t,c}$ is the frequency of term $t$ inside cluster $c$, $\mathrm{tf}_t$ is its frequency across the whole corpus, and $A$ is the average number of words per cluster. The $\log(1 + A/\mathrm{tf}_t)$ factor penalizes terms that appear everywhere (stopwords) and rewards terms that are concentrated inside one cluster. Sorting terms by c-TF-IDF and keeping the top 10 gives the keyword label that BERTopic reports.

31.7.3.2 BERTopic vs. Classical Topic Models

31.7.3.3 Practical Example

This snippet demonstrates a practical hybrid search query that combines dense and sparse retrieval.

# BERTopic: embedding-based topic modeling
# Uses the same sentence transformers as RAG embedding pipelines
from bertopic import BERTopic
from sentence_transformers import SentenceTransformer
from umap import UMAP
from hdbscan import HDBSCAN
# Step 1: Configure each pipeline component
embedding_model = SentenceTransformer("all-MiniLM-L6-v2")
umap_model = UMAP(
    n_neighbors=15, n_components=5,
    min_dist=0.0, metric="cosine",
    )
hdbscan_model = HDBSCAN(
    min_cluster_size=15,
    metric="euclidean",
    prediction_data=True,
    )
# Step 2: Build the BERTopic model with custom components
topic_model = BERTopic(
    embedding_model=embedding_model,
    umap_model=umap_model,
    hdbscan_model=hdbscan_model,
    verbose=True,
    )
# Step 3: Fit on your documents (e.g., customer support tickets)
documents = [
    "My order has not arrived after two weeks",
    "How do I reset my password for the dashboard?",
    "The API returns a 500 error on large payloads",
    "I was charged twice for the same subscription",
    "Can I export my data as a CSV file?",
    # ... thousands more documents
    ]
topics, probs = topic_model.fit_transform(documents)
# Step 4: Inspect discovered topics
topic_info = topic_model.get_topic_info()
print(topic_info.head(10))
# Step 5: Get the top terms for a specific topic
for topic_id in range(min(5, len(topic_model.get_topics()))):
    terms = topic_model.get_topic(topic_id)
    print(f"\nTopic {topic_id}:")
    for term, score in terms[:5]:
        print(f" {score:.3f} {term}")

31.7.3.4 Plug-In Representation Models

The default c-TF-IDF keyword list per topic is a reasonable starting point but rarely the most readable one. BERTopic exposes a representation_model hook so the four-stage pipeline (embed, reduce, cluster, represent) can be customized at the final step without touching the upstream UMAP or HDBSCAN configuration. Three plug-ins cover almost every production need.

KeyBERTInspired re-ranks the top c-TF-IDF candidate words for each topic by their cosine similarity to the topic's centroid embedding. The intuition is that c-TF-IDF favors words that distinguish the cluster from other clusters statistically, but those words are not always the words a human would call descriptive. KeyBERTInspired pushes the keyword list toward terms that are semantically central to the topic, not just statistically distinctive. MaximalMarginalRelevance applies the MMR algorithm (see Section 35.2.1.5) to the candidate keywords, so the final list trades relevance against diversity and avoids near-duplicates like "summary" and "summaries" both surfacing for the same topic. OpenAI hands a few representative documents per topic to an LLM and asks it to write a one-line topic label, the same trick promised in the fun-fact above but now an actual line of code. The three plug-ins stack:

from bertopic import BERTopic
from bertopic.representation import (
    KeyBERTInspired, MaximalMarginalRelevance, OpenAI,
)
from openai import OpenAI as OpenAIClient

client = OpenAIClient()

representation = {
    "Main":      KeyBERTInspired(),                          # centroid-aware keywords
    "MMR":       MaximalMarginalRelevance(diversity=0.3),    # de-duped keywords
    "LLM Label": OpenAI(client=client, model="gpt-4o-mini"), # one-line topic name
}

topic_model = BERTopic(
    embedding_model=embedder,
    representation_model=representation,
)
topics, _ = topic_model.fit_transform(docs)
print(topic_model.get_topic_info().head())

31.7.3.5 BERTopic Visualizations

BERTopic ships an interactive visualization layer that turns the topic model into a quick analytical UI without writing any plotting code. Four methods cover the common reporting needs: visualize_documents() plots every document as a point in 2-D UMAP space, colored by topic, with hover tooltips showing the document text; visualize_topics() draws an inter-topic distance map so reviewers can see which topics overlap and which are well separated; visualize_hierarchy() renders the dendrogram of agglomerative clustering over topic embeddings, which is the input to the next method; reduce_topics(nr_topics=20) collapses fine-grained topics into a target number using that same hierarchy, useful when HDBSCAN produces 80 small topics but the business audience wants 10 broad themes. All four return Plotly figures that drop directly into a Jupyter notebook or a Streamlit dashboard.

document = (
    "# Introduction to Machine Learning\n\n"
    "Machine learning is a branch of artificial intelligence that enables "
    "computers to learn from data without being explicitly programmed.\n\n"
    "## Supervised Learning\n\n"
    "In supervised learning, the algorithm learns from labeled training data. "
    "Each example consists of an input and a desired output.\n\n"
    "### Classification\n\n"
    "Classification predicts categorical labels. For example, an email spam "
    "filter classifies emails as spam or not spam. Popular algorithms include "
    "logistic regression, SVMs, and random forests.\n\n"
    "### Regression\n\n"
    "Regression predicts continuous numerical values. For instance, predicting "
    "house prices based on features like square footage and location.\n\n"
    "## Unsupervised Learning\n\n"
    "Unsupervised learning works with unlabeled data, seeking to discover "
    "hidden patterns and structures without target labels.\n\n"
    "### Clustering\n\n"
    "Clustering groups similar data points together. K-means partitions data "
    "into k groups. DBSCAN discovers clusters of arbitrary shape based on "
    "density. Hierarchical clustering builds a tree of nested clusters.\n\n"
    "### Dimensionality Reduction\n\n"
    "Dimensionality reduction compresses high-dimensional data. PCA finds "
    "directions of maximum variance. t-SNE and UMAP create 2D visualizations "
    "that preserve local neighborhood structure.\n\n"
    "## Deep Learning\n\n"
    "Deep learning uses neural networks with many layers to learn hierarchical "
    "representations. It has achieved breakthroughs in vision, NLP, and games."
)
print(f"Document: {len(document)} chars, {document.count(chr(10))} lines")

import numpy as np
from sentence_transformers import SentenceTransformer
model = SentenceTransformer("all-MiniLM-L6-v2")
# Strategy 1: Fixed-size with overlap
def fixed_chunk(text, size=300, overlap=50):
    chunks, start = [], 0
    while start < len(text):
        chunk = text[start:start+size].strip()
        if chunk:
            chunks.append(chunk)
            start += size - overlap
            return chunks
# Strategy 2: Recursive splitting on headers
def recursive_chunk(text, max_size=500):
    # TODO: Split on "## " first, then "### " for oversized sections
    # Merge chunks smaller than max_size/3 with their neighbors
    sections = text.split("\n## ")
    chunks = []
    for section in sections:
        section = section.strip()
        if not section:
            continue
        if len(section) <= max_size:
            chunks.append(section)
        else:
            for sub in section.split("\n### "):
                sub = sub.strip()
                if sub:
                    chunks.append(sub)
                    return chunks
# Strategy 3: Semantic chunking
def semantic_chunk(text, model, threshold=0.5):
    # TODO: Split into sentences, encode them, find similarity drops
    sentences = [s.strip() for s in text.replace('\n', ' ').split('. ')
        if len(s.strip()) > 10]
    if len(sentences) <= 1:
        return sentences
    embs = model.encode(sentences)
    norms = np.linalg.norm(embs, axis=1)
    chunks, current = [], [sentences[0]]
    for i in range(len(sentences) - 1):
        sim = np.dot(embs[i], embs[i+1]) / (norms[i] * norms[i+1] + 1e-8)
        if sim < threshold:
            chunks.append(". ".join(current))
            current = [sentences[i+1]]
        else:
            current.append(sentences[i+1])
            if current:
                chunks.append(". ".join(current))
                return chunks
            c_fixed = fixed_chunk(document)
            c_recursive = recursive_chunk(document)
            c_semantic = semantic_chunk(document, model)
            for name, chunks in [("Fixed", c_fixed), ("Recursive", c_recursive),
                ("Semantic", c_semantic)]:
                print(f"\n{name}: {len(chunks)} chunks")
                for i, c in enumerate(chunks):
                    print(f" [{i}] {len(c)} chars: {c[:60]}...")

import numpy as np
queries_expected = [
    ("What is classification in ML?", "classification"),
    ("How does clustering work?", "clustering"),
    ("What is PCA used for?", "dimensionality"),
    ("What is deep learning?", "deep learning"),
    ]
def search_chunks(query, chunks, model, top_k=1):
    qe = model.encode(query)
    ce = model.encode(chunks)
    scores = np.dot(ce, qe) / (np.linalg.norm(ce, axis=1) * np.linalg.norm(qe))
    idx = np.argsort(scores)[::-1][:top_k]
    return [(chunks[i], scores[i]) for i in idx]
for query, keyword in queries_expected:
    print(f"\nQuery: {query}")
    for name, chunks in [("Fixed", c_fixed), ("Recursive", c_recursive),
        ("Semantic", c_semantic)]:
        top_chunk, score = search_chunks(query, chunks, model)[0]
        hit = "PASS" if keyword.lower() in top_chunk.lower() else "MISS"
        print(f" {name:10s} [{hit}] score={score:.3f} | {top_chunk[:55]}...")

import numpy as np
from sentence_transformers import SentenceTransformer
model = SentenceTransformer("all-MiniLM-L6-v2")
document = (
    "# Introduction to Machine Learning\n\n"
    "Machine learning enables computers to learn from data.\n\n"
    "## Supervised Learning\n\nLearns from labeled data.\n\n"
    "### Classification\n\nPredicts categorical labels like spam/not-spam.\n\n"
    "### Regression\n\nPredicts continuous values like house prices.\n\n"
    "## Unsupervised Learning\n\nFinds patterns in unlabeled data.\n\n"
    "### Clustering\n\nGroups similar points. K-means, DBSCAN, hierarchical.\n\n"
    "### Dimensionality Reduction\n\nCompresses data. PCA, t-SNE, UMAP.\n\n"
    "## Deep Learning\n\nNeural networks with many layers for hierarchical representations."
    )
def fixed_chunk(text, size=300, overlap=50):
    chunks, s = [], 0
    while s < len(text):
        c = text[s:s+size].strip()
        if c: chunks.append(c)
        s += size - overlap
        return chunks
def recursive_chunk(text, max_size=500):
    chunks = []
    for sec in text.split("\n## "):
        sec = sec.strip()
        if not sec: continue
        if len(sec) <= max_size: chunks.append(sec)
    else:
        for sub in sec.split("\n### "):
            sub = sub.strip()
            if sub: chunks.append(sub)
            return chunks
def semantic_chunk(text, model, threshold=0.5):
    sents = [s.strip() for s in text.replace('\n',' ').split('. ') if len(s.strip())>10]
    if len(sents) <= 1: return sents
    embs = model.encode(sents)
    norms = np.linalg.norm(embs, axis=1)
    chunks, cur = [], [sents[0]]
    for i in range(len(sents)-1):
        sim = np.dot(embs[i],embs[i+1])/(norms[i]*norms[i+1]+1e-8)
        if sim < threshold: chunks.append(". ".join(cur)); cur = [sents[i+1]]
    else: cur.append(sents[i+1])
    if cur: chunks.append(". ".join(cur))
    return chunks
cf, cr, cs = fixed_chunk(document), recursive_chunk(document), semantic_chunk(document, model)
def search(q, chunks, model):
    qe = model.encode(q); ce = model.encode(chunks)
    scores = np.dot(ce,qe)/(np.linalg.norm(ce,axis=1)*np.linalg.norm(qe))
    i = np.argmax(scores)
    return chunks[i], scores[i]
for q, kw in [("What is classification?","classification"),("How does clustering work?","clustering"),
    ("What is PCA?","dimensionality"),("What is deep learning?","deep learning")]:
    print(f"\n{q}")
    for nm, ch in [("Fixed",cf),("Recursive",cr),("Semantic",cs)]:
        c, s = search(q, ch, model)
        print(f" {nm:10s} [{'PASS' if kw in c.lower() else 'MISS'}] {s:.3f} | {c[:55]}")

from sklearn.metrics.pairwise import cosine_similarity
import numpy as np

def search_chunks(query, chunks, model, top_k=1):
    qe = model.encode([query])
    ce = model.encode(chunks)
    sims = cosine_similarity(qe, ce)[0]
    idx = np.argsort(sims)[::-1][:top_k]
    return [(chunks[i], sims[i]) for i in idx]

from langchain_text_splitters import RecursiveCharacterTextSplitter

splitter = RecursiveCharacterTextSplitter(
    chunk_size=500, chunk_overlap=50,
    separators=["\n## ", "\n### ", "\n\n", "\n", ". ", " "],
)
chunks = splitter.split_text(document)

Exercises