Building a RAG Pipeline from Scratch - Every Line Explained

Most tutorials on Retrieval-Augmented Generation (RAG) skip the hard parts. They show you a five-line function using LangChain, call it a pipeline, and move on. But when you try to use that in production - with real documents, real users, real latency requirements - it falls apart.

This post takes the opposite approach. We will build every single component of a RAG pipeline from scratch, in raw Python, with no magic abstractions. By the time you finish, you will understand not just how RAG works, but why each stage exists, what tradeoffs it makes, and how to debug it when something breaks.

What you will learn: Document ingestion and text extraction · Semantic chunking strategies · Embedding generation and vector storage · Similarity search and retrieval · Context assembly and LLM generation · Query routing · Evaluation and production monitoring

1. The Problem RAG Solves

Large language models are trained on public internet data up to a fixed cutoff date. Ask ChatGPT about your company's internal policies, a recent acquisition, or a proprietary codebase - it cannot answer because it was never trained on that data.

You have two options: fine-tune the model on your data, or retrieve relevant documents at query time and feed them as context. Fine-tuning is expensive, slow, and brittle - retrain every time your data changes. RAG is cheap, instant, and always up to date.

For most real-world applications - document Q&A, customer support, internal knowledge bases - RAG is the right answer. Fine-tuning makes sense only when you need the model to learn a new skill (like writing in a specific style), not when you need it to know new facts.

2. The Full Pipeline: An Overview

A RAG pipeline has two phases: indexing (done once, ahead of time) and querying (done per user question). Here is the full flow:

Let us build each stage, one function at a time. Every function is independent and testable - you can swap any stage without touching the others.

3. Stage 1: Document Ingestion

We start by loading documents and extracting clean text. The goal is to go from arbitrary file formats (PDF, HTML, DOCX, Markdown) to a single string of clean text with no formatting noise.

import pymupdf4llm  # pip install pymupdf4llm
import re

def load_document(path: str) -> str:
    """Read a file and return clean text regardless of format."""
    if path.endswith(".pdf"):
        # pymupdf4llm converts PDFs to Markdown - preserves structure
        text = pymupdf4llm.to_markdown(path)
    elif path.endswith(".md") or path.endswith(".txt"):
        with open(path, "r") as f:
            text = f.read()
    elif path.endswith(".html"):
        from bs4 import BeautifulSoup
        with open(path, "r") as f:
            soup = BeautifulSoup(f.read(), "html.parser")
            # Remove script/style tags, get clean text
            for tag in soup(["script", "style"]):
                tag.decompose()
            text = soup.get_text(separator="\n")
    else:
        raise ValueError(f"Unsupported format: {path}")

    # Normalise whitespace - collapse multiple newlines
    text = re.sub(r"\n{3,}", "\n\n", text).strip()
    return text

The key insight: garbage in, garbage out. If your ingestion is noisy - headers, footers, page numbers mixed into the text - every downstream stage suffers. Always clean aggressively at ingestion time.

4. Stage 2: Chunking

LLMs have limited context windows (typically 4K-200K tokens). You cannot pass an entire book as context. You need to split documents into chunks that are small enough to fit in the context window but large enough to contain complete ideas.

The naive approach is fixed-size chunking - split every N tokens regardless of content. It works poorly because it cuts sentences in half. Instead, we use semantic chunking that respects natural boundaries:

from typing import List

def semantic_chunks(text: str, max_tokens: int = 512) -> List[str]:
    """Split text into chunks at paragraph boundaries, respecting max_tokens."""
    paragraphs = text.split("\n\n")
    chunks = []
    current = []

    for para in paragraphs:
        para_tokens = len(para.split())  # rough token count
        current_tokens = sum(len(c.split()) for c in current)

        if current_tokens + para_tokens > max_tokens and current:
            # Save current chunk and start a new one
            chunks.append("\n\n".join(current))
            current = [para]
        else:
            current.append(para)

    if current:
        chunks.append("\n\n".join(current))

    return chunks

Chunk size is a critical tuning parameter. Here is how different sizes affect retrieval:

• 256 tokens: High precision, low recall. Great for factual lookups (dates, names, numbers).
• 512 tokens: Balanced. The default for most use cases.
• 1024 tokens: High recall, lower precision. Good for summarisation or when context matters more than exact matches.

Always add overlap (50-100 tokens) between consecutive chunks so no context is lost at chunk boundaries:

def overlapping_chunks(text: str, chunk_size: int = 512, overlap: int = 64) -> List[str]:
    """Split into overlapping chunks. Each chunk shares 'overlap' tokens with the next."""
    tokens = text.split()
    chunks = []
    i = 0

    while i < len(tokens):
        chunk = tokens[i : i + chunk_size]
        chunks.append(" ".join(chunk))
        i += chunk_size - overlap  # slide forward by (chunk_size - overlap)

    return chunks

5. Stage 3: Embedding

An embedding is a vector of numbers (typically 768-1536 dimensions) that represents the meaning of a piece of text. The key property: similar texts have similar vectors, as measured by cosine similarity.

We use OpenAI's text-embedding-3-small for the best quality-per-cost ratio, but the interface is the same for any embedding model:

from openai import OpenAI

client = OpenAI()  # Reads OPENAI_API_KEY from env

def embed_text(text: str, model: str = "text-embedding-3-small") -> List[float]:
    """Convert a single text string into a vector embedding."""
    resp = client.embeddings.create(input=text, model=model)
    return resp.data[0].embedding

def embed_chunks(chunks: List[str]) -> List[dict]:
    """Embed all chunks and return them with metadata."""
    records = []
    for i, chunk in enumerate(chunks):
        vector = embed_text(chunk)
        records.append({
            "id": f"chunk-{i:05d}",
            "text": chunk,
            "vector": vector,
            "metadata": {"index": i, "length": len(chunk.split())},
        })
    return records

A couple of practical tips for embeddings:

• Batch your requests: The OpenAI API accepts up to 2048 inputs per call. Batching is 10-50x faster than one-at-a-time.
• Normalise vectors: Cosine similarity on normalised vectors = dot product. Store normalised vectors for faster search.
• Consider dimensionality: text-embedding-3-small defaults to 1536 dimensions, but you can use dimensions parameter to reduce to 256 for 4x cheaper storage with only ~2% quality loss.

def embed_batch(texts: List[str]) -> List[List[float]]:
    """Embed multiple texts in a single API call."""
    resp = client.embeddings.create(
        input=texts,
        model="text-embedding-3-small",
        dimensions=512  # smaller = cheaper, still accurate
    )
    return [d.embedding for d in resp.data]

6. Stage 4: Vector Storage and Search

Once we have embeddings, we need to store them in a way that supports fast similarity search. For production, use Pinecone, Qdrant, or Weaviate. For learning, we build a minimal vector store from scratch:

import numpy as np
from typing import List, Tuple

class SimpleVectorStore:
    """A minimal in-memory vector store with cosine similarity search."""

    def __init__(self):
        self.records = []  # list of dicts with id, text, vector

    def add(self, records: List[dict]):
        """Add records to the store."""
        self.records.extend(records)

    def similarity_search(
        self, query_vector: List[float], k: int = 5
    ) -> List[dict]:
        """Return the top-k most similar records by cosine similarity."""
        query = np.array(query_vector)
        similarities = []

        for record in self.records:
            vec = np.array(record["vector"])
            # Cosine similarity: dot product of normalised vectors
            score = np.dot(query, vec) / (
                np.linalg.norm(query) * np.linalg.norm(vec)
            )
            similarities.append((score, record))

        # Sort descending by score
        similarities.sort(key=lambda x: x[0], reverse=True)
        return [{"score": s, **r} for s, r in similarities[:k]]


# Usage
store = SimpleVectorStore()
store.add(embedded_chunks)

query_vec = embed_text("What is our refund policy?")
results = store.similarity_search(query_vec, k=3)

for r in results:
    print(f"Score: {r['score']:.3f} | {r['text'][:80]}...")

7. Stage 5: Generation

The final stage takes the retrieved chunks and the user's question, assembles them into a prompt, and calls the LLM to generate an answer:

SYSTEM_PROMPT = """You are a helpful assistant. Answer the user's question
based ONLY on the provided context. If the context does not contain the
answer, say "I don't have enough information to answer that." Do NOT make
up information. Cite the relevant chunks when possible."""

def generate(
    question: str, context_chunks: List[dict], model: str = "gpt-4o"
) -> str:
    """Generate an answer using retrieved context."""
    # Assemble context
    context_text = "\n\n".join(
        f"[Chunk {r['id']}]:\n{r['text']}"
        for r in context_chunks
    )

    # Build messages
    user_prompt = f"Context:\n{context_text}\n\nQuestion: {question}"

    response = client.chat.completions.create(
        model=model,
        messages=[
            {"role": "system", "content": SYSTEM_PROMPT},
            {"role": "user", "content": user_prompt},
        ],
        temperature=0.1,  # low temperature for factual answers
    )

    return response.choices[0].message.content

A few important details about the generation stage:

• Low temperature (0.1): We want factual answers, not creative ones. Higher temperatures increase hallucination risk.
• System prompt with constraints: Explicit instructions to refuse when context is insufficient. This prevents the model from guessing.
• Chunk tagging: Label each chunk with its ID in the prompt so the model can reference sources. This enables provenance.

8. Putting It All Together: The Full Pipeline

Here is every stage composed into a single, runnable pipeline:

class RAGPipeline:
    """A complete RAG pipeline from ingestion to generation."""

    def __init__(self):
        self.vector_store = SimpleVectorStore()

    def index_document(self, path: str):
        """Ingest, chunk, embed, and store a document."""
        print(f"  Loading: {path}")
        text = load_document(path)

        print(f"  Chunking: {len(text.split())} tokens")
        chunks = semantic_chunks(text, max_tokens=512)

        print(f"  Embedding: {len(chunks)} chunks")
        records = embed_chunks(chunks)

        print(f"  Storing: {len(records)} vectors")
        self.vector_store.add(records)

        print(f"  Done. ({len(records)} chunks indexed)")

    def query(self, question: str, k: int = 5) -> dict:
        """Ask a question against indexed documents."""
        print(f"  Embedding query...")
        query_vec = embed_text(question)

        print(f"  Searching top-{k}...")
        results = self.vector_store.similarity_search(query_vec, k=k)

        print(f"  Generating answer...")
        answer = generate(question, results)

        return {
            "answer": answer,
            "sources": [r["id"] for r in results],
            "scores": [r["score"] for r in results],
        }


# ── Usage ──
pipeline = RAGPipeline()
pipeline.index_document("employee-handbook.pdf")
pipeline.index_document("company-policies.md")

result = pipeline.query("What is our remote work policy?")
print(result["answer"])
print(f"Sources: {result['sources']}")

9. Advanced: Query Routing

A simple RAG pipeline always retrieves everything. But not all questions need retrieval. Adding a query router lets you handle different question types differently:

from enum import Enum

class QueryType(Enum):
    FACTUAL = "factual"       # needs retrieval
    GENERAL = "general"       # LLM can answer from training data
    SUMMARIZE = "summarize"   # needs broad context retrieval
    UNKNOWN = "unknown"       # refuse politely

def classify_question(question: str) -> QueryType:
    """Use a lightweight classifier to route the question."""
    factual_keywords = ["policy", "procedure", "when", "where",
                        "who", "how many", "what is the"]
    summary_keywords = ["summarize", "overview", "tl;dr", "brief"]

    q_lower = question.lower()

    for kw in summary_keywords:
        if kw in q_lower:
            return QueryType.SUMMARIZE

    for kw in factual_keywords:
        if kw in q_lower:
            return QueryType.FACTUAL

    return QueryType.GENERAL


def routed_query(pipeline: RAGPipeline, question: str) -> str:
    """Route question to the right handler based on type."""
    qtype = classify_question(question)

    if qtype == QueryType.FACTUAL:
        # Standard RAG - search for specific chunks
        result = pipeline.query(question, k=3)
        return result["answer"]

    elif qtype == QueryType.SUMMARIZE:
        # Broad retrieval - get many chunks, then summarise
        result = pipeline.query(question, k=10)
        summary_prompt = f"Summarise these documents:\n\n{result['answer']}"
        return client.chat.completions.create(...)

    elif qtype == QueryType.GENERAL:
        # No retrieval needed - LLM can answer directly
        return client.chat.completions.create(
            model="gpt-4o-mini",
            messages=[{"role": "user", "content": question}],
        )

    else:
        return "I can only answer questions about company documents."

10. Production Concerns: What We Did Not Cover

The pipeline above works on your laptop. Moving it to production means adding:

• Persistent vector storage: Use Qdrant (open source) or Pinecone (managed) instead of the in-memory store. Both have Python SDKs and handle millions of vectors.
• Reranking: Add a cross-encoder reranker (e.g., cross-encoder/ms-marco-MiniLM-L-6-v2) between retrieval and generation to re-rank the top-20 results by relevance. This improves answer quality by 10-20%.
• Hybrid search: Combine vector similarity with keyword search (BM25) to catch exact phrase matches that embeddings sometimes miss.
• Caching: Cache embeddings for frequently asked questions. Cache LLM responses for identical questions.
• Evaluation: Measure retrieval precision (is the right chunk in the top-5?) and answer accuracy (does the LLM use the retrieved context correctly?). Tools like ragas can help.

# Reranking example with sentence-transformers
from sentence_transformers import CrossEncoder

reranker = CrossEncoder("cross-encoder/ms-marco-MiniLM-L-6-v2")

def rerank(query: str, candidates: list, k: int = 5) -> list:
    """Rerank candidates by query relevance."""
    pairs = [(query, c["text"]) for c in candidates]
    scores = reranker.predict(pairs)
    ranked = sorted(
        zip(candidates, scores),
        key=lambda x: x[1],
        reverse=True
    )
    return [r[0] for r in ranked[:k]]

11. Summary: What You Built and What Comes Next

Your next challenge

# TODO: Add evaluation to your RAG pipeline
#
# 1. Create a test set of 20 questions with known answers
# 2. Measure retrieval precision@k (is the right chunk retrieved?)
# 3. Measure answer accuracy (does the LLM give the right answer?)
# 4. Add a reranker and compare before/after scores
# 5. Add hybrid search (vector + keyword) and compare again

def evaluate(pipeline, test_set):
    for question, expected_answer in test_set:
        result = pipeline.query(question)
        # Compare result["answer"] to expected_answer
        # Calculate precision, recall, F1
        pass

Production RAG is 20% building and 80% measuring. Start measuring now, and you will know exactly where to improve when things go wrong.