Overview
Corrective RAG (CRAG) adds a self-correction layer that evaluates retrieved documents and takes action when they're irrelevant or insufficient. It can refine searches, add web results, or regenerate entirely.
Pattern #13 in the 37 RAG Patterns Collection
The Problem
Standard RAG blindly trusts retrieval:
Query: "Latest iPhone features"
Retrieved: [2020 iPhone article] ❌ Outdated
LLM: Generates answer from outdated source
CRAG detects this and corrects:
Query: "Latest iPhone features"
Retrieved: [2020 article]
↓
Relevance Grader: ❌ "Outdated, not current"
↓
Action: Add web search for current information
↓
New context: [2026 iPhone features]
↓
LLM: Generates accurate, current answer ✅
Architecture
graph TB
A[Query] --> B[Initial Retrieval]
B --> C{Relevance Grader}
C -->|Relevant| D[Use Retrieved Docs]
C -->|Ambiguous| E[Knowledge Refinement]
C -->|Irrelevant| F[Web Search]
E --> G[Decompose + Filter]
G --> H[Refined Context]
F --> I[Web Results]
D --> J[Generate Answer]
H --> J
I --> J
style A fill:#f9f,stroke:#333
style C fill:#ff9,stroke:#333
style J fill:#9f9,stroke:#333Core Components
1. Relevance Grader
python
def grade_relevance(query, documents):
"""Grade each document's relevance"""
grades = []
for doc in documents:
prompt = f"""Query: {query}
Document: {doc['content'][:500]}
Is this document relevant and sufficient to answer the query?
Grade: [RELEVANT | AMBIGUOUS | IRRELEVANT]
Reasoning: [brief explanation]
Return JSON."""
grade = llm.invoke(prompt, schema=GRADE_SCHEMA)
grades.append({
'doc': doc,
'grade': grade['grade'],
'reasoning': grade['reasoning']
})
return grades2. Corrective Actions
python
def corrective_action(grades, query):
"""Decide what action to take based on grades"""
relevant = [g for g in grades if g['grade'] == 'RELEVANT']
ambiguous = [g for g in grades if g['grade'] == 'AMBIGUOUS']
irrelevant = [g for g in grades if g['grade'] == 'IRRELEVANT']
# All relevant → proceed
if len(relevant) >= 3:
return "USE_RETRIEVED", [g['doc'] for g in relevant]
# Some ambiguous → refine
elif ambiguous:
return "REFINE_KNOWLEDGE", ambiguous
# All irrelevant → search web
else:
return "WEB_SEARCH", query3. Knowledge Refinement
python
def refine_knowledge(ambiguous_docs, query):
"""Extract relevant portions from ambiguous documents"""
refined = []
for doc_grade in ambiguous_docs:
prompt = f"""Query: {query}
Document: {doc_grade['doc']['content']}
Extract only the portions relevant to answering the query.
Decompose into atomic facts."""
relevant_parts = llm.invoke(prompt, schema=FACTS_SCHEMA)
refined.extend(relevant_parts['facts'])
return refined4. Web Search Fallback
python
def web_search_fallback(query):
"""Search web when retrieval fails"""
# Use web search API
web_results = search_api.search(query, num_results=3)
# Extract and grade web content
web_docs = [extract_content(url) for url in web_results]
grades = grade_relevance(query, web_docs)
return [g['doc'] for g in grades if g['grade'] == 'RELEVANT']Complete Implementation
python
def corrective_rag(query):
# Step 1: Initial retrieval
documents = retriever.search(query, top_k=5)
# Step 2: Grade relevance
grades = grade_relevance(query, documents)
# Step 3: Decide action
action, data = corrective_action(grades, query)
# Step 4: Execute correction
if action == "USE_RETRIEVED":
context = data
elif action == "REFINE_KNOWLEDGE":
context = refine_knowledge(data, query)
elif action == "WEB_SEARCH":
# Combine internal + web
relevant_internal = [
g['doc'] for g in grades if g['grade'] == 'RELEVANT'
]
web_results = web_search_fallback(query)
context = relevant_internal + web_results
# Step 5: Generate with corrected context
response = llm.generate(
f"Context: {context}\n\nQuestion: {query}"
)
return response, {
'action': action,
'grades': grades,
'context_sources': [doc['source'] for doc in context]
}Complete Notebook
🚀 Run the full implementation:
Includes:
- ▸Complete CRAG pipeline
- ▸Relevance grading prompts
- ▸Knowledge refinement logic
- ▸Web search integration
- ▸Benchmark comparisons
Performance
| Metric | CRAG | Standard RAG |
|---|---|---|
| Accuracy | 88% | 72% |
| Hallucinations | 5% | 15% |
| Latency | ~800ms | ~300ms |
| Cost/query | $0.15 | $0.08 |
When to Use
Best For
- ▸High-stakes applications (medical, legal, financial)
- ▸Dynamic domains where information changes
- ▸Quality-critical systems where errors are costly
- ▸Mixed sources (internal docs + web)
Combine With
- ▸Self-RAG (#14) for multi-dimensional quality checks
- ▸Caching (#33) to cache grading decisions
- ▸Adaptive RAG (#8) to learn from corrections
Grading Strategies
Binary Grading (Simple)
python
grades = ["RELEVANT", "IRRELEVANT"]Three-Level Grading (Standard)
python
grades = ["RELEVANT", "AMBIGUOUS", "IRRELEVANT"]Confidence Scoring (Advanced)
python
{
"grade": "RELEVANT",
"confidence": 0.85,
"reasoning": "..."
}Optimization Tips
1. Parallel Grading
python
import concurrent.futures
def parallel_grade(query, documents):
with concurrent.futures.ThreadPoolExecutor() as executor:
futures = [
executor.submit(grade_single, query, doc)
for doc in documents
]
return [f.result() for f in futures]2. Cached Grades
python
grade_cache = {}
def grade_with_cache(query, doc):
key = f"{hash(query)}_{hash(doc['id'])}"
if key not in grade_cache:
grade_cache[key] = grade_relevance(query, [doc])[0]
return grade_cache[key]3. Adaptive Thresholds
python
def adaptive_threshold(query_complexity):
"""Adjust relevance threshold based on query"""
if query_complexity == "simple":
return 2 # Need 2 relevant docs
elif query_complexity == "complex":
return 4 # Need 4 relevant docs
return 3 # defaultRelated Papers
- ▸CRAG: Corrective Retrieval Augmented Generation (2024)
- ▸Self-RAG: Learning to Retrieve, Generate and Critique (2023)
Next Steps
- ▸Try it: Run the CRAG notebook
- ▸Upgrade: Combine with Self-RAG (#14) for full quality control
- ▸Deploy: Use Production RAG (#35) patterns
- ▸Monitor: Track correction rates to improve retrieval
Part of: 37 RAG Patterns Collection
See also: Complete Pattern Index