Bry

Posted on Sep 15 Originally published at Medium

Building a RAG Pipeline from Scratch: Embeddings, Retrieval, and Claude

#ai #llm #python #programming

Key Points

RAG (Retrieval-Augmented Generation) connects an LLM to your private data at query time — no fine-tuning, no retraining, no data leakage into model weights.
The pipeline has five stages: Ingest → Chunk → Embed → Store → Query (retrieve, augment, generate). Getting chunking and retrieval right matters more than which LLM you pick.
ChromaDB runs in-process for local dev with zero infrastructure — collection.add() to insert, collection.query() to retrieve. Production options include Pinecone and pgvector.
Common failure modes — chunks too large, no deduplication, no query expansion, skipping evaluation — are all avoidable. This article shows how.

Introduction

Large language models hallucinate when they don't know the answer. The standard fix — fine-tuning on your private data — costs tens of thousands of dollars, takes weeks, and produces a static model that goes stale as soon as your data changes. RAG solves a different problem: instead of baking knowledge into weights, it retrieves the relevant facts at query time and gives them to the model as context. The model answers from evidence, not memory. I've built RAG pipelines for internal knowledge bases and customer-facing chat tools — the architecture is the same whether you're indexing 500 internal docs or 500,000 support tickets; the parameters are what change.

This article builds a complete RAG pipeline in Python using ChromaDB for vector storage, sentence-transformers for local embeddings, and Claude for generation. You will understand each stage, why the design choices matter, and what breaks in production if you skip them. The full working implementation is in src/pipeline.py and src/main.py.

What RAG Is — and What It Replaces

Before writing code, align on why RAG exists.

Fine-tuning trains the model on your data. It is expensive (typically $5,000–$50,000+ for a production run), takes days to weeks, and produces a checkpoint that bakes in your data as of the training date. If your documentation changes next week, the model doesn't know. Fine-tuning is correct when you need the model to learn a style, a domain vocabulary, or a task format — not when you need it to answer questions from a document set.

Prompt-only (stuffing) puts your documents directly into the context window. It works for small document sets (tens of pages) but breaks on large corpora: context windows have limits, filling them with irrelevant text degrades answer quality, and at scale the cost per query becomes prohibitive.

RAG indexes your documents, retrieves only the chunks relevant to each query, and gives the model a focused context. It handles large corpora, stays current as documents are updated, and costs a fraction of fine-tuning. The tradeoff is a retrieval layer you have to build and maintain.

Approach	Data scale	Latency	Cost	Freshness
Fine-tuning	Any	Low (no retrieval)	High (training + inference)	Static — retrain to update
Prompt-only	Small (< 50 pages)	Low	Medium	Fresh
RAG	Large (any)	Medium (retrieval + LLM)	Low–Medium	Fresh (re-embed on update)

RAG Pipeline Architecture

A RAG pipeline has two sides: the ingest side (run once per document update) and the query side (run on every user request). Together they form five stages.

Diagram: Ingest side (left) runs document processing once. Query side (right) runs on every user request.

The five stages:

Ingest — load raw documents. Source can be text files, PDFs, database records, or API responses.
Chunk — split documents into pieces small enough to embed meaningfully. Chunk size is the most consequential parameter in the pipeline.
Embed — convert each chunk to a vector. The embedding model maps semantic meaning to a point in high-dimensional space.
Store — persist vectors (with their source text and metadata) in a vector database.
Query — embed the user's question, find the nearest chunks, inject them into a prompt, and generate an answer.

Chunking Strategies

Chunking is where most RAG pipelines fail. A bad chunking strategy produces irrelevant retrievals; irrelevant retrievals produce hallucinated answers. The model can't conjure information that wasn't in the retrieved chunks. In practice, I spend more time tuning chunk size and overlap than on any other pipeline parameter — getting it wrong is invisible until a user catches the model confidently citing the wrong passage.

Fixed-Size Chunking

Split every N characters or tokens. Simple to implement, ignores document structure.

# Fixed-size chunking — simple but blunt
def chunk_fixed(text: str, size: int = 400, overlap: int = 50) -> list[str]:
    chunks = []
    start = 0
    while start < len(text):
        end = start + size
        chunks.append(text[start:end])
        start += size - overlap  # overlap preserves context at boundaries
    return chunks

The overlap matters: without it, a sentence split across two chunks retrieves each half separately and both halves are incomplete.

Recursive Chunking

Try splitting on paragraph breaks (\n\n), then line breaks (\n), then spaces. Each level is a fallback when the previous splitter still produces chunks over the target size. This respects document structure — paragraphs before sentences before words.

def _chunk(self, text: str, max_tokens: int = 400) -> list[str]:
    """Recursively split text into chunks, preferring natural boundaries.

    Args:
        text: The input text to split.
        max_tokens: Approximate maximum chunk size in tokens (1 token ≈ 4 chars).

    Returns:
        List of text chunks, each within the max_tokens limit.
    """
    max_chars = max_tokens * 4  # rough token → char estimate

    if len(text) <= max_chars:
        return [text.strip()] if text.strip() else []

    # Try splitting on paragraph breaks first, then newlines, then spaces
    for separator in ["\n\n", "\n", " "]:
        parts = text.split(separator)
        if len(parts) > 1:
            chunks: list[str] = []
            current = ""
            for part in parts:
                candidate = (current + separator + part).strip() if current else part.strip()
                if len(candidate) <= max_chars:
                    current = candidate
                else:
                    if current:
                        chunks.append(current)
                    # Recurse on oversized parts
                    if len(part) > max_chars:
                        chunks.extend(self._chunk(part, max_tokens))
                        current = ""
                    else:
                        current = part.strip()
            if current:
                chunks.append(current)
            return [c for c in chunks if c]

    # No separator found — hard split
    return [text[:max_chars].strip(), text[max_chars:].strip()]

Semantic Chunking

Group sentences by semantic similarity — a sentence joins the current chunk if its embedding is similar enough; otherwise it starts a new chunk. Produces the most coherent chunks but is slower (requires embedding every sentence during ingestion).

Diagram: Chunking strategy selection — recursive approach, falling back from paragraphs to lines to words.

Embeddings

An embedding model converts text into a dense vector — a list of floats (e.g., 384 dimensions for all-MiniLM-L6-v2) where similar texts produce similar vectors. Semantic similarity becomes geometric proximity.

Local Embeddings: `sentence-transformers`

sentence-transformers runs entirely locally — no API key, no per-token cost, no latency from network calls.

from sentence_transformers import SentenceTransformer

model = SentenceTransformer("all-MiniLM-L6-v2")

texts = [
    "Python is a high-level programming language.",
    "Guido van Rossum created Python in 1991.",
    "The weather is sunny today.",
]

embeddings = model.encode(texts)  # shape: (3, 384)
print(embeddings.shape)           # (3, 384)

all-MiniLM-L6-v2 is 80MB on disk, encodes ~14,000 sentences per second on CPU, and produces 384-dimensional vectors. It is the right default for local development and prototyping.

API Embeddings

For production, Anthropic's and OpenAI's embedding APIs trade local cost for higher-quality vectors at scale. Use them when your retrieval accuracy on domain-specific content drops below acceptable thresholds.

Option	Model	Dims	Cost	Notes
`sentence-transformers`	`all-MiniLM-L6-v2`	384	Free (local)	Best for dev/prototyping
`sentence-transformers`	`all-mpnet-base-v2`	768	Free (local)	Higher quality, 3× slower
OpenAI API	`text-embedding-3-small`	1536	$0.02 / 1M tokens	Good price/quality tradeoff
Anthropic API	Voyage-3 (via Voyage AI)	1024	$0.06 / 1M tokens	SOTA quality for RAG

Keep your embedding model consistent between ingestion and query time. If you embed documents with all-MiniLM-L6-v2 and query with text-embedding-3-small, your similarity scores will be meaningless.

Vector Storage with ChromaDB

ChromaDB is an in-process vector database for Python. It requires no server, no Docker container, no cloud account — import it and use it.

import chromadb

# In-memory client — resets between runs (good for unit tests)
client = chromadb.Client()

# Persistent client — stores to disk (good for development)
client = chromadb.PersistentClient(path="./chroma_db")

collection = client.get_or_create_collection(name="my_docs")

# Add documents with pre-computed embeddings
collection.add(
    ids=["chunk_0", "chunk_1", "chunk_2"],
    embeddings=[[0.1, 0.2, ...], [0.3, 0.1, ...], [0.9, 0.1, ...]],
    documents=["Python is...", "Guido van Rossum...", "Weather is..."],
    metadatas=[{"source": "python_intro.txt"}, {"source": "python_intro.txt"}, {"source": "weather.txt"}],
)

# Query — returns top-3 nearest chunks
results = collection.query(
    query_embeddings=[[0.15, 0.18, ...]],
    n_results=3,
    include=["documents", "metadatas", "distances"],
)

for doc, meta, dist in zip(
    results["documents"][0],
    results["metadatas"][0],
    results["distances"][0],
):
    print(f"[{dist:.3f}] {meta['source']}: {doc[:80]}")

ChromaDB uses cosine similarity by default. Lower distance = more similar.

Choosing a Vector Database

Feature	ChromaDB	Pinecone	pgvector
Setup	Zero (in-process)	Managed cloud	Add extension to Postgres
Scale	Millions of vectors	Billions	Millions (with tuning)
Cost	Free	$70/month+ (managed)	Postgres hosting cost
Filtering	Metadata filters	Metadata filters	Full SQL WHERE clauses
Persistence	Local disk	Cloud-managed	Postgres storage
Best for	Local dev, prototypes	Production at scale	Teams already on Postgres

Start with ChromaDB locally — it removes all infrastructure friction during development, which is where you need to iterate fastest. I reach for pgvector over Pinecone when the team is already on Postgres: the operational overhead is near-zero and full SQL filtering eliminates an entire class of retrieval bugs that metadata-only filters can't handle.

Retrieval Strategies

Retrieval is not just "find the nearest vectors." Three strategies matter in practice.

Cosine Similarity Search

The default. Compute the cosine similarity between the query vector and every stored vector; return the top-k most similar chunks. Fast, well-understood, and works well when the query language matches the document language.

Maximal Marginal Relevance (MMR)

Standard top-k retrieval can return five chunks that all say the same thing — high similarity, low diversity. MMR trades some similarity for diversity: each additional chunk is selected to be both similar to the query and different from already-selected chunks.

ChromaDB does not implement MMR natively. Implement it post-retrieval:

def _mmr_rerank(
    query_embedding: list[float],
    candidate_embeddings: list[list[float]],
    candidate_docs: list[str],
    k: int = 5,
    lambda_: float = 0.5,
) -> list[str]:
    """Select top-k chunks using Maximal Marginal Relevance.

    Args:
        query_embedding: Embedded user query.
        candidate_embeddings: Embeddings of candidate chunks (over-fetch, e.g., top-20).
        candidate_docs: Text of each candidate chunk.
        k: Number of chunks to return.
        lambda_: Trade-off between relevance (1.0) and diversity (0.0).

    Returns:
        k chunks selected for both relevance and diversity.
    """
    import numpy as np

    q = np.array(query_embedding)
    cands = np.array(candidate_embeddings)

    # Cosine similarity: query vs candidates
    sim_to_query = (cands @ q) / (np.linalg.norm(cands, axis=1) * np.linalg.norm(q) + 1e-9)

    selected_indices: list[int] = []
    remaining = list(range(len(candidate_docs)))

    for _ in range(min(k, len(remaining))):
        if not selected_indices:
            # First pick: highest similarity to query
            best = max(remaining, key=lambda i: sim_to_query[i])
        else:
            # Subsequent picks: balance relevance vs redundancy
            selected_vecs = cands[selected_indices]
            scores = []
            for i in remaining:
                sim_q = sim_to_query[i]
                sim_selected = max(
                    (cands[i] @ selected_vecs[j]) /
                    (np.linalg.norm(cands[i]) * np.linalg.norm(selected_vecs[j]) + 1e-9)
                    for j in range(len(selected_vecs))
                )
                scores.append(lambda_ * sim_q - (1 - lambda_) * sim_selected)
            best = remaining[scores.index(max(scores))]

        selected_indices.append(best)
        remaining.remove(best)

    return [candidate_docs[i] for i in selected_indices]

Hybrid Search (BM25 + Semantic)

Semantic search handles paraphrase and synonymy well. BM25 (keyword search) handles exact terms — product codes, names, technical identifiers — better. Hybrid search runs both and combines scores (typically via Reciprocal Rank Fusion). Use hybrid when your documents contain precise identifiers that semantic search might miss. I default to MMR over top-k similarity the moment a domain has redundant content — policy documentation, API reference pages, and anything generated from a template will poison straight similarity retrieval with near-duplicate chunks every time.

Context Augmentation

Retrieved chunks are useful only if the prompt tells Claude how to use them. A weak prompt ("here are some documents, answer the question") produces weak answers. A well-structured prompt produces grounded, citable answers.

def _build_prompt(self, question: str, chunks: list[str]) -> str:
    """Format retrieved chunks and question into a grounded generation prompt.

    Args:
        question: The user's original question.
        chunks: Retrieved document chunks, ordered by relevance (most relevant first).

    Returns:
        A formatted prompt that instructs Claude to cite sources and acknowledge gaps.
    """
    context_block = "\n\n---\n\n".join(
        f"[Source {i + 1}]\n{chunk}" for i, chunk in enumerate(chunks)
    )

    return f"""You are a helpful assistant. Answer the question below using only the provided sources.

Rules:
- Cite which source(s) support each claim (e.g., "According to Source 2...").
- If the sources do not contain enough information to answer the question, say so explicitly.
- Do not use knowledge outside the provided sources.

Sources:
{context_block}

Question: {question}

Answer:"""

The three rules matter. Citing sources forces the model to ground claims in retrieved text rather than training data. Acknowledging gaps prevents confident-sounding hallucinations. Forbidding outside knowledge keeps the model from blending retrieved context with parametric knowledge unpredictably.

Generation with Claude

The query method ties everything together: embed the question, retrieve chunks, build the prompt, call Claude, and return the answer.

Diagram: Full query path from user question to Claude-generated answer.

import anthropic

client = anthropic.Anthropic()  # reads ANTHROPIC_API_KEY from environment

response = client.messages.create(
    model="claude-sonnet-4-6",
    max_tokens=1024,
    messages=[{"role": "user", "content": prompt}],
)

# Check why generation stopped
match response.stop_reason:
    case "end_turn":
        answer = response.content[0].text
    case "max_tokens":
        # Response was cut off — increase max_tokens or reduce prompt length
        answer = response.content[0].text + "\n\n[Response truncated — max_tokens reached]"
    case _:
        answer = f"[Generation stopped: {response.stop_reason}]"

Always check stop_reason. end_turn means the model finished naturally. max_tokens means the answer was cut off — a common silent failure when retrieved chunks are large and the combined prompt + answer exceeds your token budget.

For cheap classification tasks within a RAG system (e.g., query routing, intent detection, relevance filtering), use claude-haiku-4-5-20251001 instead of claude-sonnet-4-6. Haiku is significantly faster and cheaper for short-context classification where generation quality is less critical.

The `RAGPipeline` Class

The full implementation encapsulates all stages into a single class with three public methods: ingest, query, and the private helpers.

Diagram: RAGPipeline class structure — three public methods, three private helpers.

The constructor initializes all three dependencies. Chunk and embed are private because callers should not need to call them directly — they are implementation details of ingest and query.

Evaluation

A RAG pipeline with no evaluation is a guess. Add measurement before shipping.

Three metrics from the RAGAS framework cover the core failure modes:

Metric	What it measures	How it catches failures
Faithfulness	Is every claim in the answer supported by a retrieved chunk?	Catches hallucinations — the model added facts not in context
Answer Relevancy	Does the answer actually address the question?	Catches topic drift — the model answered a different question
Context Recall	Did retrieval surface the chunks needed to answer?	Catches retrieval failures — the right chunks weren't found

Run evaluation on a golden dataset — 20–50 question/answer pairs you've manually verified. Automate it in CI so a configuration change that breaks retrieval doesn't silently ship.

# RAGAS evaluation — requires: pip install ragas
from ragas import evaluate
from ragas.metrics import faithfulness, answer_relevancy, context_recall
from datasets import Dataset

eval_data = {
    "question": ["What is Python?", "Who created Python?"],
    "answer": ["Python is a programming language.", "Guido van Rossum."],
    "contexts": [
        ["Python is a high-level language..."],
        ["Guido van Rossum created Python in 1991..."],
    ],
    "ground_truth": ["Python is a high-level programming language.", "Guido van Rossum."],
}

result = evaluate(
    Dataset.from_dict(eval_data),
    metrics=[faithfulness, answer_relevancy, context_recall],
)
print(result)

Common Mistakes

Mistake 1: Chunks larger than 600 tokens drown the retrieval signal.
A 600-token chunk covers multiple topics. When you embed it, the vector averages over those topics and becomes less specific to any one of them. Similarity search returns chunks that are vaguely related to the query, not precisely relevant. Keep chunks under 400 tokens. If a passage requires more context, overlap consecutive chunks by 50–100 tokens rather than enlarging the chunk size.

Mistake 2: Not deduplicating similar chunks before storing.
If your corpus contains repeated passages (headers, boilerplate, legal disclaimers), those chunks will dominate retrieval — every query returns five variations of the same paragraph. Deduplicate before ingestion: compute a hash of each chunk's text and discard exact duplicates, then use a similarity threshold (cosine similarity > 0.95) to collapse near-duplicates. I've seen this sink a support-bot demo: the corpus was a product manual where the safety warning appeared verbatim on 40 of 200 pages — top-5 retrieval returned the warning for nearly every query, and the model dutifully answered "please refer to a qualified technician" regardless of what was asked.

Mistake 3: Embedding the raw query without expansion.
A user asking "how does Python handle memory?" may not use the same words as your documentation ("garbage collection", "reference counting", "memory management"). Query expansion generates alternative phrasings before embedding: original_query + hypothetical_answer_keywords. Hypothetical Document Embeddings (HyDE) generates a fake answer to the question and embeds that instead — the fake answer often matches document language better than the raw question.

Mistake 4: Skipping evaluation entirely.
Most RAG pipelines are shipped based on developer intuition ("the answers look right in testing"). Without a golden dataset and automated metrics, you won't know when a configuration change — a new chunk size, a different top-k, an updated embedding model — makes retrieval worse. Define your evaluation dataset before you tune parameters, not after. I've watched a team spend two weeks tuning chunk size upward because answers "felt" more complete — then discover via RAGAS that faithfulness had dropped from 0.89 to 0.71 because larger chunks were diluting the retrieved context with off-topic sentences.

Mistake 5: Using the same model for embedding and generation.
Anthropic's Claude models are generation models, not embedding models. Using a generation model's hidden states as embeddings produces inferior vectors compared to models trained specifically for semantic similarity (sentence-transformers, OpenAI's text-embedding-3-*, Voyage AI). Keep embedding and generation as separate concerns: sentence-transformers (or a dedicated embedding API) for vectors, Claude for generation.

Full Example

The complete implementation lives in two files:

src/pipeline.py — RAGPipeline class with all methods documented and implemented.
src/main.py — demo that ingests five documents about Python and runs three queries.

Setup:

python -m venv .venv
source .venv/bin/activate        # Windows: .venv\Scripts\activate
pip install -r requirements.txt

cp .env.example .env
# Add your ANTHROPIC_API_KEY to .env

python src/main.py

Expected output:

Ingested 5 documents → 12 chunks stored.

Q: What is Python's GIL?
A: According to Source 1, Python's Global Interpreter Lock (GIL) is a mutex that protects access to Python objects, preventing multiple threads from executing Python bytecodes simultaneously...

Q: Who created Python and when?
A: According to Source 3, Python was created by Guido van Rossum and first released in 1991...

Q: What is list comprehension?
A: According to Source 2, list comprehension is a concise syntax for creating lists based on existing iterables...

Full source: GitHub link → ai-integration/rag-pipeline/ — see README for setup steps.

Conclusion

RAG is the right architecture when your LLM needs access to private, large, or frequently updated data. The implementation is not what makes or breaks it — the parameters are: chunk size, overlap, retrieval strategy, and whether you bother to measure any of it. Most pipelines that fail in production fail because they were tuned by intuition and never measured. Add the RAGAS evaluation step before you ship anything to users, keep your chunks under 400 tokens, and you will avoid the failure modes that make RAG look unreliable. The architecture isn't fragile — the shortcuts are.

DEV Community

Building a RAG Pipeline from Scratch: Embeddings, Retrieval, and Claude

Key Points

Introduction

What RAG Is — and What It Replaces

RAG Pipeline Architecture

Chunking Strategies

Fixed-Size Chunking

Recursive Chunking

Semantic Chunking

Embeddings

Local Embeddings: `sentence-transformers`

API Embeddings

Vector Storage with ChromaDB

Choosing a Vector Database

Retrieval Strategies

Cosine Similarity Search

Maximal Marginal Relevance (MMR)

Hybrid Search (BM25 + Semantic)

Context Augmentation

Generation with Claude

The `RAGPipeline` Class

Evaluation

Common Mistakes

Full Example

Conclusion

Further Reading

Top comments (0)

Key Points

Introduction

What RAG Is — and What It Replaces

RAG Pipeline Architecture

Chunking Strategies

Fixed-Size Chunking

Recursive Chunking

Semantic Chunking

Embeddings

Local Embeddings: sentence-transformers

API Embeddings

Vector Storage with ChromaDB

Choosing a Vector Database

Retrieval Strategies

Cosine Similarity Search

Maximal Marginal Relevance (MMR)

Hybrid Search (BM25 + Semantic)

Context Augmentation

Generation with Claude

The RAGPipeline Class

Evaluation

Common Mistakes

Full Example

Conclusion

Further Reading

Local Embeddings: `sentence-transformers`

The `RAGPipeline` Class