Build a RAG Pipeline Step by Step in Python

Six lessons of theory. Time to build the thing. By the end of this page you'll have a complete, working RAG pipeline in about thirty lines of Python — ingestion, chunking, indexing, retrieval, a grounded prompt, citations and an honest refusal. Not pseudocode: the actual shape of the code running in production systems, minus the scale. Interviewers who say "code a RAG pipeline" want roughly this — not a framework import.

The two phases, one more time

Before any code, fix this in your head, because mixing them up is the most common architectural mistake in beginner RAG: indexing runs ONCE (and again when documents change); retrieval and generation run PER QUESTION. If your script embeds documents inside the function that answers a question, you've built something that re-reads the entire library for every visitor.

Step 1 — ingest, and check what you got

Loading documents is where the silent failures live. A scanned PDF returns empty text. A parser mangles tables. A file type is skipped entirely. Nothing errors; the document is simply invisible to your bot forever.

from pathlib import Path
from pypdf import PdfReader

def load_documents(folder: str):
    for path in Path(folder).glob("**/*.pdf"):
        text = "\n".join(page.extract_text() or "" for page in
                          PdfReader(path).pages)
        if len(text.strip()) < 200:                  # sanity check!
            print(f"WARNING: {path.name} produced almost no text "
                  f"({len(text)} chars) - scanned or unparsed?")
            continue
        yield {"source": path.name, "text": text}

WARNING: Leave_Policy_scan.pdf produced almost no text (14 chars) - scanned or unparsed?
loaded 47 documents

Step 2 — chunk (with the heading attached)

Straight from lesson 5: split on structure where it exists, fall back to size + overlap, and carry context into every chunk.

def chunk_document(doc, size=350, overlap=60):
    words = doc["text"].split()
    chunks, start = [], 0
    while start < len(words):
        piece = " ".join(words[start:start + size])
        chunks.append({
            "text": f"Document: {doc['source']}\n\n{piece}",  # enriched
            "source": doc["source"],
        })
        if start + size >= len(words):
            break
        start += size - overlap        # the overlap step
    return chunks

Step 3 — index, with stable ids

Deterministic ids are what make updates and deletes possible later (lesson 4's maintenance loop). Derive them from the source, never from a counter that shifts when a document changes.

import chromadb

db = chromadb.PersistentClient(path="./index")
col = db.get_or_create_collection("handbook")

def index_document(doc):
    chunks = chunk_document(doc)
    col.upsert(                                   # upsert, not add
        ids=[f"{doc['source']}::{i}" for i in range(len(chunks))],
        documents=[c["text"] for c in chunks],
        metadatas=[{"source": c["source"]} for c in chunks],
    )
    return len(chunks)

total = sum(index_document(d) for d in load_documents("./policies"))
print(f"indexed {total} chunks")

indexed 1,284 chunks

Step 4 — retrieve (and know when to give up)

Fetch the top few chunks — and check whether they're actually any good. The distance threshold is the difference between a system that admits ignorance and one that hallucinates on empty context.

def retrieve(question, k=4, max_distance=0.6):
    res = col.query(query_texts=[question], n_results=k,
                    include=["documents", "metadatas", "distances"])
    keep = [
        (doc, meta) for doc, meta, dist in zip(
            res["documents"][0], res["metadatas"][0], res["distances"][0])
        if dist <= max_distance                    # tune on your own data
    ]
    return keep

Step 5 — generate, grounded

The prompt carries five things (the prompt skeleton from the GenAI course): context, question, a grounding instruction, an escape hatch, and a citation requirement.

SYSTEM = """You are the HR assistant. Answer ONLY from the context.
Cite the source document in square brackets after each fact.
If the context does not contain the answer, reply exactly:
"I don't have that information - please contact HR." Never guess."""

def generate(question, hits):
    context = "\n\n---\n\n".join(
        f"[{meta['source']}]\n{doc}" for doc, meta in hits)
    res = client.chat.completions.create(
        model="gpt-4o-mini", temperature=0,
        messages=[
            {"role": "system", "content": SYSTEM},
            {"role": "user",
             "content": f"Context:\n{context}\n\nQuestion: {question}"},
        ],
    )
    return res.choices[0].message.content

The whole thing

def answer(question: str) -> str:
    hits = retrieve(question)
    if not hits:                                   # nothing relevant enough
        return "I don't have that information - please contact HR."
    return generate(question, hits)

print(answer("how many casual leaves do I get?"))
print(answer("what is the company's stance on time travel?"))

Employees get 12 casual leaves per calendar year. [Leave_Policy.pdf]
I don't have that information - please contact HR.

That's a real RAG system. Everything in lessons 8–12 — reranking, self-correction, evaluation, caching, scaling — is upgrades to these five functions, added when measurement says so.

Selection-round radar: if asked to code RAG on a whiteboard, write these five functions and narrate the two phases. Say the words upsert with stable ids, distance threshold, grounding instruction and refusal path — each one signals you've run this, not just read about it.

Wait — this works on 50 documents. Where does it break at 50,000?

Honest answer, in the order it happens. 1 · Retrieval quality goes first: with 50 documents almost anything retrieves fine; at 50,000 vocabulary gaps and identifier misses appear, and you need hybrid search plus reranking (lesson 6). 2 · Ingestion becomes a pipeline, not a script — incremental updates, deletes, failure alerts (lesson 10). 3 · Latency and cost become real: caching and model routing enter the picture. 4 · Trust: with real users you can no longer eyeball quality, so you need a golden set (lesson 9).

Notice what does not break: the five-step shape. The architecture you just wrote is the architecture that scales — it simply grows a stage at each step.

Common mistakes

• Re-indexing inside the answer function — the library re-read for every visitor.
• No ingestion sanity checks, so a scanned PDF is silently invisible forever.
• add() instead of upsert() with unstable ids — duplicates multiply on every run.
• No distance threshold, so empty retrieval becomes a confident hallucination.
• Source metadata never reaching the prompt — citations become impossible.
• Temperature left at default for a factual assistant.

Quick recap

Practice Zone — PYQs from real selection rounds

Six MCQs and two build tasks — write the pipeline from memory, then design the ingestion checks that keep it honest.

Q1Building RAGeasy

In a minimal RAG script, which step happens only ONCE (offline) rather than per question?

Asked in

aEmbedding the user's questionbLoading, chunking and embedding the documents into the indexcCalling the LLMdBuilding the augmented prompt

Q2Building RAGmedium

Your ingestion step reads PDFs. What quality problem should you check for BEFORE embedding anything?

Asked in

aWhether the PDFs are password protected onlybWhether text extraction actually worked — scanned/image PDFs yield empty or garbled text, and tables often extract as scrambled linescThe PDF version numberdNothing — extraction is always reliable

Q3Building RAGmedium

What belongs in the generation prompt of a well-built RAG system?

Asked in

aJust the question — the model will find the documents itselfbThe retrieved chunks, the question, an instruction to answer only from the context, a refusal rule for missing information, and a citation requirementcThe entire document corpusdThe embedding vectors as numbers

Q4Building RAGmedium

Where should the source filename and section be stored so answers can cite them?

Asked in

aNowhere — the LLM remembers where chunks came frombIn each chunk's metadata at indexing time, then passed into the prompt alongside the chunk textcIn a separate log file read after generationdIn the embedding vector itself

Q5Building RAGhard

Your prototype works on 50 documents but is unusably slow on 50,000. What is the most likely cause?

Asked in

aThe LLM got slowerbBrute-force similarity search over every vector — fine at 50 documents, fatal at scale; you need an ANN index (or a real vector database)cChunks became larger automaticallydThe embedding model degraded

Q6Building RAGmedium

Which is the safest way to handle 'retrieval returned nothing above the similarity threshold'?

Asked in

aSend the question to the LLM with no context and let it answer from memorybReturn a clear 'I don't have that information — here's how to reach support', optionally suggesting related documentscRetry with top-k = 100dReturn the closest chunk regardless of score

Hands-on tasks:

FAQ

Next lesson: what to add when the basic pipeline starts failing — Lesson 8: Advanced RAG →