Build a RAG Pipeline From Scratch Without a Framework

Retrieval-augmented generation is five stages and none of them needs a framework: split the text, turn each piece into a vector, store the vectors, find the nearest ones to the question, and put those in the prompt. Written out in Python’s standard library with two HTTP calls it comes to a little under two hundred lines — and every one of them is a line you can put a print statement in when the answers are wrong.

The five stages, and what each can break

Build it as five functions with visible intermediate output, because when a RAG system answers badly the cause is nearly always upstream of the model, and a framework that hides the stages hides the cause.

Stage	Description
Parse	Bytes to text. Fails silently on PDFs, tables and anything with columns.
Chunk	Text to passages. Too big buries the answer; too small severs it from its context.
Embed	Passage to vector. Cheap, but re-running it on everything is the cost you notice.
Retrieve	Question to passages. Where 'the answer was in the corpus but not in the prompt' happens.
Generate	Passages to answer. Where a model ignores the passages and answers from memory.

Parse and chunk

Start with plain text or Markdown. Chunk on paragraph boundaries up to a character budget, with an overlap, so a sentence that straddles a boundary still appears whole in one chunk.

# rag.py — part 1
import json, math, os, re, sqlite3, urllib.request

MAX_CHARS, OVERLAP = 1200, 200

def chunk(text):
    paras = [p.strip() for p in re.split(r"\n\s*\n", text) if p.strip()]
    chunks, buf = [], ""
    for p in paras:
        if len(buf) + len(p) + 2 <= MAX_CHARS:
            buf = (buf + "\n\n" + p) if buf else p
        else:
            if buf:
                chunks.append(buf)
            buf = (buf[-OVERLAP:] + "\n\n" + p) if buf else p
            while len(buf) > MAX_CHARS:          # a single huge paragraph
                chunks.append(buf[:MAX_CHARS])
                buf = buf[MAX_CHARS - OVERLAP:]
    if buf:
        chunks.append(buf)
    return chunks

1,200 characters is roughly 300 tokens of English prose. That number is not sacred: it is a trade between recall and precision, and the right value depends on how your documents are written. Chunking strategy is the single highest-leverage knob in a RAG system, and it is worth trying 600 and 2,400 on your own corpus before tuning anything else.

Embed, once

An embedding call takes a list of strings and returns a list of vectors in the same order. Batch them — one request per chunk is dominated by round-trip time — and store the result, because re-embedding a corpus you have already embedded is the most common avoidable cost in this whole design.

# rag.py — part 2
BASE = os.environ["LLM_BASE_URL"]
KEY  = os.environ["LLM_API_KEY"]
EMBED_MODEL = os.environ.get("EMBED_MODEL", "text-embedding-3-small")

def post(path, payload):
    req = urllib.request.Request(
        BASE + path,
        data=json.dumps(payload).encode(),
        headers={"Authorization": "Bearer " + KEY,
                 "Content-Type": "application/json"},
    )
    with urllib.request.urlopen(req, timeout=120) as r:
        return json.load(r)

def embed(texts, batch=64):
    out = []
    for i in range(0, len(texts), batch):
        res = post("/embeddings", {"model": EMBED_MODEL,
                                   "input": texts[i:i + batch]})
        out.extend(d["embedding"] for d in
                   sorted(res["data"], key=lambda d: d["index"]))
    return out

The sorted(..., key=index) is not superstition. The response carries an index field precisely because ordering is not guaranteed by the shape of the JSON, and a silently reordered batch gives you a corpus where every chunk’s vector belongs to a different chunk — which produces retrieval that is confidently, consistently wrong and takes a day to find.

Store vectors as JSON text in SQLite. It is not elegant and it is perfectly adequate to a few tens of thousands of chunks:

DB = sqlite3.connect("rag.db")
DB.execute("""CREATE TABLE IF NOT EXISTS chunk (
  id     INTEGER PRIMARY KEY,
  source TEXT NOT NULL,
  ord    INTEGER NOT NULL,
  text   TEXT NOT NULL,
  vec    TEXT NOT NULL
)""")

def index_document(source, text):
    pieces = chunk(text)
    vecs = embed(pieces)
    DB.executemany(
        "INSERT INTO chunk (source, ord, text, vec) VALUES (?,?,?,?)",
        [(source, i, p, json.dumps(v)) for i, (p, v) in enumerate(zip(pieces, vecs))],
    )
    DB.commit()

Retrieve with a dot product

Cosine similarity is the dot product of two unit vectors. Normalise once at write time and retrieval becomes a dot product and a sort — which is why the “do I need a vector database” question has a boring answer at small scale.

def normalise(v):
    n = math.sqrt(sum(x * x for x in v)) or 1.0
    return [x / n for x in v]

def search(question, k=5):
    q = normalise(embed([question])[0])
    scored = []
    for cid, source, text, vec in DB.execute(
            "SELECT id, source, text, vec FROM chunk"):
        v = normalise(json.loads(vec))
        score = sum(a * b for a, b in zip(q, v))
        scored.append((score, cid, source, text))
    scored.sort(reverse=True)
    return scored[:k]

Print the scores while you are developing. Cosine similarity on a modern embedding model tends to sit in a narrow band — the useful signal is the gap between the first result and the fifth, not the absolute value. A question whose top five scores are all within 0.02 of each other is a question your corpus does not answer, and that is the cheapest abstention signal you will ever get.

Answer, with the sources attached

The generation prompt does two jobs: it supplies the passages, and it forbids answering without them. The second is what separates a RAG system from a model that has read some notes.

SYSTEM = (
  "Answer using only the numbered passages below. "
  "Cite the passages you used as [1], [2] and so on. "
  "If the passages do not contain the answer, say exactly: "
  "'The documents do not answer that.' Do not use outside knowledge."
)

def answer(question, k=5):
    hits = search(question, k)
    context = "\n\n".join(
        "[" + str(i + 1) + "] (" + s + ")\n" + t
        for i, (_score, _cid, s, t) in enumerate(hits)
    )
    res = post("/chat/completions", {
        "model": os.environ.get("LLM_MODEL", "gpt-4o-mini"),
        "temperature": 0,
        "messages": [
            {"role": "system", "content": SYSTEM},
            {"role": "user", "content": context + "\n\nQuestion: " + question},
        ],
    })
    return res["choices"][0]["message"]["content"], hits

The exact refusal string is deliberate. A fixed sentence is something you can count in logs, alert on, and use to decide whether the corpus needs more documents — an open-ended “I’m not sure” is not. Making abstention a first-class output is worth more to a support-facing system than any amount of retrieval tuning.

Counting the finished file: parse and chunk is 20 lines, the HTTP helpers 18, indexing 15, search 12, generation 20, plus a small CLI — about 110 lines of substance, 180 with the schema, imports and argument parsing. The plan for this page promised two hundred; it is a little under, and the honest reason is that sqlite3 and urllib are doing the work a dependency would otherwise be credited with.

When brute force stops working

Scan-and-sort is O(n) per query. Here is the arithmetic, so you can see where your own corpus sits rather than guessing.

One chunk of 1,200 characters  ~ 300 tokens
1,536-dimension float vector   = 1,536 x 4 bytes  = 6.1 KB as floats
                               ~ 30 KB as JSON text (the lazy storage above)

10,000 chunks   -> 61 MB of floats, ~300 MB of JSON
                -> 15.4M multiply-adds per query in pure Python
                -> order of 1-3 seconds per query on one core

100,000 chunks  -> 610 MB of floats
                -> 154M multiply-adds  -> tens of seconds. Unusable.

So the honest threshold is around ten thousand chunks in pure Python, and roughly ten times that if you move the dot products into NumPy where the whole corpus becomes one matrix multiply. Past that you want an index — HNSW is the structure nearly every vector database uses — and the migration is a swap of the search function only, which is the payoff for having written the stages separately. Whether you need a dedicated database at all is mostly this arithmetic and nothing else.

What goes wrong first

• The answer is in the corpus but never in the prompt. Retrieval, not generation. Search for a distinctive phrase from the passage you expect and see what rank it comes back at. If it is not in the top fifty, the chunk boundary probably split it; if it is at rank six with k=5, raise k and add a reranking pass.
• The model answers from its own knowledge. Visible as an answer with no citations, or citations that do not support the claim. Temperature 0 and an explicit refusal instruction fix most of it; the rest is a citation check that verifies the quoted span exists in the passage it cites.
• Keyword queries fail. Embeddings are poor at exact identifiers — part numbers, error codes, surnames. The fix is not a better embedding model, it is running BM25 alongside the vector search and merging. SQLite’s FTS5 extension gives you the keyword half without a new dependency.

Testing it without a labelled dataset

You will want to change the chunk size, the embedding model, the value of k and the prompt, and you cannot tell whether any of those helped by reading a few answers. The cheapest honest evaluation separates the two halves, because they fail independently and only one of them needs a model to judge.

1. Write twenty questions and note the chunk that answers each. Not the answer — the chunk. An hour of work, and it never has to be redone unless the corpus changes.
2. Measure retrieval alone. For each question, record the rank at which the correct chunk appears. Report recall at 5 and the mean reciprocal rank. This is deterministic, costs one embedding call per question, and it is the number that moves when you change chunking.
3. Measure generation only on questions retrieval got right. Otherwise a generation regression and a retrieval regression look identical in the aggregate, which is how teams spend a week tuning the wrong prompt.
4. Add ten questions the corpus does not answer. The expected output is the refusal string, exactly. Abstention rate on these is the metric that stops a change to the prompt quietly turning a careful system into a confident one.

def evaluate(cases, k=5):
    """cases: [{"q": str, "gold_chunk_id": int}]"""
    hits_at_k, rr = 0, 0.0
    for c in cases:
        ranked = [cid for _s, cid, _src, _t in search(c["q"], k=50)]
        if c["gold_chunk_id"] in ranked[:k]:
            hits_at_k += 1
        if c["gold_chunk_id"] in ranked:
            rr += 1.0 / (ranked.index(c["gold_chunk_id"]) + 1)
    n = len(cases)
    return {"recall_at_k": hits_at_k / n, "mrr": rr / n}

Twenty questions is a small sample and the numbers will be noisy — a change of one question in twenty is five percentage points. Treat a movement of less than about 10 points as noise, and grow the set to a hundred once you have a system worth defending. Building an evaluation set and evaluating retrieval and generation separately are worth reading before you scale this up, because the mistakes are well known and expensive to unwind.

Related

• Build a Semantic Search Engine for a Website
• Build a Document Q&A App Over Your Own PDFs