RAG from Scratch with Python: understand how it works

TL;DR

What you’ll learn: how RAG works under the hood — chunking, embeddings, cosine similarity, retrieval, and augmented prompt construction — implementing each piece in plain Python before touching any framework.

Requirements:

• Python 3.10+
• pip install anthropic sentence-transformers numpy
• An Anthropic API key (ANTHROPIC_API_KEY in your environment)

Estimated time: 45–60 minutes reading and running the code.

---

What is RAG and why isn’t a long prompt enough?

RAG (Retrieval-Augmented Generation) is an architecture that adds a retrieval step before calling the LLM: instead of stuffing all your knowledge into the prompt, you search for only the relevant fragments and inject them into context.

There are three reasons a long prompt isn’t enough. First, models degrade attention over very long contexts — information at the middle of a 200-page prompt gets processed differently than what’s at the top. Second, token costs grow linearly with context size: if your knowledge base has 500 pages, passing everything on every call is economically unsound. Third, and most importantly: the model doesn’t need everything — it needs the right pieces. RAG solves exactly that.

For a deeper look at what happens inside the model when you feed it that context, the post on why LLMs don’t think in embeddings is worth reading before or after this tutorial.

---

Step 1: split the document (chunking)

Chunking converts a long document into smaller fragments that make sense on their own. Size matters: too small and you lose context, too large and you introduce noise.

A chunk should contain one complete idea. The practical rule is 200–500 words, with overlap between consecutive chunks to avoid losing sentences that fall on the boundary.

def chunk_text(text: str, chunk_size: int = 400, overlap: int = 50) -> list[str]:
    """
    Split text into overlapping chunks.

    Args:
        text: Full text to split.
        chunk_size: Approximate size of each chunk in characters.
        overlap: Characters of overlap between consecutive chunks.

    Returns:
        List of strings, one per chunk.
    """
    chunks = []
    start = 0
    while start < len(text):
        end = start + chunk_size
        chunk = text[start:end]
        # Try to cut at a space to avoid breaking words
        if end < len(text):
            last_space = chunk.rfind(" ")
            if last_space > 0:
                chunk = chunk[:last_space]
                end = start + last_space
        chunks.append(chunk.strip())
        start = end - overlap  # Overlap links adjacent chunks
    return [c for c in chunks if c]  # Drop empty chunks


# Example usage
document = """
Transformers are a neural network architecture proposed in 2017
in the paper 'Attention is All You Need'. They rely on attention
mechanisms that allow the model to relate any pair of positions
in the input sequence. This makes them especially efficient
for processing text in parallel, unlike RNNs which process
sequences one step at a time.

The original encoder-decoder architecture is used in translation tasks.
Decoder-only models like GPT generate text autoregressively.
Encoder-only models like BERT are used for classification and text understanding.
"""

chunks = chunk_text(document, chunk_size=300, overlap=50)
for i, chunk in enumerate(chunks):
    print(f"Chunk {i}: {chunk[:80]}...")

In production you’ll use more sophisticated splitters (by paragraph, by sentence, or semantic splitters), but the logic is the same: divide and control the overlap.

---

Step 2: convert chunks into embeddings

An embedding is a vector representation of text — an array of numbers where the distance between vectors reflects semantic similarity. Two sentences with the same meaning have nearby vectors even if they use completely different words.

sentence-transformers is the standard library for this. The all-MiniLM-L6-v2 model is lightweight (80 MB), fast, and a solid starting point for English text. For multilingual or Spanish-heavy content, paraphrase-multilingual-MiniLM-L12-v2 performs better (see FAQ).

from sentence_transformers import SentenceTransformer
import numpy as np

# Load the embedding model (downloaded on first run, ~80 MB)
model = SentenceTransformer("sentence-transformers/all-MiniLM-L6-v2")

def embed_chunks(chunks: list[str]) -> np.ndarray:
    """
    Convert a list of texts into an embedding matrix.

    Args:
        chunks: List of strings to embed.

    Returns:
        Array of shape (n_chunks, embedding_dim). For all-MiniLM-L6-v2,
        embedding_dim is 384.
    """
    embeddings = model.encode(chunks, convert_to_numpy=True)
    return embeddings


# Embed the chunks from the previous step
chunk_embeddings = embed_chunks(chunks)

print(f"Embedding shape: {chunk_embeddings.shape}")
# Expected output: (n_chunks, 384)
print(f"First vector (first 5 dims): {chunk_embeddings[0][:5]}")

model.encode() accepts a list of strings and returns a np.ndarray. The convert_to_numpy=True parameter is included explicitly to make the code self-documenting; it is the default value in current versions of sentence-transformers.

---

Step 3: store embeddings (minimal vector store)

A vector store is a structure that lets you search for vectors similar to a query efficiently. In production you’d use Chroma, Qdrant, or Pinecone. To understand the mechanism, an in-memory dataclass is enough.

from dataclasses import dataclass

@dataclass
class VectorStore:
    """Minimal in-memory vector store."""
    chunks: list[str]
    embeddings: np.ndarray  # Shape: (n_chunks, embedding_dim)

    @classmethod
    def from_chunks(cls, chunks: list[str], embed_fn) -> "VectorStore":
        """Build the store from chunks and an embedding function."""
        embeddings = embed_fn(chunks)
        return cls(chunks=chunks, embeddings=embeddings)


# Build the store
store = VectorStore.from_chunks(chunks, embed_chunks)
print(f"Store with {len(store.chunks)} indexed chunks.")

In a real system, embeddings are persisted to disk or a vector database so you don’t recompute them on every startup. The embedding cost is the indexing cost — you only pay it once.

---

Step 4: retrieve the most relevant chunks (cosine similarity)

Cosine similarity measures the angle between two vectors. If the angle is 0°, the vectors point in the same direction — maximum similarity. At 90°, they’re unrelated. The formula: cos(θ) = (A · B) / (‖A‖ × ‖B‖).

To retrieve the most relevant chunks for a query: embed the query, compute its similarity against every chunk, return the top K.

def cosine_similarity(vec_a: np.ndarray, vec_b: np.ndarray) -> float:
    """
    Compute cosine similarity between two vectors.

    Args:
        vec_a: 1D array.
        vec_b: 1D array of the same size as vec_a.

    Returns:
        Float between -1 and 1. Closer to 1 = more similar.
    """
    dot_product = np.dot(vec_a, vec_b)
    norm_a = np.linalg.norm(vec_a)
    norm_b = np.linalg.norm(vec_b)
    if norm_a == 0 or norm_b == 0:
        return 0.0
    return dot_product / (norm_a * norm_b)


def retrieve(
    query: str,
    store: VectorStore,
    embed_fn,
    top_k: int = 3
) -> list[tuple[str, float]]:
    """
    Retrieve the top_k chunks most similar to the query.

    Args:
        query: Search query or question.
        store: The indexed VectorStore.
        embed_fn: Function that converts text to an embedding.
        top_k: Number of chunks to retrieve.

    Returns:
        List of (chunk_text, score) tuples sorted by relevance desc.
    """
    query_embedding = embed_fn([query])[0]  # Shape: (embedding_dim,)

    scores = [
        cosine_similarity(query_embedding, chunk_emb)
        for chunk_emb in store.embeddings
    ]

    # Sort by score descending and return top_k
    ranked = sorted(
        zip(store.chunks, scores),
        key=lambda x: x[1],
        reverse=True
    )
    return ranked[:top_k]


# Test retrieval
query = "What's the difference between encoder and decoder?"
results = retrieve(query, store, embed_chunks, top_k=2)

for chunk, score in results:
    print(f"Score: {score:.4f} | Chunk: {chunk[:100]}...")

---

Step 5: build the augmented prompt and call the LLM

With the retrieved chunks, you build a prompt that includes the relevant context and the user’s question. The model responds based on that context, not on its training knowledge.

import anthropic
import os

def build_prompt(query: str, context_chunks: list[tuple[str, float]]) -> str:
    """
    Build the augmented prompt with retrieved chunks.

    Args:
        query: Original user question.
        context_chunks: List of (text, score) from retrieve().

    Returns:
        Complete prompt string to send to the LLM.
    """
    context_text = "\n\n---\n\n".join(
        f"[Fragment {i+1}]\n{chunk}"
        for i, (chunk, _) in enumerate(context_chunks)
    )

    return f"""Use ONLY the following information to answer the question.
If the answer is not in the provided context, say "I don't have information about that in the documents."

CONTEXT:
{context_text}

QUESTION: {query}

ANSWER:"""


def rag_query(
    query: str,
    store: VectorStore,
    embed_fn,
    top_k: int = 3
) -> str:
    """
    Full RAG pipeline: retrieve + generate.

    Args:
        query: User question.
        store: Indexed VectorStore.
        embed_fn: Embedding function.
        top_k: Chunks to retrieve.

    Returns:
        LLM-generated answer.
    """
    # 1. Retrieve relevant context
    context_chunks = retrieve(query, store, embed_fn, top_k=top_k)

    # 2. Build augmented prompt
    prompt = build_prompt(query, context_chunks)

    # 3. Call the LLM
    client = anthropic.Anthropic(api_key=os.environ.get("ANTHROPIC_API_KEY"))

    response = client.messages.create(
        model="claude-sonnet-4-6",
        max_tokens=1024,
        messages=[
            {"role": "user", "content": prompt}
        ]
    )

    return response.content[0].text


# Run the full pipeline
answer = rag_query(
    query="What are transformers and what are they used for?",
    store=store,
    embed_fn=embed_chunks,
    top_k=2
)
print(answer)

---

The complete pipeline at a glance

Putting all the pieces together, the flow is linear:

# rag_pipeline.py — complete minimal RAG system
import os
import numpy as np
from dataclasses import dataclass
from sentence_transformers import SentenceTransformer
import anthropic


# --- Config ---
EMBED_MODEL = "sentence-transformers/all-MiniLM-L6-v2"
LLM_MODEL = "claude-sonnet-4-6"
CHUNK_SIZE = 400
OVERLAP = 50
TOP_K = 3

embed_model = SentenceTransformer(EMBED_MODEL)


# --- Chunking ---
def chunk_text(text: str, chunk_size: int = CHUNK_SIZE, overlap: int = OVERLAP) -> list[str]:
    chunks, start = [], 0
    while start < len(text):
        end = start + chunk_size
        chunk = text[start:end]
        if end < len(text):
            last_space = chunk.rfind(" ")
            if last_space > 0:
                chunk = chunk[:last_space]
                end = start + last_space
        chunks.append(chunk.strip())
        start = end - overlap
    return [c for c in chunks if c]


# --- Embeddings ---
def embed(texts: list[str]) -> np.ndarray:
    return embed_model.encode(texts, convert_to_numpy=True)


# --- Vector Store ---
@dataclass
class VectorStore:
    chunks: list[str]
    embeddings: np.ndarray

    @classmethod
    def build(cls, text: str) -> "VectorStore":
        chunks = chunk_text(text)
        return cls(chunks=chunks, embeddings=embed(chunks))


# --- Retrieval ---
def retrieve(query: str, store: VectorStore, top_k: int = TOP_K) -> list[str]:
    q_emb = embed([query])[0]
    scores = [
        np.dot(q_emb, emb) / (np.linalg.norm(q_emb) * np.linalg.norm(emb) + 1e-9)
        for emb in store.embeddings
    ]
    ranked = sorted(zip(store.chunks, scores), key=lambda x: x[1], reverse=True)
    return [chunk for chunk, _ in ranked[:top_k]]


# --- Generation ---
def generate(query: str, context: list[str]) -> str:
    ctx = "\n\n---\n\n".join(context)
    prompt = f"Context:\n{ctx}\n\nQuestion: {query}\n\nAnswer:"
    client = anthropic.Anthropic()  # Uses ANTHROPIC_API_KEY from environment
    response = client.messages.create(
        model=LLM_MODEL,
        max_tokens=1024,
        messages=[{"role": "user", "content": prompt}]
    )
    return response.content[0].text


# --- Pipeline ---
def rag(document: str, question: str) -> str:
    store = VectorStore.build(document)
    context = retrieve(question, store)
    return generate(question, context)


# --- Run ---
if __name__ == "__main__":
    doc = open("my_document.txt").read()
    answer = rag(doc, "What is the main idea of this document?")
    print(answer)

---

From here to real frameworks: what changes (and what doesn’t)

Once you understand the pieces, frameworks are just abstraction layers over exactly what you just built.

LangChain wraps chunking, embeddings, and retrieval in configurable chains. It has more options (semantic splitters, hybrid BM25, rerankers), but the flow is identical to what you implemented here.

LlamaIndex focuses on indexing structured documents (PDFs, tables, node trees). Useful when your knowledge base isn’t plain text.

Chroma / Qdrant / Pinecone replace the in-memory dataclass with a persistent vector store backed by HNSW indices for efficient approximate search over millions of vectors.

What doesn’t change: the concepts. Chunk → Embed → Store → Retrieve → Augment → Generate. Always.

To see how RAG fits into more complex systems, build AI agents for free covers the full landscape of frameworks available today. And if you want to understand what MCP adds on top of this pattern — dynamic, real-time context instead of static retrieval — MCP servers explained is the natural next step.

---

FAQ

What chunk size is best? It depends on the content. For narrative text, 300–500 characters with 50–100 overlap. For dense technical documentation, smaller chunks (150–200) work better. The best approach is to experiment: measure retrieval quality with test questions where you already know the correct answer.

Why cosine similarity and not Euclidean distance? Euclidean distance is sensitive to vector magnitude. Two texts with the same meaning but different lengths can produce vectors of different magnitudes and appear “far apart” in Euclidean space. Cosine measures the angle, not the distance, making it robust to text length.

How many chunks should I retrieve (top_k)? Start with 3–5. More chunks means more context, which means more token cost and more risk of attention dilution. If the model ignores relevant information, lower the k. If it says “I don’t have information” when it should know something, raise it.

Can I use a different embedding model? Yes. all-MiniLM-L6-v2 is the standard starting point. For multilingual use cases, paraphrase-multilingual-MiniLM-L12-v2 performs better. For maximum quality, BAAI/bge-m3 is the state of the art for multilingual retrieval in 2026.

What if the relevant chunk doesn’t make it into top_k? That’s a recall problem. The most common causes: chunks too small (losing context), embedding model not suited to the domain, or highly technical vocabulary. The standard fix is reranking: retrieve more candidates (top_20) then reorder them with a more precise cross-encoder before keeping the best few.

How do I know if my RAG is working well? Build an evaluation set of questions with known correct answers. Measure: precision@k (are the retrieved chunks relevant?), recall@k (is the chunk containing the answer in the retrieved set?), and the quality of the final answer. Ragas is the most widely used framework for this.

RAG vs fine-tuning: when to use each? RAG when knowledge changes frequently or is voluminous. Fine-tuning when you need to change the model’s response style or embed knowledge that’s used in virtually every query. In most business cases, RAG is faster, cheaper, and more maintainable.

Does this code work with other LLMs? Chunking, embeddings, and retrieval are LLM-agnostic. You only need to swap the call in generate() for whichever SDK you use — OpenAI, Mistral, a local model via Ollama. The architecture doesn’t change.