Build a basic RAG pipeline using embeddings and a vector database.

What you need to know

Four ideas carry the whole pipeline.

• Chunking. A 40-page policy PDF is too big for one prompt and too broad for one vector. You cut it into pieces of a few hundred characters or tokens. A small overlap between neighbouring chunks means a sentence cut at a boundary still appears whole in one of them.
• Embeddings. An embedding model turns text into a list of numbers (a vector) so that texts with similar meaning get vectors pointing in similar directions. The same model must embed both the chunks and the query, or the two sets of vectors are not comparable.
• Normalisation. If you divide each vector by its length once, at index time, then the dot product of two vectors is their cosine similarity. Search becomes one matrix-vector product.
• Grounded prompt. The retrieved chunks go into the prompt with numbers, plus an instruction to answer only from them and to say "I don't know" otherwise. That instruction is what turns retrieval into fewer hallucinations.

The "vector database" in an interview can be a NumPy matrix. Say that out loud, and say that Chroma, pgvector or Pinecone replace only the index and retrieve methods.

1from collections.abc import Callable2import numpy as np34EmbedFn = Callable[[list[str]], list[list[float]]]56def chunk_text(text: str, size: int, overlap: int) -> list[str]:7    """Fixed-size character chunks; each starts `size - overlap` after the last."""8    step, start, pieces = size - overlap, 0, []9    while start < len(text):10        piece = text[start:start + size].strip()11        if piece:12            pieces.append(piece)13        if start + size >= len(text):        # this chunk reached the end: stop14            break15        start += step16    return pieces1718class RAGPipeline:19    """Chunk -> embed -> store -> retrieve -> generate, with an in-memory index."""2021    def __init__(self, embed_fn: EmbedFn, llm_fn: Callable[[str], str], top_k: int = 4):22        self.embed_fn, self.llm_fn, self.top_k = embed_fn, llm_fn, top_k23        self.chunks: list[str] = []24        self.matrix: np.ndarray | None = None          # (n, d), rows have length 12526    def index(self, docs: list[str], size: int = 500, overlap: int = 50) -> None:27        if not 0 <= overlap < size:28            raise ValueError("need 0 <= overlap < size")29        known = set(self.chunks)30        new = [c for d in docs if d for c in chunk_text(d, size, overlap)]31        new = [c for c in dict.fromkeys(new) if c not in known]   # dedupe, keep order32        if not new:33            return34        v = np.asarray(self.embed_fn(new), dtype=np.float32)35        v /= np.linalg.norm(v, axis=1, keepdims=True) + 1e-1036        self.matrix = v if self.matrix is None else np.vstack([self.matrix, v])37        self.chunks.extend(new)3839    def retrieve(self, query: str) -> list[tuple[str, float]]:40        if self.matrix is None:41            return []42        q = np.asarray(self.embed_fn([query])[0], dtype=np.float32)43        scores = self.matrix @ (q / (np.linalg.norm(q) + 1e-10))44        top = np.argsort(-scores, kind="stable")[: self.top_k]45        return [(self.chunks[i], float(scores[i])) for i in top]4647    def answer(self, query: str) -> str:48        hits = self.retrieve(query)49        if not hits:50            return "No indexed context available."51        context = "\n\n".join(f"[{n}] {c}" for n, (c, _) in enumerate(hits, 1))52        return self.llm_fn(53            "Answer only from the context below and cite sources as [n]. If the "54            "context does not contain the answer, say you don't know.\n\n"55            f"Context:\n{context}\n\nQuestion: {query}\nAnswer:"56        )

The tricky lines:

• The stop condition in chunk_text. A plain for i in range(0, len(text), step) keeps going after a chunk has already reached the end, and emits a last fragment that is fully inside the previous chunk. Stopping as soon as start + size >= len(text) removes that duplicate.
• + 1e-10 stops a division by zero when a text embeds to an all-zero vector.
• Only new chunks are embedded. Calling index twice does not re-pay for the first batch, and exact duplicate chunks are dropped before they waste an embedding call and a context slot.
• kind="stable" makes ties come back in insertion order, so tests are repeatable.

Complexity: chunking is O(L) for L characters. Indexing is one embedding call for n chunks plus O(n·d) to normalise, and the index holds n·d floats. A query is O(n·d) for the product plus O(n log n) for the sort (use argpartition for O(n) when n is large).

A real-life example

First the chunker on a 10-character string, chunk_text("abcdefghij", size=4, overlap=1), so the step is 3:

Result: ['abcd', 'defg', 'ghij']. The older range version returns ['abcd', 'defg', 'ghij', 'j'] — the stray 'j' is the bug.

Now retrieval, with a toy embedder that counts three keywords so every number can be checked by hand:

1VOCAB = ["refund", "delivery", "password"]23def toy_embed(texts: list[str]) -> list[list[float]]:4    """A 3-dimensional 'embedding': how often each keyword appears."""5    return [[float(t.lower().count(w)) for w in VOCAB] for t in texts]67rag = RAGPipeline(toy_embed, llm_fn=lambda prompt: prompt, top_k=2)8rag.index([9    "Refunds reach your UPI account in 5 days. A refund needs the order id.",10    "Delivery delays can trigger a refund.",11    "Reset your password from the login page.",12])13for text, score in rag.retrieve("How many days does a refund take?"):14    print(f"{score:.3f}  {text}")15# 1.000  Refunds reach your UPI account in 5 days. A refund needs the order id.16# 0.707  Delivery delays can trigger a refund.

Step by step: the three chunks embed to [2,0,0], [1,1,0] and [0,0,1]. After normalising they are [1,0,0], [0.707,0.707,0] and [0,0,1]. The query embeds to [1,0,0], so the dot products are 1.0, 0.707 and 0.0, and the top two go into the prompt as [1] and [2].

This is the exact shape of a food-delivery support bot that answers "where is my refund?" from the refund-policy pages instead of from the model's memory.

Follow-up questions to expect

• "What changes at 10 million chunks?" — Brute force at 768 dimensions is about 30 GB of float32 and a full scan per query. I'd move to an approximate index (HNSW or IVF) in a vector store, and accept a small recall loss for millisecond queries.
• "How do you handle a document that is edited?" — Store a document id and a content hash with every chunk. When the hash changes, delete that document's chunks and re-index only them.
• "What if nothing relevant is retrieved?" — Add a minimum score. Below it, skip the model call and say you have no source, which is cheaper and more honest than a guessed answer.
• "How would you test it?" — With a deterministic fake embedder like the one above: assert the known-relevant chunk ranks first and that the prompt contains its text. For quality, measure recall@k on a labelled question set.