How many chunks should I retrieve?

Start with 3 to 6 and test. Too few risks missing the answer, and too many adds noise and cost. Reranking can let you retrieve more candidates and keep only the best few.

Why enforce access control before retrieval?

If restricted text is retrieved and put in the prompt, the model may reveal it. Filtering first guarantees the model never sees content the user should not access.

RAG Step 3: Building a Reusable Retrieval Pipeline

Introduction

The previous lesson used loose variables. In a real project you want a reusable component that indexes chunks, searches them, filters by metadata, and persists to disk so you do not re-embed on every run. This lesson builds a minimal vector store you fully understand.

Prerequisites

Embeddings and vector search lesson

Step-by-Step Guide

Step 1: Create a minimal vector store

The store keeps texts, metadata, and a matrix of normalized vectors. Adding chunks embeds and appends them, and searching scores the query against all stored vectors.

import numpy as np
from sentence_transformers import SentenceTransformer

class VectorStore:
    def __init__(self, model_name: str = "all-MiniLM-L6-v2"):
        self.model = SentenceTransformer(model_name)
        dim = self.model.get_sentence_embedding_dimension()
        self.vectors = np.empty((0, dim), dtype="float32")
        self.records: list[dict] = []

    def add(self, records: list[dict]) -> None:
        texts = [r["text"] for r in records]
        new = self.model.encode(texts, normalize_embeddings=True)
        self.vectors = np.vstack([self.vectors, new])
        self.records.extend(records)

Step 2: Add search with a minimum score

Returning only results above a similarity threshold prevents the model from being shown irrelevant text when nothing in your documents matches. The right threshold depends on your embedding model and data, so tune it using real questions.

    def search(self, query: str, k: int = 4, min_score: float = 0.25) -> list[dict]:
        q = self.model.encode([query], normalize_embeddings=True)[0]
        scores = self.vectors @ q
        order = np.argsort(scores)[::-1][:k]
        results = []
        for i in order:
            if scores[i] >= min_score:
                results.append({**self.records[i], "score": float(scores[i])})
        return results

Step 3: Filter by metadata

Restrict the search to a subset, such as one department or documents a user is allowed to read, by scoring only the matching rows. Apply access control before retrieval, not after, so restricted text never reaches the model.

    def search_filtered(self, query: str, allowed_sources: set[str], k: int = 4):
        q = self.model.encode([query], normalize_embeddings=True)[0]
        indices = [i for i, r in enumerate(self.records) if r["source"] in allowed_sources]
        if not indices:
            return []
        scores = self.vectors[indices] @ q
        best = np.argsort(scores)[::-1][:k]
        return [{**self.records[indices[j]], "score": float(scores[j])} for j in best]

Step 4: Save and load the index

Embedding is the slow part, so persist the vectors and records. Store the embedding model name with the index so you never query it with a different model by accident.

    def save(self, folder: str) -> None:
        import json, pathlib
        path = pathlib.Path(folder)
        path.mkdir(exist_ok=True)
        np.save(path / "vectors.npy", self.vectors)
        (path / "records.json").write_text(json.dumps(self.records), encoding="utf-8")

Step 5: Use it

Index once, then search many times. Keep the store behind a small interface so you can later swap in a vector database without changing the rest of your application.

store = VectorStore()
store.add(records)
store.save("index")

for hit in store.search("What is the refund policy?"):
    print(round(hit["score"], 2), hit["source"], hit["text"][:100])

FAQs

How many chunks should I retrieve?: Start with 3 to 6 and test. Too few risks missing the answer, and too many adds noise and cost. Reranking can let you retrieve more candidates and keep only the best few.
Why enforce access control before retrieval?: If restricted text is retrieved and put in the prompt, the model may reveal it. Filtering first guarantees the model never sees content the user should not access.

Summary

A small vector store with metadata, score thresholds, filtering, and persistence is enough for many applications. Keep it behind a clean interface so you can scale it later.