Building Your First RAG Pipeline from Scratch

Building Your First RAG Pipeline from Scratch

Most RAG tutorials hand you a framework, five lines of code, and a demo that collapses the first time a real user asks something phrased differently than the docs. The pipeline is not actually hard; it is six components and a handful of decisions that frameworks hide from you until they break. This article builds the entire thing from raw Python: load documents, chunk them, embed them, store vectors, retrieve at query time, generate grounded answers with citations. Then we measure whether it works, because an unevaluated RAG system is just a chatbot with extra steps.

What we are building: question answering over a folder of markdown files, returning answers that cite source filenames and admit when the documents do not contain an answer. No frameworks. OpenAI SDK for embeddings and generation, numpy for similarity, JSON for persistence.

The Six Components#

Every RAG system, from a weekend prototype to an enterprise deployment, decomposes into the same parts. Ingestion runs offline: a loader reads documents, a chunker splits them into retrievable units, an embedder converts each chunk to a vector, and a store persists vectors alongside their text and metadata. Query time runs live: the retriever finds the nearest stored chunks for the incoming question, and the generator writes an answer constrained to those chunks.

Keep these boundaries clean even in throwaway code. Every later upgrade (a real vector database, hybrid search, a reranker) replaces exactly one box.

Ingestion: Load and Chunk#

The chunker is where quality begins. Documents get split into overlapping windows on paragraph boundaries, because embeddings pool meaning per chunk, and a chunk cut mid-sentence embeds as garbage:

from pathlib import Path

def chunk_text(text: str, size: int = 1200, overlap: int = 150) -> list[str]:
    paragraphs = [p.strip() for p in text.split("\n\n") if p.strip()]
    chunks, current = [], ""
    for para in paragraphs:
        if len(current) + len(para) > size and current:
            chunks.append(current.strip())
            current = current[-overlap:]          # carry tail context forward
        current += para + "\n\n"
    if current.strip():
        chunks.append(current.strip())
    return chunks

The overlap line deserves attention: carrying roughly the last paragraph of the previous chunk into the next one means a fact sitting at a boundary appears fully inside at least one retrievable unit. Size 1200 characters is a reasonable starting point for prose; the tuning discussion comes after evaluation exists.

Loading walks the directory and pairs every chunk with metadata naming its origin:

def load_documents(docs_dir: str) -> list[dict]:
    items = []
    for path in sorted(Path(docs_dir).glob("**/*.md")):
        text = path.read_text(encoding="utf-8", errors="ignore")
        for i, chunk in enumerate(chunk_text(text)):
            items.append({
                "source": path.name,
                "chunk_index": i,
                "text": chunk,
            })
    return items

Indexing: Embed and Persist#

Batch embedding keeps costs and latency sane; the API charges per token regardless, but batches avoid per-call overhead:

import numpy as np
import json
from openai import OpenAI

client = OpenAI()

def embed_batch(texts: list[str]) -> np.ndarray:
    resp = client.embeddings.create(model="text-embedding-3-small", input=texts)
    vecs = np.array([d.embedding for d in resp.data], dtype=np.float32)
    return vecs / np.linalg.norm(vecs, axis=1, keepdims=True)   # normalize once

def build_index(items: list[dict], out_dir: str = "./index") -> None:
    vecs = embed_batch([it["text"] for it in items])
    Path(out_dir).mkdir(exist_ok=True)
    np.save(f"{out_dir}/vectors.npy", vecs)
    with open(f"{out_dir}/chunks.json", "w") as f:
        json.dump(items, f)

if __name__ == "__main__":
    build_index(load_documents("./docs"))

Normalizing at write time makes cosine similarity a plain dot product forever after, which numpy computes in one matrix multiply over the whole corpus. Persistence here is deliberately boring: one .npy of float32 vectors plus one .json of chunk records. At ten thousand chunks this costs about 74 megabytes and loads instantly. The honest upgrade trigger is different: swap in a vector database when you need filtered queries, deletes that stick, or scale past what RAM holds comfortably, not before.

Query Time: Retrieve#

Retrieval mirrors indexing: embed the question with the same model, dot against every stored vector, take the top k:

class Retriever:
    def __init__(self, index_dir: str = "./index"):
        self.vecs = np.load(f"{index_dir}/vectors.npy")
        self.chunks = json.load(open(f"{index_dir}/chunks.json"))

    def search(self, question: str, k: int = 4) -> list[dict]:
        qvec = embed_batch([question])[0]
        scores = self.vecs @ qvec
        top = np.argsort(scores)[-k:][::-1]
        hits = [{**self.chunks[i], "score": float(scores[i])} for i in top]
        return self._dedupe(hits)

    def _dedupe(self, hits: list[dict]) -> list[dict]:
        seen, unique = set(), []
        for h in hits:                       # overlapping neighbors waste slots
            key = (h["source"], h["chunk_index"])
            if key not in seen:
                seen.add(key)
                unique.append(h)
        return unique[:4]

Two judgment calls live in this function. Top-k of four balances coverage against context dilution: more retrieved chunks means more chances the answer is present, but also more filler the generator must wade through, and models attend least reliably to the middle of long contexts, a failure mode documented by Liu and colleagues in 2023 as "lost in the middle." And a minimum-score threshold is worth adding (if best_score < 0.25: return []) so irrelevant questions produce an explicit "not found" instead of confident nonsense built from unrelated chunks.

Generation With Grounding#

The prompt does three jobs: fence the model to retrieved evidence, force citations, and authorize refusal:

SYSTEM = """Answer the user's question using ONLY the numbered document excerpts.
Cite sources inline like [1] or [2].
If the excerpts do not contain the answer, reply exactly:
"I don't have information about that in the documentation."
Do not use outside knowledge."""

def answer(question: str) -> str:
    hits = Retriever().search(question)
    if not hits:
        return "I don't have information about that in the documentation."
    context = "\n\n".join(
        f"[{i+1}] ({h['source']})\n{h['text']}" for i, h in enumerate(hits)
    )
    resp = client.chat.completions.create(
        model="gpt-4o-mini",
        temperature=0.2,
        messages=[
            {"role": "system", "content": SYSTEM},
            {"role": "user", "content": f"Excerpts:\n{context}\n\nQuestion: {question}"},
        ],
    )
    return resp.choices[0].message.content or ""

Low temperature matters here: this is extraction-adjacent work, not creative writing, and every degree of sampling entropy is a chance to drift off-evidence.

Citations deserve their own validation pass, because models occasionally cite an excerpt number that does not exist or skip citations entirely:

import re

CITE = re.compile(r"\[(\d+)\]")

def citations_valid(answer: str, n_sources: int) -> bool:
    used = {int(m) for m in CITE.findall(answer)}
    return used.issubset(set(range(1, n_sources + 1))) and bool(used)

An answer failing that check gets one regeneration attempt with a sterner instruction, then falls back to displaying the raw excerpts with their sources, which is honest and still useful. Cheap guards like this are why grounded systems keep user trust: every claim is traceable, or the system visibly declines to make one.

One operational note belongs beside retrieval: indexes rot. Documents change while your vectors stay frozen, and the first symptom is confidently cited answers quoting a policy that was deleted last week. A crude but effective defense hashes source files and rebuilds when the hash moves:

import hashlib

def corpus_digest(docs_dir: str) -> str:
    h = hashlib.sha256()
    for path in sorted(Path(docs_dir).rglob("*.md")):
        h.update(path.name.encode())
        h.update(path.read_bytes())
    return h.hexdigest()

Compare against the stored digest on every startup or cron run; rebuild only when it differs. Thirty lines of laziness here beats any amount of apology emails later.

Evaluating Before Tuning#

Here is the step most first pipelines skip, and it is the only one that turns opinions into engineering. Build twenty to fifty questions with known correct sources before changing any parameter:

GOLDEN = [
    {"q": "How many vacation days can I carry over?",   "source": "handbook.md"},
    {"q": "What is the refund window?",                 "source": "refund-policy.md"},
    # ... enough to cover every doc and several phrasings
]

def hit_rate(retriever, k: int = 4) -> float:
    hits = sum(
        1 for g in GOLDEN
        if any(h["source"] == g["source"]
               for h in retriever.search(g["q"], k=k))
    )
    return hits / len(GOLDEN)

print(f"retrieval hit@4: {hit_rate(Retriever()):.0%}")

That single number drives every subsequent decision. Hit rate low but nonzero? Chunking or embedding choice is suspect. Specific questions always missing? Inspect what got retrieved instead of guessing; usually the answer lives across a boundary your chunker cut badly. For answer quality beyond retrieval, spot-check generations against sources manually at this stage; automated LLM grading is worth adding once volume justifies it.

Break the metric down by document category and the diagnosis writes itself:

In runs like this, the weak category usually shares a root cause: long tutorials get shredded by fixed cuts, and API references lose tables that were split from their headers. Category-level numbers tell you where to aim the fix; a single aggregate number only tells you that something hurts.

Where First Versions Fail#

The recurring failure catalog, roughly in order of frequency. Boundary cuts destroy facts that span two chunks; overlap mitigates but structure-aware splitting fixes properly. Top-k set by vibes rather than measured against hit rate; four is a start, not a law. No refusal threshold, so the system confidently answers questions about topics absent from the corpus by hallucinating around nearest-neighbor noise. Stale indexes after documents change, producing authoritative citations of deleted policies; rebuild on change events even if crudely. Tables, code blocks, and lists treated as prose and shredded by paragraph splitting; special-case them during loading. And cost surprises: each query pays one embedding call plus one generation call whose input includes all retrieved text, so token math belongs in the design, not the invoice review.

A final note before shipping: version your index alongside your code. The digest check above tells you when documents changed; recording which chunker version and embedding model produced each index file tells you why two environments behave differently three weeks from now. One JSON manifest with model name, chunk size, and creation date costs nothing and answers the question every team eventually asks, which is why this index answers differently than that one.

Ship the Ugly Version. The pipeline above is maybe ninety lines, has no dependencies beyond one SDK and numpy, and already delivers grounded, cited answers with a measurable quality number. That is further than most teams get while evaluating frameworks. Ship this, collect real questions, watch the hit rate, and let the failures tell you which component to upgrade first: a reranker when retrieval returns right-topic-wrong-chunk, hybrid lexical search when exact terms miss, a vector database when filters and scale demand it. Each upgrade slots into one clean boundary you preserved. The engineers who struggle are the ones who start at the destination architecture without ever measuring the simple version, and therefore cannot tell whether any of it helped.