How to reduce hallucinations with RAG

Retrieval-augmented generation is the most reliable way to reduce hallucinations with RAG, but only if you build the pipeline correctly. Most teams ship a naive vector search + prompt template and wonder why the model still invents facts. The difference between a demo and a production system comes down to three things: retrieval quality, context construction, and a feedback loop that catches failures before users do. Below is an end-to-end process you can run locally, with verification at each stage.

Step 1: Define your evaluation set before you write retrieval code

You cannot improve what you do not measure. Start with 50–100 representative questions drawn from real user traffic (or synthetic data if you have none). For each question, record the ground-truth answer and the source documents that should support it. Store this as JSONL so you can version it alongside code.

{"id": "q-001", "question": "What is the refund policy for annual plans?", "answer": "Annual plans are refundable within 30 days for a prorated amount minus a $15 processing fee.", "source_doc_ids": ["policy-2024-03", "terms-2024-01"]}
{"id": "q-002", "question": "Does the API support batch inference?", "answer": "Yes, the /v1/batch endpoint accepts up to 10,000 requests per file with 24-hour turnaround.", "source_doc_ids": ["api-reference-2024-06"]}

Save this as eval/questions.jsonl. This becomes your regression test. Every pipeline change runs against it.

Step 2: Chunk with overlap and preserve document hierarchy

Naive fixed-size chunking destroys context. Use a structure-aware splitter that respects headings, tables, and code blocks, then apply a sliding window with 10–15% overlap. The goal: every chunk should be self-contained enough to answer a question without its neighbors, but overlapping enough that no answer falls in the gap.

# chunking.py
from langchain_text_splitters import MarkdownHeaderTextSplitter, RecursiveCharacterTextSplitter

header_splitter = MarkdownHeaderTextSplitter(
    headers_to_split_on=[
        ("#", "h1"), ("##", "h2"), ("###", "h3"),
    ]
)

def chunk_document(markdown: str, doc_id: str) -> list[dict]:
    sections = header_splitter.split_text(markdown)
    chunks = []
    for section in sections:
        # Each section keeps its header path in metadata
        header_path = " > ".join([
            section.metadata.get("h1", ""),
            section.metadata.get("h2", ""),
            section.metadata.get("h3", ""),
        ]).strip(" >")
        # Fine-grained split inside section
        fine_splitter = RecursiveCharacterTextSplitter(
            chunk_size=512,
            chunk_overlap=64,
            separators=["\n\n", "\n", ". ", " ", ""],
        )
        for i, chunk in enumerate(fine_splitter.split_text(section.page_content)):
            chunks.append({
                "doc_id": doc_id,
                "chunk_id": f"{doc_id}-{section.metadata.get('h1','')}-{i}".replace(" ", "-").lower(),
                "text": chunk,
                "metadata": {
                    "header_path": header_path,
                    "source": section.metadata,
                },
            })
    return chunks

Verify: Spot-check 10 chunks. Each should read like a coherent paragraph with its section header visible in metadata. No mid-sentence cuts, no orphaned list items.

Step 3: Embed with a model matched to your domain

General-purpose embeddings (text-embedding-3-small, bge-small-en) work for broad corpora. If your documents are legal, medical, financial, or code-heavy, use a domain-adapted model. The embedding model determines retrieval ceiling — no reranker can recover what the embedder never distinguished.

# embed.py
from sentence_transformers import SentenceTransformer
import numpy as np

# Example: legal-trained model
model = SentenceTransformer("mixedbread-ai/mxbai-embed-large-v1", device="cuda")

def embed_chunks(chunks: list[dict]) -> np.ndarray:
    texts = [c["text"] for c in chunks]
    # Normalize for cosine similarity
    embeddings = model.encode(texts, normalize_embeddings=True, batch_size=64, show_progress_bar=True)
    return embeddings.astype(np.float32)

Persist embeddings to a vector index (FAISS, Qdrant, pgvector). Store the model name and version in the index metadata — you will need it for reproducibility.

Verify: Run a quick nearest-neighbor sanity check. Pick 5 questions from your eval set, embed them, and confirm the top-5 results contain the expected source_doc_ids at least 80% of the time. If not, your embedder or chunking is the bottleneck — fix there before adding reranking.

Step 4: Hybrid search (dense + sparse) beats pure vector search

Vector search misses exact terminology, acronyms, and rare tokens. Add BM25 over the same chunks and fuse results with Reciprocal Rank Fusion (RRF). This is low-effort, high-impact.

# hybrid_search.py
from rank_bm25 import BM25Okapi
import numpy as np

class HybridRetriever:
    def __init__(self, chunks: list[dict], dense_index, embed_model):
        self.chunks = chunks
        self.dense_index = dense_index
        self.embed_model = embed_model
        # BM25 corpus
        tokenized = [c["text"].lower().split() for c in chunks]
        self.bm25 = BM25Okapi(tokenized)
        self.chunk_id_to_idx = {c["chunk_id"]: i for i, c in enumerate(chunks)}

    def search(self, query: str, k: int = 50, rrf_k: int = 60) -> list[dict]:
        # Dense
        q_emb = self.embed_model.encode([query], normalize_embeddings=True).astype(np.float32)
        dense_scores, dense_idxs = self.dense_index.search(q_emb, k)
        dense_rank = {self.chunks[i]["chunk_id"]: rank for rank, i in enumerate(dense_idxs[0])}

        # Sparse
        bm25_scores = self.bm25.get_scores(query.lower().split())
        sparse_rank = {self.chunks[i]["chunk_id"]: rank for rank, i in enumerate(np.argsort(bm25_scores)[::-1][:k])}

        # RRF fusion
        all_ids = set(dense_rank) | set(sparse_rank)
        fused = []
        for cid in all_ids:
            dr = dense_rank.get(cid, k)
            sr = sparse_rank.get(cid, k)
            score = 1.0 / (rrf_k + dr + 1) + 1.0 / (rrf_k + sr + 1)
            fused.append((score, cid))
        fused.sort(reverse=True)
        return [self.chunks[self.chunk_id_to_idx[cid]] for _, cid in fused[:k]]

Verify: On your eval set, compare recall@10 for dense-only vs hybrid. Hybrid should improve recall by 10–20 points on keyword-heavy queries (product names, error codes, section numbers) without hurting semantic queries.

Step 5: Rerank with a cross-encoder

Bi-encoders (embedding models) score query and document independently. Cross-encoders attend jointly, capturing fine-grained relevance. Rerank the top-50 hybrid results down to top-5–8 for the context window. This is the single biggest quality lever after hybrid search.

# rerank.py
from sentence_transformers import CrossEncoder

reranker = CrossEncoder("cross-encoder/ms-marco-MiniLM-L-6-v2", device="cuda", max_length=512)

def rerank(query: str, candidates: list[dict], top_n: int = 8) -> list[dict]:
    pairs = [[query, c["text"]] for c in candidates]
    scores = reranker.predict(pairs, batch_size=32, show_progress_bar=False)
    ranked = sorted(zip(scores, candidates), key=lambda x: x[0], reverse=True)
    return [c for _, c in ranked[:top_n]]

Verify: Measure answer correctness on your eval set with and without reranking (using the same generator prompt). Expect a 15–30% relative reduction in hallucinated claims. Log the reranker scores — if top results consistently score < 0.3, your retrieval is still returning junk; go back to Step 3 or 4.

Step 6: Build context with citations and token budgets

Stuffing all retrieved chunks into the prompt wastes tokens and dilutes attention. Instead: (1) truncate each chunk to a max length, (2) prepend the header path, (3) assign a citation ID, (4) pack into the context window with a hard token budget, (5) instruct the model to cite inline.

# context.py
import tiktoken

enc = tiktoken.encoding_for_model("gpt-4o")

def build_context(chunks: list[dict], max_tokens: int = 3000, max_chunk_tokens: int = 400) -> tuple[str, list[dict]]:
    """Returns (context_string, citation_map)"""
    cited_chunks = []
    total = 0
    for i, chunk in enumerate(chunks):
        header = chunk["metadata"].get("header_path", "")
        text = chunk["text"]
        # Truncate chunk
        chunk_tokens = enc.encode(text)
        if len(chunk_tokens) > max_chunk_tokens:
            text = enc.decode(chunk_tokens[:max_chunk_tokens]) + "..."
        citation_id = f"[{i+1}]"
        block = f"{citation_id} {header}\n{text}\n"
        block_tokens = len(enc.encode(block))
        if total + block_tokens > max_tokens:
            break
        cited_chunks.append({**chunk, "citation_id": citation_id, "display_text": block})
        total += block_tokens
    context = "\n".join(c["display_text"] for c in cited_chunks)
    return context, cited_chunks

Prompt template (trimmed for clarity):

You are a precise technical assistant. Answer the question using ONLY the provided context.
Cite every claim with the citation ID in square brackets, e.g., [1], [2].
If the context does not contain the answer, say "I don't know based on the provided documents."
Do not use external knowledge.

Context:
{context}

Question: {question}

Answer:

Verify: Generate answers for your eval set. Check two things: (1) every factual claim has a citation, (2) no citation references a chunk that doesn’t support the claim. Automate this with a simple citation validator (see Step 8).

Step 7: Add a verification step — self-consistency or a critic model

Single-pass generation still hallucinates. Run the generator 3 times with temperature 0.3–0.5, then take the majority-vote answer (self-consistency). Or use a smaller critic model to score faithfulness: “Does the answer contradict any cited source? Score 0–1.”

# verify.py
from openai import OpenAI

client = OpenAI()

CRITIC_PROMPT = """You are a fact-checker. Given a question, an answer with citations, and the source context,
rate the answer's faithfulness on a scale of 0.0 to 1.0.
- 1.0: Every claim is directly supported by the cited sources.
- 0.5: Some claims are supported, others are not in the sources or contradict them.
- 0.0: The answer is largely fabricated or contradicts the sources.

Return only a JSON object: {"score": 0.0, "reason": "..."}"""

def critic_score(question: str, answer: str, context: str) -> float:
    resp = client.chat.completions.create(
        model="gpt-4o-mini",
        messages=[
            {"role": "system", "content": CRITIC_PROMPT},
            {"role": "user", "content": f"Question: {question}\n\nAnswer: {answer}\n\nContext: {context}"},
        ],
        temperature=0,
        response_format={"type": "json_object"},
    )
    import json
    return json.loads(resp.choices[0].message.content)["score"]

Verify: On your eval set, correlate critic scores with human labels. If correlation > 0.8, you can use the critic as an automated gate: reject answers below 0.7 and trigger a fallback (retrieve more, escalate to human, return “I don’t know”).

Step 8: Close the loop with automated regression testing

Wire the full pipeline into a nightly (or per-PR) test that runs your eval set and fails on regression. Track three metrics:

Faithfulness (critic score or human-labeled): % of answers where every claim is supported
Answer relevance: % of questions where the answer addresses the question (not “I don’t know” when context exists)
Citation precision: % of citations that actually support the attached claim

# test_rag.py
import json
from pipeline import rag_answer  # your assembled pipeline

def run_eval():
    with open("eval/questions.jsonl") as f:
        cases = [json.loads(line) for line in f]

    results = []
    for case in cases:
        answer, context, citations = rag_answer(case["question"])
        score = critic_score(case["question"], answer, context)
        results.append({
            "id": case["id"],
            "faithfulness": score,
            "answer": answer,
        })

    faithful = sum(1 for r in results if r["faithfulness"] >= 0.7)
    print(f"Faithfulness@0.7: {faithful}/{len(results)} = {faithful/len(results):.1%}")

    # Fail if regression
    if faithful / len(results) < 0.85:
        raise AssertionError("Faithfulness dropped below 85%")

    # Save for dashboards
    with open("eval/results_latest.jsonl", "w") as f:
        for r in results:
            f.write(json.dumps(r) + "\n")

if __name__ == "__main__":
    run_eval()

Verify: Run this locally before every merge. The threshold (85% here) should be set from your baseline — don’t cargo-cult a number. The key is that the test fails when quality drops, forcing you to investigate.

Step 9: Monitor production with the same signals

Your eval set is a snapshot. Production traffic drifts. Log every request with: question, retrieved chunk IDs, reranker scores, generated answer, critic score, and user feedback (thumbs up/down, or implicit signals like follow-up clarification). Build a dashboard with:

Daily faithfulness trend (critic score)
Retrieval recall proxy: % of requests where reranker top-1 score > 0.5
“I don’t know” rate (should be low but non-zero — zero means you’re overconfident)
Citation click-through if you render sources in UI

When faithfulness dips, you have the artifacts to debug: pull the low-scoring cases, inspect retrieved chunks, check if new document types broke chunking or embedding.

Step 10: Iterate on the right component

When quality is low, the fix is rarely “better prompt.” Use this decision tree:

Symptom	Likely cause	Fix
Correct answer not in top-50 hybrid	Embedding model mismatch or chunking too coarse	Step 2 or 3
Correct chunk retrieved but reranked low	Cross-encoder domain mismatch	Fine-tune reranker on your data
Correct chunk in context but answer wrong	Prompt ambiguity or context overflow	Step 6 (budget, citation format)
Answer cites correct chunk but hallucinates detail	Generator model too weak or temperature too high	Lower temp, stronger model, or add critic gate
High “I don’t know” on answerable questions	Retrieval recall too low or critic threshold too aggressive	Lower critic threshold, expand k

Each iteration should move a metric on your eval set. If it doesn’t, revert.

What “done” looks like

You have reduced hallucinations with RAG when:

Your nightly eval passes faithfully at your target threshold (e.g., ≥90% critic score ≥0.7).
Production critic scores track within 5 points of eval scores — no silent drift.
When a user reports a hallucination, you can reproduce it in the eval set within an hour, add it as a regression case, and fix the pipeline without guessing.
The team trusts the system enough to put it in front of customers without a human-in-the-loop for every response.

The pipeline above is not theoretical — it’s the architecture that separates prototypes that hallucinate 30% of the time from systems that hallucinate <2%. The work is in the eval set, the chunking, the hybrid retrieval, the reranker, and the critic gate. Everything else is plumbing.