Build a minimal RAG pipeline in Python

Most RAG tutorials begin by installing a large framework. That is fine for production, but it hides what is actually happening. In this tutorial you will build the core of a retrieval-augmented generation system with plain Python and numpy, so that every moving part is visible. Once you understand it, any framework becomes easier to use and to debug.

You will build four things: a chunker, an index, a retriever and a prompt builder. Then you will add a small test that tells you how well retrieval works, which is the step most people skip.

What you need

• Python 3.9 or newer and numpy.
• A folder of plain text documents, such as policies or manuals. A few thousand words is plenty.
• An embedding function, embed(texts), that returns one vector per text. Any provider or local model will do.
• A language model you can call with a prompt. We will leave that call as a placeholder, because the retrieval is the part worth learning.

pip install numpy sentence-transformers

The package sentence-transformers is optional. It lets you run a small embedding model on your own machine, so no data leaves your computer. Here is how to wrap it:

from sentence_transformers import SentenceTransformer

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

def embed(texts):
    # returns a (n_texts, dim) array of unit-length vectors
    return _model.encode(texts, normalize_embeddings=True)

Step 1: load and chunk the documents

A language model has limited space, and a search needs specific passages, not whole books. We split each document into overlapping chunks. The overlap means that a sentence at the edge of one chunk also appears at the start of the next, so an idea is rarely split in half.

from pathlib import Path

def chunk(text, size=800, overlap=100):
    step = size - overlap
    return [text[i:i + size] for i in range(0, len(text), step)]

chunks, sources = [], []
for path in Path("docs").glob("*.txt"):
    for piece in chunk(path.read_text(encoding="utf-8")):
        chunks.append(piece)
        sources.append(path.name)

print(len(chunks), "chunks")

Step 2: embed and index

We embed every chunk once and keep the vectors in a numpy array. For thousands of chunks this is fast enough that you do not need a vector database yet. When the vectors have unit length, the dot product of two of them equals their cosine similarity, which makes retrieval a single matrix operation.

import numpy as np

vectors = np.array(embed(chunks), dtype="float32")   # shape: (n_chunks, dim)
vectors /= np.linalg.norm(vectors, axis=1, keepdims=True)

Step 3: retrieve the best passages

def retrieve(question, k=4):
    q = np.array(embed([question]), dtype="float32")[0]
    q /= np.linalg.norm(q)
    scores = vectors @ q                      # cosine similarity per chunk
    best = np.argsort(-scores)[:k]
    return [(int(i), float(scores[i])) for i in best]

Try it with a question you know the answer to and print the top results. If the right passage is not in the top four, you have found a retrieval problem, and you can fix it now, before any language model is involved.

Step 4: build a grounded prompt

The prompt is where you tell the model to stay inside the evidence. Number each passage, so that the answer can cite them, and tell the model what to do when the answer is missing.

def build_prompt(question):
    hits = retrieve(question)
    context = "\n\n".join(
        f"[{n}] ({sources[i]}) {chunks[i]}" for n, (i, _) in enumerate(hits, start=1)
    )
    return (
        "Answer the question using only the passages below. "
        "Cite the passage numbers you used, like [2]. "
        "If the answer is not in the passages, say you do not know.\n\n"
        f"Passages:\n{context}\n\nQuestion: {question}"
    )

def answer(question):
    return generate(build_prompt(question))   # generate() calls your language model

The function generate is a placeholder for whichever model you use. It takes a prompt string and returns text.

Step 5: test retrieval before anything else

This step separates a demo from a system you can improve. Write down 20 to 30 real questions, and for each one note a phrase that must appear in the right passage. Then measure how often retrieval finds it.

tests = [
    ("How many days of leave can I carry over?", "carry over"),
    ("Who approves travel expenses?", "travel"),
    # ...add 20 to 30 real questions with a key phrase each
]

hits_found = 0
for question, key_phrase in tests:
    top = retrieve(question, k=4)
    if any(key_phrase.lower() in chunks[i].lower() for i, _ in top):
        hits_found += 1

print(f"Retrieval hit rate: {hits_found / len(tests):.0%}")

Now you have a number. Change the chunk size, the overlap or the embedding model, and run the test again. If the number goes up, keep the change. This simple loop will improve your system faster than any amount of prompt tweaking.

Making it better

• Split on headings and paragraphs rather than fixed lengths.
• Store metadata such as document date and type, and filter on it, for example to prefer the newest version.
• Combine keyword search with embeddings, because exact codes and names are easy to miss with meaning-based search alone.
• Re-rank the top 20 results with a stronger model and keep the best four.
• Cache the embeddings to disk so you do not recompute them each run.

Things to watch for

• Access control. Retrieval must respect who is allowed to see which document, or the assistant will reveal things it should not.
• Untrusted text. Instructions hidden inside a document can try to steer the model. Treat retrieved text as data, never as commands.
• Stale content. If documents change, rebuild the index, or the assistant will quote outdated rules.
• Privacy. Sending documents to an outside embedding service shares them with that service.

Putting it together: a tiny command-line assistant

With the four functions in place, you can wrap them in a loop and talk to your documents from a terminal. This is a useful way to explore weak spots, because you can try questions quickly and see exactly which passages were retrieved.

def show_sources(question):
    for rank, (i, score) in enumerate(retrieve(question), start=1):
        print(f"{rank}. {sources[i]}  (score {score:.2f})")
        print("   " + chunks[i][:150].replace("\n", " ") + "...")

while True:
    q = input("\nAsk a question (or press Enter to quit): ").strip()
    if not q:
        break
    show_sources(q)
    print("\n" + answer(q))

Print the sources before the answer, as above. If the answer looks wrong, the sources will usually show you why: either the right passage is missing, or it is there and the model ignored it. Those two failures have very different fixes.

Costs and speed

• Embedding the documents is a one-off cost, repeated only when documents change. Save the vectors to disk and load them next time.
• Each question costs one embedding call and one language model call. Caching answers to common questions reduces both.
• With a few thousand chunks, the numpy search takes milliseconds. When you have hundreds of thousands, move to a vector index.
• The language model call dominates the response time. Streaming the reply makes it feel faster.

Frequently asked questions

Do I need LangChain or a similar framework?

No. A framework can save time once you know what you need, and its tools can help with tracing and evaluation, but the core of RAG is the small amount of code in this tutorial.

Which embedding model should I use?

Start with a small, well-known model, and test it on your questions. Switch only if the hit rate shows a problem. Remember that changing models means re-embedding everything.

How do I handle PDFs?

Extract the text first, with a PDF library, and check the result by eye. Scanned PDFs need text recognition. Tables and multi-column layouts often need extra care.

How do I stop the model from making things up?

Retrieve good passages, instruct it to answer only from them, require citations and test on questions with no answer. Evaluation is covered in the guide on evaluating a RAG assistant.

You have now built the heart of every RAG system. Everything else, from vector databases to re-rankers, is an upgrade to one of these four functions. Keep the test set, because it will tell you whether each upgrade is worth its cost.