Corrective RAG (CRAG)

Self-RAG asks a broad question at multiple points in the pipeline: "was this good?" CRAG asks one narrow, focused question — "is what I retrieved actually good enough to answer from?" — and, critically, pairs the answer with a specific corrective action rather than just a confidence label. A simple similarity threshold can tell you retrieval was weak; it can't tell you what to do about it. CRAG can.

Hands-on

The full pipeline below is runnable, against real data: Corrective RAG → in the code repo. Needs an LLM key, and a free Tavily key for the web-search fallback.

TL;DR — quick revision

• One batched evaluator call scores every retrieved chunk 1-5 for relevance, and the maximum score across the batch decides a 3-way verdict: Correct, Incorrect, or Ambiguous
• Each verdict triggers a genuinely different action, not just a different label: Correct refines the retrieved text down to the sentences that actually hold up; Incorrect discards it entirely and falls back to a real web search; Ambiguous does both and combines them
• Verified against real data, all three verdicts, including a live Tavily web search that returned actual current stock price data the sample paper obviously doesn't contain — and the model correctly cited it as an external source
• Realistic cost: 2-4 small/cheap-model calls plus, on the Incorrect/Ambiguous paths, one real network call to a search API — and exactly one expensive generation call, regardless of which branch runs

Following one real question through the pipeline

To make the mechanism concrete, three genuinely different questions — one for each possible verdict — are followed through the same corpus (the "Attention Is All You Need" paper) below, with the actual scores and output each one produced.

1. Retrieve, then score the whole batch at once

Naive retrieval pulls the usual top-k by similarity. CRAG's first real addition is what happens next: instead of trusting that top-k blindly, or checking relevance one document at a time, one LLM call scores all of them together — the same "batch it, don't loop it" principle used for listwise reranking.

import re

import llm

EVALUATOR_PROMPT = """Score how well each passage below helps answer the \
question, on a scale of 1 (irrelevant) to 5 (directly and fully answers \
it). Score EVERY passage, even if most seem irrelevant.

Question: {question}

Passages:
{numbered_passages}

Respond with ONLY the scores, one per passage in order, comma-separated, \
wrapped in <scores></scores> tags. Example for 3 passages: <scores>2,5,1</scores>"""

UPPER_THRESHOLD = 4  # max score >= this -> Correct
LOWER_THRESHOLD = 2  # max score <= this -> Incorrect


def evaluate_retrieval(question: str, chunks: list[str], provider: str | None = None) -> list[int]:
    """One LLM call scores the WHOLE candidate set at once — same
    "batch it, don't loop it" principle as listwise reranking. Falls
    back to a score of 1 for any chunk the model doesn't return a
    parseable score for, so a malformed response degrades toward
    "treat as irrelevant" rather than crashing or silently misaligning
    scores to the wrong chunks."""
    numbered = "\n".join(f"[{i + 1}] {c}" for i, c in enumerate(chunks))
    response = llm.generate(EVALUATOR_PROMPT.format(question=question, numbered_passages=numbered), provider=provider)
    match = re.search(r"<scores>(.*?)</scores>", response, re.S)

    scores = [1] * len(chunks)
    if match:
        parts = match.group(1).replace(" ", "").split(",")
        for i, part in enumerate(parts[: len(chunks)]):
            if part.isdigit():
                scores[i] = max(1, min(5, int(part)))
    return scores


def classify(scores: list[int]) -> str:
    top = max(scores) if scores else 1
    if top >= UPPER_THRESHOLD:
        return "Correct"
    if top <= LOWER_THRESHOLD:
        return "Incorrect"
    return "Ambiguous"

Real scores, three different questions, same retrieval step:

"What optimizer did they use for training?"
  -> scores [1, 5, 2, 1, 1] -> max=5 -> Correct

"What is the current stock price of Nvidia?"
  -> scores [1, 1, 1, 1, 1] -> max=1 -> Incorrect

"What is the significance of this paper compared to more recent
 transformer variants?"
  -> scores [2, 1, 1, 3, 1] -> max=3 -> Ambiguous

The thresholds are simple — max >= 4 is Correct, max <= 2 is Incorrect, anything in between is Ambiguous — but the outcome of that classification genuinely changes what the pipeline does, which is the whole point of doing this at all instead of picking one fallback behavior for every weak retrieval.

2a. Correct → refine, don't just trust the whole chunk

A chunk can score well overall and still contain sentences that don't actually help — the retrieved 500-character chunk on the Adam optimizer question, for instance, also has a sentence about the learning-rate schedule that's adjacent but not what was asked. Refinement splits every chunk into sentence-level "knowledge strips," re-scores each one individually, and keeps only the strips that clear the bar on their own.

import re

import llm

STRIP_RELEVANCE_PROMPT = """Score how well each short passage below \
helps answer the question, on a scale of 1 (irrelevant) to 5 (directly \
answers it).

Question: {question}

Passages:
{numbered_strips}

Respond with ONLY the scores, comma-separated, wrapped in <scores></scores> \
tags. Example for 3 passages: <scores>2,5,1</scores>"""


def refine_knowledge(question: str, chunks: list[str], provider: str | None = None, keep_threshold: int = 3) -> list[str]:
    """Even a chunk that scored well overall can have sentences that
    don't actually help — split into strips (sentences), re-score each
    one at this finer grain, and keep only the strips that individually
    clear the bar. Same idea as contextual compression, but gated
    specifically on the evaluator's "Correct" verdict rather than
    applied unconditionally to every retrieval."""
    strips = [s.strip() for c in chunks for s in re.split(r"(?<=[.!?])\s+", c) if len(s.strip()) > 20]
    if not strips:
        return chunks

    numbered = "\n".join(f"[{i + 1}] {s}" for i, s in enumerate(strips))
    response = llm.generate(STRIP_RELEVANCE_PROMPT.format(question=question, numbered_strips=numbered), provider=provider)
    match = re.search(r"<scores>(.*?)</scores>", response, re.S)

    if not match:
        return chunks

    parts = match.group(1).replace(" ", "").split(",")
    kept = [strips[i] for i, part in enumerate(parts[: len(strips)]) if part.isdigit() and int(part) >= keep_threshold]
    return kept if kept else chunks

For the optimizer question, this is the path that ran, and the final answer came back fully grounded in the refined internal text: "they used the Adam optimizer for training, with β1 = 0.9, β2 = 0.98, ϵ = 10^-9... They also employed a custom learning rate schedule..." — no web search needed, because the internal corpus genuinely had the answer.

2b. Incorrect → discard everything, search the web instead

When the evaluator can't find anything worth keeping, refining bad chunks further wouldn't help — there's nothing there to refine. The corrective action is to stop trusting internal retrieval for this query entirely, rewrite the question into a short, keyword-style query the way a person would actually type it into a search engine, and fetch real external results.

import llm

SEARCH_REWRITE_PROMPT = """Rewrite the question below into a short, \
keyword-style web search query — the way you'd type it into a search \
engine, not a full sentence. Return ONLY the search query.

Question: {question}"""


def rewrite_for_search(question: str, provider: str | None = None) -> str:
    return llm.generate(SEARCH_REWRITE_PROMPT.format(question=question), provider=provider).strip()


def web_search(query: str, max_results: int = 3) -> list[str]:
    import os

    from tavily import TavilyClient

    api_key = os.environ.get("TAVILY_API_KEY")
    if not api_key:
        raise RuntimeError("TAVILY_API_KEY is not set. Get a free key at https://tavily.com and add it to .env.")

    client = TavilyClient(api_key=api_key)
    response = client.search(query, max_results=max_results)
    return [f"{r['title']}: {r['content']}" for r in response.get("results", [])]

Real, live trace for "What is the current stock price of Nvidia?" — a question the 2017 Transformer paper obviously has no way to answer:

Rewritten search query: "Nvidia stock price current"

Real Tavily result (excerpt): "NVIDIA(NVDA) stock is priced at $202.55...
As of 2026-07-18, NVIDIA(NVDA) stock has fluctuated between $197.97 and
$206.65... 52-week high of $236.54 and a 52-week low of $164.07."

Final answer: "Based on the context provided, there are multiple stock
prices listed for NVIDIA (NVDA) from different dates: $230.36 (as of
September 4)... $202.55 (as of July 18, 2026)... Note: The context
includes conflicting data points and dates... which suggests the data
may be from different sources or time periods."

This is a real network call against the live Tavily API, not a mock — the numbers above are genuine current search results, not fabricated. Worth noticing: the model correctly flagged the conflicting dates/prices across multiple search results as a real ambiguity in the source data, rather than silently picking one number and presenting it as certain.

2c. Ambiguous → do both, and let generation reconcile them

Sometimes the internal corpus has something relevant but not a confident, complete answer — refining it alone would leave gaps; discarding it and only searching the web would throw away a real partial answer. The Ambiguous branch keeps both: refined internal strips and web search results, concatenated before generation.

Real trace for "What is the significance of this paper compared to more recent transformer variants?" — the internal corpus has strong material on what the paper itself established, but nothing about how later variants compare to it:

Source used: internal (refined) + web
Search query: "significance paper vs recent transformer variants"

Final answer (excerpt): "The 2017 Transformer paper established a
foundational architecture that remains highly influential... Recent
variants focus on addressing the quadratic complexity bottleneck of
the original attention mechanism: Sub-quadratic attention variants,
Linear attention mechanisms... While these newer approaches improve
computational efficiency... they are incremental improvements built
upon the core Transformer architecture rather than revolutionary
replacements."

The internal, refined strips supplied the paper's own claimed significance; the web search supplied what came after it — exactly the combination this branch exists for.

3. Generate — same call, regardless of branch

import os
import re

import numpy as np
from sentence_transformers import SentenceTransformer
from tavily import TavilyClient

import llm

RETRIEVE_K = 5


def _naive_retrieve(question, chunks, embed_model, k):
    query_vec = embed_model.encode(question, convert_to_numpy=True)
    chunk_vecs = embed_model.encode(chunks, convert_to_numpy=True, show_progress_bar=False)
    sims = [float(np.dot(query_vec, v) / (np.linalg.norm(query_vec) * np.linalg.norm(v) + 1e-8)) for v in chunk_vecs]
    top_indices = np.argsort(sims)[::-1][:k]
    return [chunks[i] for i in top_indices]


def _score_passages(prompt_prefix, question, passages, provider):
    numbered = "\n".join(f"[{i + 1}] {p}" for i, p in enumerate(passages))
    response = llm.generate(
        f"{prompt_prefix}\n\nQuestion: {question}\n\nPassages:\n{numbered}\n\n"
        f"Respond with ONLY the scores, comma-separated, wrapped in <scores></scores> tags.",
        provider=provider,
    )
    match = re.search(r"<scores>(.*?)</scores>", response, re.S)
    scores = [1] * len(passages)
    if match:
        parts = match.group(1).replace(" ", "").split(",")
        for i, part in enumerate(parts[: len(passages)]):
            if part.isdigit():
                scores[i] = max(1, min(5, int(part)))
    return scores


def _refine_knowledge(question, chunks, provider):
    """The real strip-level refinement (see refine_knowledge above for
    the standalone version with full docs) — split into sentences,
    re-score each one, keep only strips that individually clear the bar."""
    strips = [s.strip() for c in chunks for s in re.split(r"(?<=[.!?])\s+", c) if len(s.strip()) > 20]
    if not strips:
        return chunks
    scores = _score_passages(
        "Score how well each short passage below helps answer the question, "
        "on a scale of 1 (irrelevant) to 5 (directly answers it).",
        question, strips, provider,
    )
    kept = [s for s, sc in zip(strips, scores) if sc >= 3]
    return kept if kept else chunks


def _web_search(question, provider):
    """The real web-search fallback (see rewrite_for_search and
    web_search above for the standalone versions) — rewrite to
    keywords, then a real Tavily API call."""
    search_query = llm.generate(
        f"Rewrite the question below into a short, keyword-style web search query "
        f"— the way you'd type it into a search engine, not a full sentence. "
        f"Return ONLY the search query.\n\nQuestion: {question}",
        provider=provider,
    ).strip()
    client = TavilyClient(api_key=os.environ["TAVILY_API_KEY"])
    results = client.search(search_query, max_results=3)
    return search_query, [f"{r['title']}: {r['content']}" for r in results.get("results", [])]


def corrective_rag(
    question: str,
    chunks: list[str],
    embed_model: SentenceTransformer,
    provider: str | None = None,
    k: int = RETRIEVE_K,
) -> dict:
    """Retrieve -> evaluate the whole candidate set in one call ->
    branch on the verdict -> generate. Each branch is a genuinely
    different corrective ACTION, not just a different confidence
    label."""
    retrieved = _naive_retrieve(question, chunks, embed_model, k)
    scores = _score_passages(
        "Score how well each passage below helps answer the question, on a scale "
        "of 1 (irrelevant) to 5 (directly and fully answers it). Score EVERY passage, "
        "even if most seem irrelevant.",
        question, retrieved, provider,
    )
    top = max(scores) if scores else 1
    verdict = "Correct" if top >= 4 else "Incorrect" if top <= 2 else "Ambiguous"

    trace = {"retrieved": len(retrieved), "scores": scores, "verdict": verdict}

    if verdict == "Correct":
        context = _refine_knowledge(question, retrieved, provider)
        trace["source"] = "internal (refined)"

    elif verdict == "Incorrect":
        search_query, web_results = _web_search(question, provider)
        context = web_results
        trace["source"] = "web"
        trace["search_query"] = search_query

    else:  # Ambiguous
        refined = _refine_knowledge(question, retrieved, provider)
        search_query, web_results = _web_search(question, provider)
        context = refined + web_results
        trace["source"] = "internal (refined) + web"
        trace["search_query"] = search_query

    trace["context"] = context
    context_text = "\n---\n".join(context) if context else "(no context available)"
    trace["answer"] = llm.generate(
        f"Answer the question using the context below. If context from the web is "
        f"included, you may cite it as such.\n\nContext:\n{context_text}\n\nQuestion: {question}",
        provider=provider,
    )
    return trace

Whichever branch ran, generation itself is the same single call over whatever context that branch produced — internal strips, web results, or both. The branching logic is entirely upstream of generation, which is what keeps this at "one expensive call regardless of verdict," not three different generation paths to maintain.

Mechanics: call counts and batching

Is evaluation per-document or batched? Batched, not sequential — one LLM call scores every retrieved chunk together (or, in the original paper, one forward pass through a small fine-tuned classifier). Never k separate round-trips for k documents.

Realistic cost per query:

• Correct: 1 evaluator call + 1 refinement call + 1 generation call — ~2 small-model calls + 1 big-model call
• Incorrect: 1 evaluator call + 1 query-rewrite call + 1 web search API call (network request, not an LLM call) + 1 generation call — ~1 small-model call + 1 API call + 1 big-model call
• Ambiguous: both branches combined — ~3 small-model calls + 1 API call + 1 big-model call

Never more than one expensive generation call, and never one LLM call per retrieved document.

Use it for / skip it for

Use it when both are true: your knowledge base has real coverage gaps (a fast-moving domain, a still-being-built corpus, or queries legitimately outside what's indexed), and a wrong or unsupported answer is more costly than the extra latency of a fallback. Customer support over a KB that updates slower than the product, research tools where the internal corpus is necessarily incomplete, regulated domains where an ungrounded answer has real consequences — all good fits.

Skip it when: your KB is narrow and comprehensive enough that retrieval failure is rare (you're paying evaluator overhead for a problem that almost never occurs), latency is tight enough that added round-trips aren't affordable, or there's no sensible external fallback for the domain at all (querying private financial records, say — web search doesn't help there, and CRAG degrades into just an expensive relevance filter with nothing to fall back to).

Relation to Self-RAG

Self-RAG is the broader framework — the generation model reflecting on retrieval relevance, its own output's groundedness, and usefulness, at multiple points. CRAG is narrower and more mechanical: one evaluator, one focused verdict, three concrete corrective actions. The two aren't mutually exclusive — a system could run CRAG's evaluator-and-fallback logic and Self-RAG's groundedness/usefulness checks on whatever CRAG hands to generation.

Scenario Check

Sources & further reading

• Corrective Retrieval Augmented Generation — the original CRAG paper
• Tavily — search API built for LLM/agent use, used for the web-search fallback demoed above
• Self-RAG — the broader self-reflection framework CRAG's evaluator complements