RAG that actually works: chunking, hybrid search, reranking and citations

Retrieval-augmented generation sounds simple: embed your documents, find the closest chunks to a question and pass them to the model. The first prototype usually works well enough to impress. Then real users arrive with acronyms, product codes, vague questions and questions that span three documents, and the answers start to drift.

When we debug a RAG system, the problem is almost always that the right text never reached the model. Generation quality is capped by retrieval quality. This guide covers the pieces that move retrieval quality the most: how you chunk, how you search, how you rerank and how you make answers verifiable with citations.

Measure retrieval separately from answers

Before tuning anything, build a small retrieval eval: a list of real questions, each paired with the document sections that contain the answer. Then measure recall@k, the share of questions where at least one correct section appears in the top k results. If recall@10 is poor, no prompt will save you. If recall is good but answers are bad, the problem is in generation.

from dataclasses import dataclass

@dataclass
class Case:
    question: str
    relevant_ids: set[str]  # chunk or section ids that contain the answer

def recall_at_k(cases: list[Case], search, k: int = 10) -> float:
    hits = 0
    for case in cases:
        results = search(case.question, limit=k)
        if case.relevant_ids & {r.chunk_id for r in results}:
            hits += 1
    return hits / len(cases)

# Compare configurations on the same cases:
# print(recall_at_k(cases, vector_search), recall_at_k(cases, hybrid_search))

Fifty well-chosen questions are enough to compare chunking strategies and search configurations with confidence. Pull them from real support tickets or search logs where you can, since questions written by the people who wrote the docs tend to reuse the docs' own wording and make retrieval look better than it is.

Chunking: follow the structure of the document

Fixed-size chunks of N tokens are easy to implement and frequently cut a table in half or separate a heading from the paragraph that explains it. Better chunks follow the document's own structure.

• Split on headings first, then on paragraphs, and only fall back to token windows for very long sections.
• Prepend the heading path to every chunk, for example Billing > Refunds > Partial refunds. This alone often fixes chunks that are meaningless out of context.
• Keep tables, code blocks and lists intact. Split large tables by rows and repeat the header row in each piece.
• Store metadata with every chunk: source URL, document title, section, last updated date and access permissions.
• Use a modest overlap between adjacent windows so sentences at boundaries are not lost.

import re
from dataclasses import dataclass, field

@dataclass
class Chunk:
    chunk_id: str
    doc_id: str
    heading_path: list[str]
    text: str
    metadata: dict = field(default_factory=dict)

HEADING = re.compile(r"^(#{1,6})\s+(.*)$")

def chunk_markdown(doc_id: str, markdown: str, max_chars: int = 2400) -> list[Chunk]:
    chunks: list[Chunk] = []
    path: list[str] = []
    buffer: list[str] = []

    def flush() -> None:
        body = "\n".join(buffer).strip()
        buffer.clear()
        if not body:
            return
        for i in range(0, len(body), max_chars):
            piece = body[max(0, i - 200): i + max_chars]  # ~200 chars overlap
            prefix = " > ".join(path)
            chunks.append(Chunk(
                chunk_id=f"{doc_id}:{len(chunks)}",
                doc_id=doc_id,
                heading_path=list(path),
                text=f"{prefix}\n\n{piece}" if prefix else piece,
            ))

    for line in markdown.splitlines():
        match = HEADING.match(line)
        if match:
            flush()
            level = len(match.group(1))
            path[:] = path[: level - 1] + [match.group(2).strip()]
        else:
            buffer.append(line)
    flush()
    return chunks

Hybrid search: vectors plus keywords

Embedding search is good at meaning: “how do I get my money back” finds the refunds page. It is weak at exact tokens: error codes, SKUs, names and rare acronyms. Classic full-text search is the opposite. Running both and merging the results consistently beats either one alone on real-world queries.

If you already run Postgres, you do not need a separate search cluster to start. pgvector handles embeddings and the built-in tsvector handles keyword search. Merge the two ranked lists with reciprocal rank fusion (RRF), which only uses rank positions and therefore does not require the two scoring scales to be comparable.

-- Schema (once)
-- create extension if not exists vector;
-- alter table chunks add column embedding vector(1536);
-- alter table chunks add column tsv tsvector
--   generated always as (to_tsvector('english', text)) stored;
-- create index on chunks using hnsw (embedding vector_cosine_ops);
-- create index on chunks using gin (tsv);

with semantic as (
  select id, row_number() over (order by embedding <=> $1::vector) as rank
  from chunks
  where tenant_id = $3
  order by embedding <=> $1::vector
  limit 40
),
keyword as (
  select id, row_number() over (
    order by ts_rank_cd(tsv, websearch_to_tsquery('english', $2)) desc
  ) as rank
  from chunks
  where tenant_id = $3 and tsv @@ websearch_to_tsquery('english', $2)
  limit 40
)
select c.id, c.doc_id, c.text, c.metadata,
       coalesce(1.0 / (60 + s.rank), 0) + coalesce(1.0 / (60 + k.rank), 0) as score
from chunks c
left join semantic s on s.id = c.id
left join keyword k on k.id = c.id
where s.id is not null or k.id is not null
order by score desc
limit 30;

Reranking: spend compute on the shortlist

First-stage search is optimised for speed across millions of chunks, so it is approximate. A reranker reads the question and each candidate together and scores how well that candidate actually answers the question. Retrieve 30 to 50 candidates cheaply, rerank them, and pass the top 5 to 8 to the model.

You can use a dedicated cross-encoder model, a hosted rerank API or an LLM. A cross-encoder is the usual choice for latency and cost. The example below uses an open cross-encoder through sentence-transformers; swap in whichever reranker fits your stack.

from sentence_transformers import CrossEncoder

reranker = CrossEncoder("cross-encoder/ms-marco-MiniLM-L-6-v2")

def rerank(question: str, candidates: list[dict], top_n: int = 6, min_score: float = 0.0) -> list[dict]:
    if not candidates:
        return []
    pairs = [(question, c["text"]) for c in candidates]
    scores = reranker.predict(pairs)
    ranked = sorted(zip(candidates, scores), key=lambda pair: pair[1], reverse=True)
    return [
        {**candidate, "rerank_score": float(score)}
        for candidate, score in ranked[:top_n]
        if score >= min_score
    ]

The min_score threshold is useful: if nothing clears it, the honest answer is “I could not find this in the documentation”, which is far better than a confident answer built on irrelevant context. Tune the threshold on your eval set.

Generation with citations

Citations do two jobs. They let users verify answers, and they give you a cheap automated check: if the model cites a source that was not in the context, or makes a claim with no citation, something went wrong. Number the sources in the prompt, require the model to cite them inline and validate the citations before showing the answer.

import re
import anthropic

client = anthropic.Anthropic()

SYSTEM = """Answer using only the numbered sources provided.
Cite every factual sentence with its source number in square brackets, e.g. [2].
If the sources do not contain the answer, say you could not find it. Do not guess."""

def answer(question: str, sources: list[dict]) -> dict:
    context = "\n\n".join(
        f"<source id=\"{i + 1}\" title=\"{s['metadata'].get('title', '')}\">\n{s['text']}\n</source>"
        for i, s in enumerate(sources)
    )
    response = client.messages.create(
        model="claude-sonnet-5-5",
        max_tokens=1024,
        system=SYSTEM,
        messages=[{"role": "user", "content": f"{context}\n\nQuestion: {question}"}],
    )
    text = "".join(block.text for block in response.content if block.type == "text")

    cited = {int(n) for n in re.findall(r"\[(\d+)\]", text)}
    invalid = sorted(n for n in cited if n < 1 or n > len(sources))
    return {
        "text": text,
        "citations": [sources[n - 1]["metadata"] for n in sorted(cited) if 1 <= n <= len(sources)],
        "invalid_citations": invalid,
    }

In the UI, render citations as links to the exact section, not just the document. If invalid_citations is not empty, log it and either retry or fall back to a safe response.

Keep the index fresh

Stale content is a quiet failure mode. Re-index on change rather than on a fixed schedule when you can: listen to webhooks from your CMS, wiki or ticketing system and re-chunk only the affected documents. Store a content hash per document so unchanged files are skipped, and delete chunks for removed documents so the model never cites a page that no longer exists.

Key takeaways

• Measure recall@k on real questions before tuning prompts.
• Chunk along document structure and prepend the heading path to each chunk.
• Combine vector and keyword search; reciprocal rank fusion is a simple, robust merge.
• Rerank a shortlist and use a score threshold so the system can say “not found”.
• Require numbered citations and validate them before showing the answer.
• Enforce access control inside retrieval and keep the index in sync with the source.

Building search and question answering over your company's knowledge is a core part of our Agentic AI work. If your current RAG prototype is not holding up, get in touch.