RAG Tutorial in Python, Build a Retrieval-Augmented Generation Pipeline From Scratch

By Pooja Goenka · Published 2026-10-17 · Updated 2026-10-21

Retrieval-augmented generation (RAG) is a pattern where your code finds the passages in your own documents that best match a question, pastes them into the prompt, and asks a language model to answer from them. The steps are: load documents, split them into chunks, turn each chunk into a vector, store the vectors, retrieve the top-k closest chunks for a question, build a prompt around them, and generate the answer.

I am going to build all of it in about 130 lines of standard-library Python, run it, and show you the real output. It needs no API key and no pip install. The embedding is a hand-written TF-IDF and the "LLM" is a stub, because I want you to see every moving part before a library hides it. Further down, in clearly marked snippets, I show how to swap in a real embedding model and a real LLM call. If you want the background on why this pattern exists at all, why AI invents sources covers the problem it solves, and RAG vs fine-tuning covers when to pick it.

What do you need to follow this RAG tutorial?

Python 3 and a text editor. I ran everything below with Python 3.14, using the isolated mode flag (python3 -I rag.py) so no stray packages could sneak in. The corpus is three tiny support documents about refunds, shipping and accounts, written inline so the script is one file.

The stub generator is deliberate. A deterministic generator makes it easy to tell whether retrieval or generation caused a wrong answer, which I come back to in the debugging section.

Step 1: how do you load documents for a RAG pipeline?

Here the loading is a dictionary of filename to text. In a real project it is a loop over .txt, .md or PDF files, and the part that matters is the same: keep the source name next to the text. Every chunk we create later carries its source, because an answer you cannot trace back is the same problem RAG was supposed to fix.

import math
import re
from collections import Counter

# 1. Load documents (inline here so the script runs anywhere)
DOCS = {
    "refunds.txt": (
        "Refund policy. You can request a refund within 14 days of purchase. "
        "Refunds go back to the original payment method and take 5 to 7 business days. "
        "Digital downloads that have already been opened cannot be refunded. "
        "To start a refund, email the support team with your order number."
    ),
    "shipping.txt": (
        "Shipping times. Standard shipping takes 4 to 6 business days inside the country. "
        "Express shipping takes 1 to 2 business days and costs extra. "
        "International orders ship in 10 to 15 business days and customs fees are paid by the buyer. "
        "Every order gets a tracking link by email once it leaves the warehouse."
    ),
    "accounts.txt": (
        "Account help. To reset your password, use the forgot password link on the login page. "
        "The reset link expires after 30 minutes. "
        "You can delete your account from the privacy tab in settings. "
        "Deleting an account removes saved orders and cannot be undone."
    ),
}

Step 2: how big should RAG chunks be?

A chunk is the unit you retrieve, so it decides what the model gets to read. This version splits by words into windows of size words, and each window starts size - overlap words after the last one, so neighbouring chunks share a few words.

# 2. Chunk: fixed word windows with overlap
def chunk(text, source, size=25, overlap=8):
    words = text.split()
    out, start, n = [], 0, 0
    while start < len(words):
        piece = " ".join(words[start:start + size])
        out.append({"id": f"{source}#{n}", "source": source, "text": piece})
        n += 1
        if start + size >= len(words):
            break
        start += size - overlap
    return out

I use 25 words with an 8-word overlap only because my documents are tiny. For real documents, a few hundred words is a common starting range, and I would tune it against the eval below rather than trust any rule of thumb, mine included.

Step 3: what is an embedding, and can you build one by hand?

An embedding turns text into a list of numbers so that similar texts end up close together. Real embedding models do this with a neural network and capture meaning. Mine does something much cruder: it counts words, weights rare words higher (that is the IDF in TF-IDF), and scales each vector to length one. Similarity is then the cosine between two vectors, which for unit vectors is just a dot product.

# 3. Embed: hand-written TF-IDF vectors (sparse dicts)
def tokenize(text):
    return re.findall(r"[a-z0-9]+", text.lower())

def build_idf(chunks):
    n = len(chunks)
    df = Counter()
    for c in chunks:
        df.update(set(tokenize(c["text"])))
    return {t: math.log((1 + n) / (1 + d)) + 1 for t, d in df.items()}

def embed(text, idf):
    tf = Counter(t for t in tokenize(text) if t in idf)
    vec = {t: count * idf[t] for t, count in tf.items()}
    norm = math.sqrt(sum(v * v for v in vec.values())) or 1.0
    return {t: v / norm for t, v in vec.items()}

def cosine(a, b):
    return sum(v * b.get(t, 0.0) for t, v in a.items())

The limit of this approach is that it only sees shared words and has no sense of meaning. "Money back" and "refund" share nothing, so a question phrased with one will miss a document written with the other. That gap is exactly what a real embedding model closes, and it is the first thing I would swap.

Step 4: where do you store the vectors?

For a few hundred chunks, a Python list of (chunk, vector) pairs is a perfectly good vector store. A vector database earns its place when the list stops fitting in memory or you need filtering and updates at scale. Beginners often install a database before the pipeline works, and then cannot tell which part is broken.

# 4. Store: a plain list of (chunk, vector)
def build_index(docs, **kw):
    chunks = [c for name, text in docs.items() for c in chunk(text, name, **kw)]
    idf = build_idf(chunks)
    store = [(c, embed(c["text"], idf)) for c in chunks]
    return store, idf

Step 5: how does top-k retrieval work?

Embed the question with the same function and the same vocabulary, score it against every stored chunk, sort, and keep the best k. That is the whole retriever.

# 5. Retrieve top-k
def retrieve(question, store, idf, k=2):
    q = embed(question, idf)
    scored = [(cosine(q, vec), c) for c, vec in store]
    scored.sort(key=lambda s: s[0], reverse=True)
    return scored[:k]

Step 6 and 7: build the prompt and generate

The prompt has three jobs: hand the model the retrieved text, tell it to stay inside that text, and ask it to cite the chunk id. The stub generate() picks the sentence in the context that shares the most words with the question. It is not eloquent, but it can only say things that are in the retrieved chunks, which makes retrieval problems impossible to hide.

# 6. Build the prompt
def build_prompt(question, hits):
    context = "\n".join(f"[{c['id']}] {c['text']}" for _, c in hits)
    return (
        "Answer using only the context below. If the answer is not there, "
        "say you do not know. Cite the chunk id in brackets.\n\n"
        f"Context:\n{context}\n\nQuestion: {question}\nAnswer:"
    )

# 7. Generate: a stub that returns the best-matching sentence from the top chunk
def generate(prompt):
    question = prompt.split("Question: ")[-1].split("\nAnswer:")[0]
    qtok = set(tokenize(question))
    best, best_overlap, best_id = None, 0, None
    for line in prompt.split("Context:\n")[1].split("\n\nQuestion:")[0].split("\n"):
        cid, _, body = line.partition("] ")
        for sent in re.split(r"(?<=\.)\s+", body):
            overlap = len(qtok & set(tokenize(sent)))
            if overlap > best_overlap:
                best, best_overlap, best_id = sent, overlap, cid[1:]
    if best is None:
        return "I do not know."
    return f"{best} [{best_id}]"

def answer(question, store, idf, k=2):
    hits = retrieve(question, store, idf, k)
    return generate(build_prompt(question, hits)), hits

Step 8: how do you evaluate a RAG pipeline with five questions?

Write five questions whose answers you know, and for each one note a phrase that must appear in a retrieved chunk. Then measure how often the right chunk lands in the top-k. This is a retrieval hit rate, and it is the cheapest useful metric you can build. Five questions written from your own documents will find more bugs than none.

# 8. Tiny eval: each question lists the phrase that must appear in a retrieved chunk
EVAL = [
    ("How many days do I have to ask for a refund?", "within 14 days"),
    ("How long does express shipping take?", "1 to 2 business days"),
    ("Does the password reset link expire?", "expires after 30 minutes"),
    ("Who pays customs fees on international orders?", "paid by the buyer"),
    ("Can I get my money back for an opened digital download?", "already been opened cannot be refunded"),
]

def run_eval(label, size, overlap, k):
    store, idf = build_index(DOCS, size=size, overlap=overlap)
    ok = 0
    for q, needle in EVAL:
        hits = retrieve(q, store, idf, k)
        hit = any(needle in c["text"] for _, c in hits)
        ok += hit
        print(f"  {'PASS' if hit else 'FAIL'}  {q}")
    print(f"{label}: size={size} overlap={overlap} k={k} -> {ok}/{len(EVAL)}  ({len(store)} chunks)\n")

if __name__ == "__main__":
    store, idf = build_index(DOCS)
    print(f"{len(store)} chunks indexed\n")
    q = "How long do refunds take to arrive?"
    ans, hits = answer(q, store, idf)
    print("Question:", q)
    for score, c in hits:
        print(f"  {score:.3f}  {c['id']}")
    print("Answer:", ans, "\n")
    run_eval("baseline", size=25, overlap=8, k=2)
    run_eval("top-k of 1", size=25, overlap=8, k=1)
    run_eval("tiny chunks, no overlap", size=6, overlap=0, k=1)

Save the whole thing as rag.py and run python3 -I rag.py. This is the real output from my run:

9 chunks indexed

Question: How long do refunds take to arrive?
  0.368  refunds.txt#0
  0.214  refunds.txt#1
Answer: Refunds go back to the original payment method and take 5 to 7 [refunds.txt#0] 

  PASS  How many days do I have to ask for a refund?
  PASS  How long does express shipping take?
  PASS  Does the password reset link expire?
  PASS  Who pays customs fees on international orders?
  PASS  Can I get my money back for an opened digital download?
baseline: size=25 overlap=8 k=2 -> 5/5  (9 chunks)

  FAIL  How many days do I have to ask for a refund?
  PASS  How long does express shipping take?
  PASS  Does the password reset link expire?
  PASS  Who pays customs fees on international orders?
  PASS  Can I get my money back for an opened digital download?
top-k of 1: size=25 overlap=8 k=1 -> 4/5  (9 chunks)

  FAIL  How many days do I have to ask for a refund?
  FAIL  How long does express shipping take?
  FAIL  Does the password reset link expire?
  FAIL  Who pays customs fees on international orders?
  FAIL  Can I get my money back for an opened digital download?
tiny chunks, no overlap: size=6 overlap=0 k=1 -> 0/5  (26 chunks)

Read the answer line closely. The stub returned "take 5 to 7" and stopped, because the chunk boundary fell in the middle of the sentence. That is a chunking bug showing up in the final answer, and it is worth noticing that the retrieval scores looked fine. Then look at the three eval runs. The baseline finds all five. Dropping top-k to 1 loses one question, and shrinking the chunks to six words with no overlap loses every one.

What are the common RAG failure modes?

These are the four I hit most, and the eval above reproduces three of them.

Chunks that are too small strip out the context an answer needs. At six words, "within 14 days" and the word "refund" end up in different chunks, so nothing matches well. Chunks that are too large do the opposite: one vector has to stand for many topics, scores flatten, and the prompt fills with text the model does not need.

Chunking with no overlap cuts sentences in half. The size=6, overlap=0 run is the extreme case, but the "take 5 to 7" answer above happened with a sensible setting, just at the unlucky spot where a window ended.

A top-k that is too small drops the chunk you needed. When I ran the first eval question with k=1, the top result was refunds.txt#1 (score 0.431), a chunk that repeats the word "refund" several times. The chunk with "within 14 days" scored 0.347 and came second. A larger k would have kept it. The cost of a large k is a longer prompt and more noise, so raise it until the eval stops improving, not further.

An answer that is not grounded is the generation-side failure. The model sees the right chunks and still adds something that is not in them, or ignores them and answers from memory. The instruction "if the answer is not there, say you do not know" helps, and so does asking for a citation you can check.

How do you debug a RAG pipeline: retrieval or generation?

Split the pipeline at the prompt. Print the retrieved chunks for a failing question and read them yourself. If the needed text is not in them, retrieval failed, and the fixes are chunk size, overlap, top-k, or a better embedding. If the text is there and the answer is still wrong, generation failed, and the fixes are in the prompt, the model, or how many chunks you stuffed in.

I do this before touching anything else, because the two failures feel identical from the outside (a wrong answer) and have unrelated fixes. People change the prompt for an hour when the right chunk was never retrieved. The stub generator makes the split obvious: it cannot invent anything, so any wrong answer from this script is a retrieval problem by construction.

How do you swap in a real embedding model? (not run here)

The snippets in this section were not executed for this post, because they need an API key. I checked the method names against OpenAI's embeddings guide, which documents client.embeddings.create with the text-embedding-3-small model and the vector at response.data[0].embedding. I embed one string per call to stay inside what the guide shows.

# NOT RUN HERE: needs `pip install openai` and an OPENAI_API_KEY
from openai import OpenAI

client = OpenAI()

def embed_real(text):
    response = client.embeddings.create(
        input=text, model="text-embedding-3-small"
    )
    return response.data[0].embedding   # a list of floats

def cosine_dense(a, b):
    dot = sum(x * y for x, y in zip(a, b))
    na = sum(x * x for x in a) ** 0.5
    nb = sum(y * y for y in b) ** 0.5
    return dot / (na * nb)

To use it, store embed_real(chunk["text"]) instead of the TF-IDF dictionary, embed the question the same way, and score with cosine_dense. Nothing else in the pipeline changes, and your eval tells you whether the swap helped.

How do you swap in a real LLM for generate()? (not run here)

Also not executed here. The call below follows the Responses API shown in OpenAI's text generation guide, where you call client.responses.create and read the answer from response.output_text. I left the model name as a placeholder, because current model names change and the guide's models page is the place to pick one.

# NOT RUN HERE: needs `pip install openai` and an OPENAI_API_KEY
from openai import OpenAI

client = OpenAI()
MODEL = "pick-a-model-from-the-models-page"

def generate_real(prompt):
    response = client.responses.create(model=MODEL, input=prompt)
    return response.output_text

Because answer() only calls generate(prompt), you can pass this function in instead of the stub and keep the rest of the script as it is.

Where does this go next?

A pipeline like this answers from one retrieval pass over plain chunks. Once it works, the improvements are about retrieval quality: combining keyword and vector search, reranking the candidates, and splitting documents along their real structure. In the LogicWiz GenAI course, the chapter "Nova Reads the Archives" builds RAG step by step, and "Finding the Needle" covers hybrid search and reranking. The course is completely free at logicwiz.ai/genai, and the Zero to AI Agent line shows the whole path in one view. To compare it with other free options first, read the best free GenAI courses in 2026, compared.

If your next question is what to do with retrieval once it works, the usual answer is to hand it to an agent as a tool, which is the subject of the AI agents roadmap for beginners.

FAQ

What is RAG in simple terms?

RAG means looking up relevant text from your own documents first, then giving that text to a language model along with the question so it answers from it. The model's own memory plays a smaller part, which makes answers easier to check.

Do I need a vector database to build a RAG pipeline?

No. A list of vectors and a loop that scores them works for hundreds or a few thousand chunks, as this tutorial shows. Reach for a vector database when memory, speed of search, filtering or live updates become real problems.

What chunk size should I use for RAG?

There is no universal number. Start with a few hundred words and some overlap, write a handful of questions with known answers, and compare hit rates across sizes.

Why does my RAG app give wrong answers even with the right documents?

Print the retrieved chunks for a failing question. If the answer text is missing, retrieval failed: adjust chunking, top-k or the embedding model. If it is present, the problem is the prompt or the model.

Can I build RAG without an OpenAI API key?

Yes, and this post does it. The TF-IDF version needs nothing outside the standard library. For better matching you can use any embedding model, hosted or local, because the pipeline only needs a function that turns text into a vector.