Retrieval-augmented generation (RAG)

What RAG is, and when to use it

Retrieval-augmented generation (RAG) has two phases. Ahead of time, you process your documents into a searchable index. At question time, you search that index for the passages most relevant to the question, place them in the prompt, and instruct the model to answer using only that material. The model's role changes from reciting from memory to reading and summarising, which it does far more reliably.

The rule of thumb is that RAG supplies knowledge, and fine-tuning supplies behaviour. If the problem is that the model does not know your facts, retrieve them. RAG has three further advantages that matter in production. Updating knowledge is just re-indexing a document. Every answer can cite the passage it came from, so a person can verify it. And you can enforce access control, by retrieving only documents that the asking user is permitted to see. That last point is impossible once the facts have been baked into model weights.

Indexing: load, chunk, embed, store

Loading means turning sources such as PDFs, wiki pages, tickets and code into clean text. It is unglamorous, and it determines your quality ceiling: tables flattened into nonsense, repeated page headers and footers, and navigation menus embedded as if they were content are the quiet causes of many poor RAG systems. Keep the metadata as you go: the source URL, title, section heading, last-modified date, owner and access permissions.

Chunking splits the text into passages, because an embedding represents one idea well and ten ideas badly, and because you want to retrieve the relevant paragraph and not a forty-page manual. It is the design decision with the greatest effect on results.

def chunk_text(text: str, max_chars: int = 1200, overlap: int = 200) -> list[str]:
    """Split on paragraphs, packing them into chunks of up to max_chars with overlap.
    A simple illustration: a single paragraph longer than max_chars is kept whole."""
    paragraphs = [p.strip() for p in text.split("\n\n") if p.strip()]
    chunks, current = [], ""
    for para in paragraphs:
        if current and len(current) + len(para) + 2 > max_chars:
            chunks.append(current)
            current = current[-overlap:]          # carry the tail into the next chunk
        current = f"{current}\n\n{para}" if current else para
    if current:
        chunks.append(current)
    return chunks

A chunk cut out of its document frequently loses its meaning. "Set it to 30 seconds" is useless without knowing what "it" is. A cheap and effective remedy is to put context at the front of each chunk before embedding it: the document title and the section heading, or a one-sentence description of where the chunk sits within the document.

import numpy as np

# embed is a stand-in: list[str] -> np.ndarray of shape (n, dimension)


def build_index(documents: list[dict]) -> dict:
    records = []
    for doc in documents:
        for i, chunk in enumerate(chunk_text(doc["text"])):
            records.append({
                "id": f"{doc['id']}#{i}",
                "text": chunk,
                "embed_text": f"{doc['title']}\n\n{chunk}",   # context added for embedding
                "source": doc["url"],
                "updated": doc["updated"],
                "team": doc["team"],
            })
    vectors = embed([r["embed_text"] for r in records])
    vectors = vectors / np.linalg.norm(vectors, axis=1, keepdims=True)   # normalise once
    return {"records": records, "vectors": vectors}

For a few thousand chunks, a NumPy array held in memory is a perfectly good vector store. Beyond that, use a vector database, or the vector extension of a database you already run, such as pgvector for PostgreSQL. These use approximate nearest-neighbour indexes, commonly HNSW, that trade a little accuracy for a very large gain in speed, and they give you filtering, persistence and updates. Whichever you use, plan for re-indexing: documents change and get deleted, and if you ever change the embedding model, every vector must be regenerated.

Querying: retrieve, assemble, answer

import numpy as np


def retrieve(index: dict, question: str, k: int = 5, team: str | None = None) -> list[dict]:
    query = embed([question])[0]
    query = query / np.linalg.norm(query)
    scores = index["vectors"] @ query                      # cosine similarity to every chunk

    if team is not None:                                   # metadata filter
        allowed = np.array([r["team"] == team for r in index["records"]])
        scores = np.where(allowed, scores, -np.inf)

    top = np.argsort(-scores)[:k]
    return [{**index["records"][i], "score": float(scores[i])}
            for i in top if np.isfinite(scores[i])]

The prompt that assembles the results matters as much as the search does. It should tell the model to answer only from the supplied passages, to cite them, and to say so when they do not contain the answer.

SYSTEM = """You answer questions for on-call engineers, using only the passages provided
in <passages>. Each passage has an id.

- Base every statement on the passages, and cite the id in square brackets after it, like [kb-12#3].
- If the passages do not contain the answer, say that you could not find it in the
  documentation, and suggest what to search for. Do not answer from general knowledge.
- If passages disagree, say so, and prefer the most recently updated one.
- Be concise. Give commands exactly as they are written in the passages."""


def answer(index: dict, question: str) -> dict:
    passages = retrieve(index, question, k=5)
    block = "\n\n".join(
        f'<passage id="{p["id"]}" source="{p["source"]}" updated="{p["updated"]}">\n{p["text"]}\n</passage>'
        for p in passages
    )
    messages = [
        {"role": "system", "content": SYSTEM},
        {"role": "user", "content": f"<passages>\n{block}\n</passages>\n\nQuestion: {question}"},
    ]
    text = call_llm(messages, temperature=0, max_tokens=600)   # stand-in for your provider
    return {"answer": text, "sources": [p["source"] for p in passages]}

• Citations make answers checkable. Show the sources to the user, and in code confirm that every cited id was among the passages you supplied. A citation of a passage that does not exist is a hallucination you can detect automatically.
• k is a trade-off. Too few passages, and the answer may not be among them. Too many, and you pay for noise that can distract the model. Five to ten is a common starting range, and a reranker lets you cast a wider net safely.
• Retrieved text is untrusted input. If anyone can write to a document that you index, they can plant instructions there. Keep the passages inside tags, state that they are reference material, and give a RAG system no ability to act that it does not need. The agents module covers prompt injection in detail.
• Apply permissions at retrieval time, by filtering on the asking user's access rights, and never rely on the model to withhold something it has been shown.

Making retrieval better

When a RAG system gives a bad answer, the cause is usually that the right passage was never retrieved. The model cannot use what it was not given. These are the improvements that pay off most often, roughly in order.

• Hybrid search. Embeddings capture meaning, and are weak on exact terms: error codes, product names, function names, ticket numbers. Classic keyword search, usually the BM25 algorithm, is the reverse. Run both and merge the rankings. This often gives the largest single improvement, particularly for technical content.
• Reranking. Retrieve a generous set of candidates, say 50, with fast vector search, then use a reranker, a model that reads the question and each passage together and scores their relevance much more accurately than a comparison of vectors can. Keep the best 5. It costs some latency, and usually improves precision noticeably.
• Metadata filters. Restrict the search by product, version, language, team or date before ranking. "How do I configure TLS in version 3?" should not retrieve the documentation for version 1.
• Query rewriting. Users write short, vague, context-dependent questions. In a conversation, "and how do I roll it back?" is unsearchable by itself. Use a cheap model call to rewrite the question as a standalone query, using the conversation history, before you search.
• Better chunks. Revisit the chunk size, follow the structure of the documents, and add contextual headers. Inspect what is actually in your index; you will find rubbish.
• Freshness and duplicates. Remove near-identical chunks. Prefer recent documents when several match, and delete or mark documents that have been superseded.

def reciprocal_rank_fusion(rankings: list[list[str]], k: int = 60) -> list[str]:
    """Merge several ranked lists of chunk ids into one. A common way to combine
    keyword and vector results without having to compare their raw scores."""
    scores: dict[str, float] = {}
    for ranking in rankings:
        for rank, chunk_id in enumerate(ranking):
            scores[chunk_id] = scores.get(chunk_id, 0.0) + 1.0 / (k + rank + 1)
    return sorted(scores, key=scores.get, reverse=True)


# fused = reciprocal_rank_fusion([vector_ids, keyword_ids])[:50]   -> then rerank to the top 5

Evaluating a RAG system

A RAG pipeline has two components that fail independently, so measure them separately. If you judge only the final answers, you cannot tell whether to work on the search or on the prompt.

RETRIEVAL: was the right passage found?
  recall@k     = share of questions where a relevant chunk appears in the top k
  precision@k  = share of the top k chunks that are relevant
  MRR          = mean of 1 / (rank of the first relevant chunk)

  Example: for three questions, the first relevant chunk is at rank 1, rank 3, and not in the top 5
    recall@5 = 2 / 3 = 0.67
    MRR      = (1/1 + 1/3 + 0) / 3 = 0.44

GENERATION: given the passages, was the answer good?
  faithfulness  is every claim supported by the retrieved passages?  (the hallucination check)
  relevance     does it actually answer the question that was asked?
  correctness   does it match a known-good reference answer?
  abstention    when the answer is not in the documents, does it say so?

def recall_at_k(index: dict, cases: list[dict], k: int = 5) -> float:
    """cases: [{'question': ..., 'relevant_ids': ['kb-12#3', ...]}, ...]"""
    hits = 0
    for case in cases:
        retrieved = {p["id"] for p in retrieve(index, case["question"], k=k)}
        hits += bool(retrieved & set(case["relevant_ids"]))
    return hits / len(cases)


def mean_reciprocal_rank(index: dict, cases: list[dict], k: int = 10) -> float:
    total = 0.0
    for case in cases:
        ids = [p["id"] for p in retrieve(index, case["question"], k=k)]
        ranks = [i + 1 for i, cid in enumerate(ids) if cid in case["relevant_ids"]]
        total += 1.0 / ranks[0] if ranks else 0.0
    return total / len(cases)

• Build a test set of real questions, each labelled with the chunks that answer it. Thirty to fifty are enough to begin with. Take them from support tickets, search logs and chat history, not from your imagination.
• Include questions whose answer is not in the corpus, to test that the system admits as much.
• Fix retrieval first. If recall@k is low, no prompt will rescue the answers. If recall is high and the answers are still poor, work on the prompt, on k, or on the model.
• Faithfulness and relevance are usually graded by another model call with a rubric. The next module explains how to do that reliably.
• Re-run the evaluation whenever you change the chunking, the embedding model, k, the reranker or the prompt. Every one of them moves the numbers.
• In production, log the question, the ids retrieved, the answer and the user's feedback. Questions that retrieve nothing useful tell you which documentation is missing.