Build a Local RAG Pipeline in Python with No API Costs

A complete Retrieval-Augmented Generation pipeline can be built and run entirely on a personal computer, with CPU only, free open-source models, no OpenAI key, no credit card and no rate limits. The point is to understand each piece instead of copying a black box.

The best way through is hands-on: keep a terminal and a Python file called rag.py open, run every block and predict the output before executing it. Short exercises appear after most steps, and they're where most of the learning happens.

The problem RAG addresses

A language model knows only what its training data contained. Private notes, a company wiki, last week's meeting or any unseen document are invisible to it. Asked about them, it either admits ignorance or, worse, invents something with confidence.

RAG splits the job into two parts:

1. Retrieve: find the chunks of your own documents that relate to the question. That's a search problem.
2. Generate: give those chunks to the model with the instruction to answer from them.

The model stops guessing and starts reading. Everything else is mechanics for doing those two steps well.

question ──► [embed] ──► [search your docs] ──► top chunks ──┐
                                                             ▼
                                          [LLM: "answer using this context"] ──► answer

The free toolkit

Three components cover the whole system. sentence-transformers turns text into vectors, using a small model that runs on CPU and downloads once. numpy handles the similarity math, and the entire "vector database" fits in about five lines so the workings stay visible. transformers with Flan-T5 writes the answer, since it's instruction-tuned, runs on CPU and needs no API.

Install everything once:

pip install sentence-transformers transformers torch numpy

On first load each model pulls its weights from Hugging Face. The embedder is roughly 80 MB and flan-t5-base about 1 GB. After that they're cached and work offline. With limited disk or RAM, flan-t5-small (around 300 MB) can replace the base model throughout, at lower quality but with much better speed.

Step 1: a corpus the model can't know

RAG is easiest to appreciate with documents no model could have seen, so the script starts with a tiny fictional company wiki. Create rag.py and paste in the documents:

# rag.py
DOCUMENTS = [
    """Nimbus is a fictional note-taking app launched in 2023. The free plan,
    called Nimbus Lite, allows up to 50 notes and 1 GB of storage. There are no
    collaboration features on the free plan.""",

The wiki continues with more entries:

    """Nimbus Pro costs 8 dollars per month billed annually, or 10 dollars billed
    monthly. Pro removes the note limit, gives 50 GB of storage, and unlocks
    real-time collaboration with up to 5 people per note.""",    """Nimbus stores all notes encrypted at rest using AES-256. End-to-end
    encryption is only available on the Pro plan and must be enabled manually in
    Settings > Security. Once enabled it cannot be turned off for that note.""",    """The Nimbus mobile app supports offline editing. Changes made offline are
    queued and sync automatically the next time the device is online. If two
    devices edit the same note offline, Nimbus keeps both versions and flags a
    conflict for the user to resolve.""",    """Nimbus offers a 30-day refund policy on all paid plans, no questions asked.
    Refunds are processed to the original payment method within 5 business days.
    Annual plans cancelled after 30 days are not refundable but stay active until
    the end of the billing period.""",    """Nimbus support is available via email at help@nimbus.example and live chat.
    Live chat is only staffed for Pro customers, Monday to Friday, 9am to 6pm UTC.
    Free-plan users receive email support with a typical 48-hour response time.""",
]

Before building anything, prove the problem exists. In a Python shell, ask a plain model about Nimbus:

from transformers import pipeline
gen = pipeline("text2text-generation", model="google/flan-t5-base")
print(gen("How much does Nimbus Pro cost?", max_new_tokens=50)[0]["generated_text"])

It'll produce something, and that something will be wrong or empty because it's a guess. Keep that answer in mind, since the rest of the tutorial fixes it.

Step 2: chunking

Whole documents don't go to the model. They're cut into smaller chunks for two reasons: retrieval gets more precise, because it finds the paragraph that answers the question and not an entire manual, and the model's context window is limited.

Add a chunking function:

def chunk_text(text, chunk_size=60, overlap=15):
    """Split text into overlapping chunks of `chunk_size` words."""
    words = text.split()
    chunks = []
    start = 0
    while start < len(words):
        end = start + chunk_size
        chunks.append(" ".join(words[start:end]))
        if end >= len(words):
            break
        start = end - overlap   # step back by `overlap` so context isn't cut
    return chunks

Then build the chunk list while recording which document each chunk came from:

# Build our chunk list, remembering which doc each chunk came from
chunks = []
for doc_id, doc in enumerate(DOCUMENTS):
    for c in chunk_text(doc):
        chunks.append({"doc_id": doc_id, "text": c})print(f"{len(DOCUMENTS)} documents -> {len(chunks)} chunks")
for c in chunks[:3]:
    print("-", c["text"][:70], "...")

Overlap exists because a fact can straddle a boundary, such as a refund policy cut between "within 5" and "business days". With no overlap neither chunk holds the whole fact. Repeating a few words at each seam lets facts survive the cut.

Exercise: rerun with chunk_size=20 for more, smaller chunks, then try chunk_size=500, where these short documents each become a single chunk. Before running, guess which setting retrieves more precisely but risks losing surrounding context. Small chunks are precise and context-poor, and that tension sits at the heart of chunking. No universally correct size exists.

Step 3: embeddings

An embedding is a vector of numbers encoding the meaning of a text. Pieces of text that mean similar things end up as vectors pointing the same way, even with no shared words, so "How much is it?" lands next to "What is the price?"

Import the embedding model:

from sentence_transformers import SentenceTransformer

Then embed every chunk:

embedder = SentenceTransformer("all-MiniLM-L6-v2")# Embed every chunk. normalize_embeddings=True makes the vectors unit-length,
# which lets us measure similarity with a simple dot product later.
chunk_texts = [c["text"] for c in chunks]
chunk_vectors = embedder.encode(chunk_texts, normalize_embeddings=True)print("vector shape:", chunk_vectors.shape)   # (num_chunks, 384)

Every chunk now corresponds to a 384-dimensional vector, where 384 is just the count of numbers this specific model uses to express meaning.

Exercise: watch meaning at work with a small experiment:

import numpy as np
pairs = embedder.encode(
    ["the price of the pro plan", "how much does it cost", "the weather in Paris"],
    normalize_embeddings=True,
)
print("price vs cost :", round(float(pairs[0] @ pairs[1]), 3))   # should be HIGH
print("price vs weather:", round(float(pairs[0] @ pairs[2]), 3)) # should be LOW

The expression a @ b is a dot product, which for unit vectors equals cosine similarity. A value of 1.0 means the same direction and 0 means unrelated. The first pair should score far above the second even though the phrases share no words, and that's semantic search. Try to guess both numbers within 0.2 before running.

Step 4: retrieval

For the search step, embed the question, compare it with every chunk vector and keep the top few. Since the vectors are normalized, "compare" reduces to a matrix multiply.

import numpy as np

The retrieval function:

def retrieve(question, k=3):
    q_vec = embedder.encode([question], normalize_embeddings=True)[0]
    scores = chunk_vectors @ q_vec              # cosine similarity to every chunk
    top_idx = np.argsort(scores)[::-1][:k]      # indices of the k highest scores
    return [(chunks[i]["text"], float(scores[i])) for i in top_idx]for text, score in retrieve("How much does Nimbus Pro cost?"):
    print(f"[{score:.3f}] {text[:80]}...")

The pricing chunk should rise to the top with the best score, even though the question said "cost" while the document said "costs 8 dollars per month". It's meaning matching and not keyword matching.

Exercise: try three questions and see what surfaces:

• retrieve("Can I get my money back?") should return the refund chunk, though the word "refund" never appeared in the question.
• retrieve("Is my data safe?") should surface the encryption chunk.
• retrieve("Does it still work without an internet connection?") is expected to bring back the offline-sync chunk.

Many people underestimate this stage. When retrieval returns the wrong chunks, no clever prompting rescues the answer. Garbage in, garbage out, so debugging effort belongs here first.

Step 5: generation

Now the two halves meet. Build a prompt that passes the retrieved chunks to Flan-T5 and tells it to answer only from them.

from transformers import pipeline

The generation function:

generator = pipeline("text2text-generation", model="google/flan-t5-base")def rag_answer(question, k=3):
    retrieved = retrieve(question, k=k)
    context = "\n".join(text for text, _ in retrieved)    prompt = f"""Answer the question using only the context below.
If the answer is not in the context, say you don't know.Context:
{context}Question: {question}
Answer:"""    out = generator(prompt, max_new_tokens=80)[0]["generated_text"]
    return out.strip(), retrievedanswer, sources = rag_answer("How much does Nimbus Pro cost?")
print("ANSWER:", answer)
print("\nBased on:")
for text, score in sources:
    print(f"  [{score:.3f}] {text[:70]}...")

Compare the result with the wrong or empty answer from Step 1. The model and the question are identical, and the only change is that the relevant text was supplied for reading. That gap is the entire value of RAG.

Exercise: pose a question the documents don't cover, for example about a Linux desktop client for Nimbus, and call rag_answer with it. A properly grounded setup admits it doesn't know. If yours makes up a reply, that's a genuine failure mode named hallucination despite retrieval, and a later section deals with it.

The finished script

Here's the entire rag.py as a single runnable program with an interactive loop. It begins with imports:

# rag.py — a complete, local, no-API RAG system
import numpy as np
from sentence_transformers import SentenceTransformer
from transformers import pipeline

The full body follows:

DOCUMENTS = [
    """Nimbus is a fictional note-taking app launched in 2023. The free plan,
    called Nimbus Lite, allows up to 50 notes and 1 GB of storage. There are no
    collaboration features on the free plan.""",
    """Nimbus Pro costs 8 dollars per month billed annually, or 10 dollars billed
    monthly. Pro removes the note limit, gives 50 GB of storage, and unlocks
    real-time collaboration with up to 5 people per note.""",
    """Nimbus stores all notes encrypted at rest using AES-256. End-to-end
    encryption is only available on the Pro plan and must be enabled manually in
    Settings > Security. Once enabled it cannot be turned off for that note.""",
    """The Nimbus mobile app supports offline editing. Changes made offline are
    queued and sync automatically the next time the device is online. If two
    devices edit the same note offline, Nimbus keeps both versions and flags a
    conflict for the user to resolve.""",
    """Nimbus offers a 30-day refund policy on all paid plans, no questions asked.
    Refunds are processed to the original payment method within 5 business days.
    Annual plans cancelled after 30 days are not refundable but stay active until
    the end of the billing period.""",
    """Nimbus support is available via email at help@nimbus.example and live chat.
    Live chat is only staffed for Pro customers, Monday to Friday, 9am to 6pm UTC.
    Free-plan users receive email support with a typical 48-hour response time.""",
]def chunk_text(text, chunk_size=60, overlap=15):
    words = text.split()
    chunks, start = [], 0
    while start < len(words):
        end = start + chunk_size
        chunks.append(" ".join(words[start:end]))
        if end >= len(words):
            break
        start = end - overlap
    return chunksprint("Loading models (first run downloads them)...")
embedder = SentenceTransformer("all-MiniLM-L6-v2")
generator = pipeline("text2text-generation", model="google/flan-t5-base")# Index the documents once at startup
chunks = []
for doc_id, doc in enumerate(DOCUMENTS):
    for c in chunk_text(doc):
        chunks.append({"doc_id": doc_id, "text": c})
chunk_vectors = embedder.encode(
    [c["text"] for c in chunks], normalize_embeddings=True
)def retrieve(question, k=3):
    q_vec = embedder.encode([question], normalize_embeddings=True)[0]
    scores = chunk_vectors @ q_vec
    top_idx = np.argsort(scores)[::-1][:k]
    return [(chunks[i]["text"], float(scores[i])) for i in top_idx]def rag_answer(question, k=3):
    retrieved = retrieve(question, k=k)
    context = "\n".join(text for text, _ in retrieved)
    prompt = (
        "Answer the question using only the context below. "
        "If the answer is not in the context, say you don't know.\n\n"
        f"Context:\n{context}\n\nQuestion: {question}\nAnswer:"
    )
    out = generator(prompt, max_new_tokens=80)[0]["generated_text"]
    return out.strip()if __name__ == "__main__":
    print("RAG ready. Ask about Nimbus (or type 'quit').\n")
    while True:
        q = input("You: ").strip()
        if q.lower() in {"quit", "exit", ""}:
            break
        print("Nimbus bot:", rag_answer(q), "\n")

Start it with:

python rag.py

Then try prompts like what the refund window is, whether free users get live chat or how big the free storage allowance is.

Proving the value with an A/B test

One experiment makes the case clearly by asking the same question with and without retrieved context. First, a baseline function:

def no_rag(question):
    out = generator(f"Question: {question}\nAnswer:", max_new_tokens=80)
    return out[0]["generated_text"].strip()

Then the comparison:

q = "Can free-plan Nimbus users use live chat support?"
print("WITHOUT context:", no_rag(q))
print("WITH context   :", rag_answer(q))

The answer without context is a guess, while the one with context should state that live chat is Pro-only and that free users get email. Try three different questions. This A/B is the most convincing RAG demo there is, so keep it for explaining the idea to others.

Upgrades by payoff

With a working system in hand, these upgrades matter most, roughly in order of payoff:

1. Show sources. The function already returns retrieved, so print the chunks under each answer. Real RAG products succeed or fail on this. A cited answer is trustworthy and an uncited one is a guess in a nice font.
2. Retrieve extra, then filter. Raise k to 5 and set a minimum score, discarding anything under roughly 0.25. When no chunk passes, reply "I don't know" and skip the model call entirely. That removes most hallucinations like the one in the Step 5 exercise.
3. A real vector store. chunk_vectors @ q_vec scans every chunk, which is fine for hundreds and slow for millions. Chroma (pip install chromadb) indexes vectors for fast search and saves them to disk so nothing is re-embedded at startup. The concepts of embedding, storing and cosine search stay the same, and Chroma only does them at scale.
4. A stronger free generator. Flan-T5 is small and sometimes terse. For better local answers, install Ollama, which runs models such as Llama 3.2 or Phi locally, and swap the generator(...) call for a request to its local API. A free API tier from Google's Gemini or Groq also works by replacing only the generation line in rag_answer. Only one function changes while retrieval stays as built.
5. Evaluation. Write ten questions with known answers, run them through rag_answer and check correctness. It's unglamorous, and it's what separates a toy from a product, because what isn't measured can't be improved.

Diagnosing problems

RAG is search plus a writing assistant, glued together, and nearly every issue traces to one half. A wrong answer with the right chunk retrieved is a generation or prompt problem. A wrong answer where the right chunk never appeared is a retrieval problem involving chunking, embeddings or k. Decide which half is broken before changing anything. Keeping retrieve() and rag_answer() as separate functions exists precisely so each half can be inspected on its own, and that separation is the most valuable habit here.

Troubleshooting

• Slow or failing download: models come from Hugging Face on first run and cache under ~/.cache/huggingface. Once cached, they work offline.
• Memory pressure or sluggishness: drop to google/flan-t5-small with a chunk_size of 40. The embedding model is small enough that it won't hold things up.
• Install trouble with torch: request the CPU-only wheel from the PyTorch index, pip install torch --index-url https://download.pytorch.org/whl/cpu.
• Very short answers: raise max_new_tokens, since Flan-T5 tends to stop early. A bigger generator such as one served by Ollama helps more than any tweak.
• One chunk wins every time: either the documents are near-duplicates or k is too low. Look at the raw scores, because they expose how the model ranks things.

Working through the exercises builds a deeper grasp of RAG than most people get from a library wrapper.