Build RAG tutorial

What you will build

This tutorial walks through a minimal RAG pipeline in Python. You will load documents, split them into chunks, embed them, store them in a vector index, retrieve relevant chunks for a question, and generate an answer with a language model.

The examples are intentionally simple so you can adapt them to any framework or provider.

Prerequisites

• Python 3.10 or newer
• A small set of text documents or a single text file
• An embedding model and an LLM API or local model
• Basic familiarity with HTTP APIs and Python

Data preparation

Clean text is the foundation of good RAG. Remove broken encoding, duplicate sections, and boilerplate headers. Save each cleaned document as plain text or as a record with metadata.

# Load documents into a simple list
import glob

docs = []
for path in glob.glob("data/*.txt"):
    with open(path, "r", encoding="utf-8") as f:
        docs.append({
            "source": path,
            "text": f.read()
        })

Chunking

Chunking splits documents into passages small enough to retrieve precisely but large enough to retain meaning. Start with paragraph or fixed-size chunks with overlap.

def chunk_text(text, chunk_size=300, overlap=50):
    words = text.split()
    chunks = []
    for i in range(0, len(words), chunk_size - overlap):
        chunk = " ".join(words[i:i + chunk_size])
        if chunk:
            chunks.append(chunk)
    return chunks

chunked_docs = []
for doc in docs:
    for chunk in chunk_text(doc["text"]):
        chunked_docs.append({
            "source": doc["source"],
            "text": chunk
        })

Embeddings

Turn each chunk into a dense vector using an embedding model. Many providers offer small, cheap embedding models that work well for RAG.

import requests

def embed(texts, api_key):
    response = requests.post(
        "https://api.embedding-provider.example/v1/embeddings",
        headers={"Authorization": f"Bearer {api_key}"},
        json={"model": "text-embedding-3-small", "input": texts}
    )
    response.raise_for_status()
    return [item["embedding"] for item in response.json()["data"]]

vectors = embed([d["text"] for d in chunked_docs], api_key="YOUR_KEY")

Vector database

You can use a managed vector database, a local vector store, or even a NumPy matrix for a prototype. The goal is the same: given an embedding, return the closest chunks.

import numpy as np

class VectorStore:
    def __init__(self):
        self.vectors = []
        self.chunks = []

    def add(self, chunks, vectors):
        self.chunks.extend(chunks)
        self.vectors.extend(vectors)

    def search(self, query_vector, top_k=5):
        scores = np.dot(self.vectors, query_vector)
        top_indices = np.argsort(scores)[::-1][:top_k]
        return [self.chunks[i] for i in top_indices]

store = VectorStore()
store.add(chunked_docs, vectors)

Retrieval

To answer a question, embed it and search the vector store. Consider adding keyword boosting or re-ranking once the basics work.

def retrieve(question, store, top_k=5):
    query_vector = embed([question], api_key="YOUR_KEY")[0]
    return store.search(query_vector, top_k=top_k)

question = "What is the refund policy?"
context_chunks = retrieve(question, store)

Generation

Combine the retrieved chunks with a prompt that tells the model to answer using only the provided context. Ask it to cite sources or refuse if the context is insufficient.

def build_prompt(question, chunks):
    context = "\n\n".join(
        f"Source: {c['source']}\n{c['text']}" for c in chunks
    )
    return f"""Answer the question using only the context below.
If the context does not contain the answer, say you do not know.

Context:
{context}

Question: {question}
Answer:"""

def generate(prompt, api_key):
    response = requests.post(
        "https://api.llm-provider.example/v1/chat/completions",
        headers={"Authorization": f"Bearer {api_key}"},
        json={
            "model": "gpt-4o-mini",
            "messages": [{"role": "user", "content": prompt}]
        }
    )
    response.raise_for_status()
    return response.json()["choices"][0]["message"]["content"]

prompt = build_prompt(question, context_chunks)
answer = generate(prompt, api_key="YOUR_KEY")
print(answer)

Evaluation

Measure retrieval and generation separately. Create a small set of questions with known-good answers and source chunks.

test_set = [
    {
        "question": "What is the refund policy?",
        "expected_sources": ["data/refunds.txt"],
        "expected_answer": "Refunds are issued within 30 days."
    }
]

for item in test_set:
    chunks = retrieve(item["question"], store)
    sources = {c["source"] for c in chunks}
    retrieved_expected = bool(sources.intersection(set(item["expected_sources"])))
    print(item["question"], "retrieved:", retrieved_expected)

Full pipeline

Wrap the pieces into a single function so you can iterate on chunking, embedding, and prompts without rewriting everything.

def ask(question):
    chunks = retrieve(question, store)
    prompt = build_prompt(question, chunks)
    return generate(prompt, api_key="YOUR_KEY")

print(ask("How do I reset my password?"))

Once this works, add re-ranking, hybrid search, source citations, and a more rigorous evaluation framework.