RAG from scratch

A retrieval-augmented generation pipeline has four stages: chunk the source text, embed each chunk into a vector, find the chunks closest to the user's query, and feed those chunks to a language model as context. This lesson builds the entire loop in ~60 lines of Python with no framework and no vector database — just OpenAI embeddings, numpy, and Anthropic's Claude for generation.

Load and chunk

Start with raw text. In production this comes from a document parser (L4); here we'll use a plain string to keep the focus on the retrieval loop.

import numpy as np
from openai import OpenAI
import anthropic

openai_client = OpenAI()
anthropic_client = anthropic.Anthropic()

source_text = open("knowledge_base.txt").read()

Chunking splits the source into segments small enough that each one fits inside an embedding model's context window and carries a single coherent idea. A fixed-size chunker with overlap is the simplest approach:

def chunk_text(text, chunk_size=500, overlap=50):
    words = text.split()
    chunks = []
    start = 0
    while start < len(words):
        end = start + chunk_size
        chunk = " ".join(words[start:end])
        chunks.append(chunk)
        start += chunk_size - overlap
    return chunks

chunks = chunk_text(source_text)

chunk_size=500 words keeps each chunk well under the 8191-token limit of text-embedding-3-small. The 50-word overlap ensures that sentences split across chunk boundaries appear in full in at least one chunk.

Embed

Each chunk becomes a 1536-dimensional vector. The embedding model maps semantically similar text to nearby points in this space — "retrieval-augmented generation" and "RAG pipeline architecture" land close together even though they share few words.

def get_embeddings(texts, model="text-embedding-3-small"):
    response = openai_client.embeddings.create(input=texts, model=model)
    return np.array([e.embedding for e in response.data])

chunk_embeddings = get_embeddings(chunks)
# chunk_embeddings.shape: (num_chunks, 1536)

OpenAI's embedding API accepts batches of up to 2048 texts per request. For a small corpus this is a single call. For larger corpora, batch in groups of 2048.

Query: embed, search, retrieve

The query follows the same path — embed, then compare. Cosine similarity measures how close two vectors are in direction (ignoring magnitude). OpenAI's embedding vectors are already L2-normalized, so cosine similarity reduces to a dot product:

cosine_similarity(a, b) = dot(a, b) / (norm(a) * norm(b))

With normalized vectors: cosine_similarity(a, b) = dot(a, b)

def search(query, chunk_embeddings, chunks, top_k=5):
    query_embedding = get_embeddings([query])[0]

    similarities = chunk_embeddings @ query_embedding
    top_indices = np.argsort(similarities)[-top_k:][::-1]

    results = []
    for idx in top_indices:
        results.append({
            "chunk": chunks[idx],
            "score": float(similarities[idx]),
            "index": int(idx)
        })
    return results

chunk_embeddings @ query_embedding computes the dot product of every chunk vector against the query vector in one operation. np.argsort returns indices sorted by ascending similarity; [-top_k:][::-1] takes the top-k in descending order.

For 10,000 chunks of 1536 dimensions, this brute-force search takes under 1ms on a modern CPU. It becomes impractical around 1M+ vectors — that's where vector databases and approximate nearest neighbor indexes (HNSW, IVF) take over.

Assemble the prompt

The retrieved chunks become context for the generation model. The prompt template tells the model exactly what to do with them:

def build_prompt(query, search_results):
    context_blocks = []
    for i, result in enumerate(search_results):
        context_blocks.append(f"[{i+1}] (score: {result['score']:.3f})\n{result['chunk']}")

    context = "\n\n".join(context_blocks)

    return f"""Answer the question based on the provided context. If the context
doesn't contain enough information, say so — do not fabricate information.
Cite the relevant context block numbers in your answer.

Context:
{context}

Question: {query}"""

Including the similarity score in the context block is optional but useful during development — it lets you see how confident the retrieval was. In production you'd typically strip it.

Generate

Send the assembled prompt to Claude and get a grounded answer:

def rag_query(query, chunk_embeddings, chunks, top_k=5):
    results = search(query, chunk_embeddings, chunks, top_k)
    prompt = build_prompt(query, results)

    response = anthropic_client.messages.create(
        model="claude-sonnet-4-20250514",
        max_tokens=1024,
        messages=[{"role": "user", "content": prompt}]
    )
    return {
        "answer": response.content[0].text,
        "sources": results,
        "usage": {
            "input_tokens": response.usage.input_tokens,
            "output_tokens": response.usage.output_tokens
        }
    }

The full loop:

result = rag_query("How does HNSW indexing work?", chunk_embeddings, chunks)
print(result["answer"])
print(f"\nTokens used: {result['usage']['input_tokens']} in, {result['usage']['output_tokens']} out")
for src in result["sources"]:
    print(f"  Chunk {src['index']} (score: {src['score']:.3f}): {src['chunk'][:80]}...")

What each stage costs

For a corpus of 200 chunks (~100K words) and a single query:

Embedding the corpus — one API call, ~100K words ≈ 130K tokens × $0.02/1M = $0.003
Embedding the query — one API call, ~20 tokens × $0.02/1M ≈ $0.000
Vector search — pure numpy, no API cost, <1ms
Generation — the prompt (system text + 5 chunks ≈ 2500 tokens) plus the response (~300 tokens). At Claude Sonnet pricing ($3/1M input, $15/1M output): $0.0075 + $0.0045 = $0.012

Total per query: ~$0.012. The embedding cost is paid once at ingestion; only search and generation recur per query.

Where this breaks down

This minimal implementation has every limitation that production RAG infrastructure exists to solve:

No persistence. Embeddings live in a numpy array in memory. Restart the process and you re-embed everything.
Brute-force search. O(n) in corpus size. At 1M chunks × 1536 dims, each query scans ~6GB of vectors.
No metadata filtering. You can't restrict search to "documents from 2026" or "only the API reference." Every query searches the full corpus.
No incremental updates. Adding a document means re-embedding and rebuilding the entire array.
Naive chunking. Fixed-size word splits break mid-sentence and ignore document structure (headings, code blocks, tables).

The next lesson replaces the numpy array with pgvector in Postgres, solving persistence, filtering, and incremental updates. The lesson after that introduces frameworks that handle the chunking and retrieval pipeline with production-grade defaults.

The complete script

import numpy as np
from openai import OpenAI
import anthropic

openai_client = OpenAI()
anthropic_client = anthropic.Anthropic()

def chunk_text(text, chunk_size=500, overlap=50):
    words = text.split()
    chunks = []
    start = 0
    while start < len(words):
        end = start + chunk_size
        chunks.append(" ".join(words[start:end]))
        start += chunk_size - overlap
    return chunks

def get_embeddings(texts, model="text-embedding-3-small"):
    response = openai_client.embeddings.create(input=texts, model=model)
    return np.array([e.embedding for e in response.data])

def search(query, chunk_embeddings, chunks, top_k=5):
    query_embedding = get_embeddings([query])[0]
    similarities = chunk_embeddings @ query_embedding
    top_indices = np.argsort(similarities)[-top_k:][::-1]
    return [{"chunk": chunks[i], "score": float(similarities[i]), "index": int(i)}
            for i in top_indices]

def build_prompt(query, results):
    context = "\n\n".join(
        f"[{i+1}] (score: {r['score']:.3f})\n{r['chunk']}"
        for i, r in enumerate(results)
    )
    return f"""Answer the question based on the provided context. If the context
doesn't contain enough information, say so — do not fabricate information.
Cite the relevant context block numbers in your answer.

Context:
{context}

Question: {query}"""

def rag_query(query, chunk_embeddings, chunks, top_k=5):
    results = search(query, chunk_embeddings, chunks, top_k)
    prompt = build_prompt(query, results)
    response = anthropic_client.messages.create(
        model="claude-sonnet-4-20250514",
        max_tokens=1024,
        messages=[{"role": "user", "content": prompt}]
    )
    return {
        "answer": response.content[0].text,
        "sources": results,
        "usage": {"input_tokens": response.usage.input_tokens,
                  "output_tokens": response.usage.output_tokens}
    }

# Run it
source_text = open("knowledge_base.txt").read()
chunks = chunk_text(source_text)
chunk_embeddings = get_embeddings(chunks)
result = rag_query("How does HNSW indexing work?", chunk_embeddings, chunks)
print(result["answer"])

Fifty-eight lines from raw text to grounded answer. Every framework you'll encounter wraps these same stages — embed, search, prompt, generate — with configuration and abstractions. Understanding the bare loop means you can debug any of them.

Key takeaways

RAG has four stages: chunk, embed, search, generate. Each is a separate function with independent failure modes.
Cosine similarity on normalized vectors reduces to a dot product — similarities = embeddings @ query_vector is the entire search operation for brute-force retrieval.
Brute-force numpy search handles corpora up to ~100K chunks (<1ms per query). Beyond that, approximate nearest neighbor indexes become necessary.
The prompt template is load-bearing infrastructure. It controls whether the model uses the retrieved context or hallucinates, and whether it cites sources.
Embedding cost is paid once at ingestion; per-query cost is dominated by the generation call.

← Previous