Corrective RAG with Hallucination Detection: Automatically Validating and Replacing Unreliable Retrieved Documents at Runtime

Introduction

You've built a RAG pipeline. The retriever pulls documents, the LLM synthesizes an answer, and most of the time it works beautifully. Then a user asks something slightly outside your corpus's comfort zone, and the retriever hands back a document that's tangentially related at best — a paragraph about quarterly revenue trends when the question was about refund policy. The LLM, ever the obliging assistant, takes that weak context and confidently fabricates an answer. Nobody flags it. It goes into a support ticket response, a financial summary, or a medical FAQ. That's the RAG failure mode that keeps architects up at night.

Standard RAG is retrieve-then-generate: you pull some documents and trust them. Corrective RAG (CRAG) is retrieve-then-evaluate-then-decide: you pull documents, score their reliability, and then either proceed, augment with better sources, or fall back to a different strategy entirely — all before the LLM generates a single token of the final answer. The correction happens at runtime, not in an offline evaluation loop. This turns your pipeline from a passive document conduit into an active reasoning system.

By the end of this lesson, you'll have implemented a production-ready Corrective RAG pipeline with hallucination detection at both the retrieval validation and generation validation stages. You'll understand the architectural decisions behind each component and be able to tune the system for your specific quality/latency trade-off.

What you'll learn:

• How to build a retrieval confidence scorer that goes beyond embedding similarity cosine scores
• How to implement document-level relevance grading using an LLM-as-judge pattern with calibrated prompts
• How to architect a dynamic fallback strategy that switches between local corpus, web search, and knowledge synthesis
• How to detect hallucinations in generated outputs by cross-referencing claims against retrieved sources
• How to integrate all of this into a coherent LangGraph state machine that orchestrates the correction loop at runtime

---

Prerequisites

This lesson assumes you're comfortable with the following:

• Python at an intermediate-to-advanced level — you should be able to read and write async Python and understand class-based design
• Core RAG concepts — you've built a basic retrieval-augmented generation pipeline before (vector store, embeddings, chain)
• LangChain fundamentals — familiarity with ChatOpenAI, VectorStore, and basic chain composition
• Basic LangGraph knowledge — you've seen state graphs before, even if you haven't built complex ones
• Working API keys for OpenAI (or another provider) and optionally Tavily for web search

You do not need to have read the original CRAG paper (Shi et al., 2024), though it's worth reading after this lesson to compare the paper's architecture with the pragmatic implementation choices we make here.

---

Understanding Why Standard RAG Fails — and Where Correction Must Happen

Before writing any code, you need a precise mental model of the failure modes. There are two distinct places where quality degrades in a RAG pipeline, and they require different correction strategies.

Retrieval failure happens when the documents pulled from your vector store don't actually contain the information needed to answer the question. The cosine similarity between the query embedding and document embeddings might look acceptable — 0.72, 0.68, 0.71 — but those scores measure semantic proximity, not informational sufficiency. A question about "how to appeal a denied insurance claim" might return documents about "insurance claim processing timelines" and "appeals court procedures" because those phrases live in the same semantic neighborhood. Neither document answers the question.

Generation failure happens when the retrieved documents are fine but the LLM hallucinates details not present in them. This is subtler. The LLM might correctly use the documents as a foundation and then extrapolate beyond them — inventing a specific dollar threshold, citing a policy number that doesn't exist, or stating a deadline that was never mentioned. Classic hallucination.

The architecture insight is this: these two failure modes require intervention at different points in the pipeline, and you can't fix generation failure by fixing retrieval alone, or vice versa. A complete Corrective RAG system has two validation checkpoints:

[Query]
   │
   ▼
[Retrieve Documents]
   │
   ▼
[Retrieval Validator] ◄─── Checkpoint 1: Are the docs trustworthy?
   │
   ├── HIGH CONFIDENCE ──► [Generate Answer]
   │                              │
   ├── LOW CONFIDENCE ──► [Fetch Better Sources]   [Generation Validator] ◄── Checkpoint 2: Does the answer
   │                              │                        │                    match the sources?
   └── AMBIGUOUS ──────► [Augment + Generate]      ├── CONSISTENT ──► [Return Answer]
                                                    └── INCONSISTENT ──► [Regenerate / Flag]

Let's build this system layer by layer.

---

Setting Up the Project and Dependencies

First, let's establish a clean project structure:

crag_system/
├── __init__.py
├── models.py          # State definitions and data models
├── retrievers.py      # Vector store + web search retrieval
├── validators.py      # Relevance grading and hallucination detection
├── graph.py           # LangGraph state machine
└── main.py            # Entry point

Install the required packages:

pip install langchain langchain-openai langchain-community langgraph \
            tavily-python chromadb tiktoken pydantic

# models.py
from typing import List, Optional, Literal
from pydantic import BaseModel, Field
from langchain_core.documents import Document


class RelevanceScore(BaseModel):
    """Structured output from the relevance grader."""
    score: Literal["yes", "no", "partial"] = Field(
        description="Whether the document is relevant to the question"
    )
    reasoning: str = Field(
        description="Brief explanation of the relevance judgment"
    )
    confidence: float = Field(
        ge=0.0, le=1.0,
        description="Confidence in the relevance judgment (0-1)"
    )


class HallucinationAssessment(BaseModel):
    """Structured output from the hallucination detector."""
    is_grounded: bool = Field(
        description="True if all claims in the answer are supported by the documents"
    )
    unsupported_claims: List[str] = Field(
        default_factory=list,
        description="List of specific claims not found in the retrieved documents"
    )
    confidence: float = Field(
        ge=0.0, le=1.0,
        description="Confidence in the grounding assessment"
    )


class GraphState(BaseModel):
    """The shared state that flows through the LangGraph nodes."""
    question: str
    documents: List[Document] = Field(default_factory=list)
    web_results: List[Document] = Field(default_factory=list)
    generation: Optional[str] = None
    relevance_scores: List[RelevanceScore] = Field(default_factory=list)
    hallucination_assessment: Optional[HallucinationAssessment] = None
    retrieval_decision: Optional[Literal["proceed", "fallback", "augment"]] = None
    generation_attempts: int = 0
    final_answer: Optional[str] = None
    error: Optional[str] = None

The GraphState is the backbone of the entire system. Every node in our graph reads from and writes to this state object. Notice that we track generation_attempts — you'll need this to prevent infinite correction loops, which is a real production concern.

---

Building the Retrieval Confidence Scorer

Here's where most CRAG tutorials cut corners. They use raw cosine similarity as the relevance signal. The problem: cosine similarity is a distance metric in embedding space, not an answer quality metric. A document about "employee benefit enrollment windows" and a question about "when can I change my health insurance" might score 0.81 similarity even if the document is from 2019 and describes a policy that no longer exists.

We're going to build a two-stage relevance scorer: a fast embedding-based pre-filter, followed by an LLM-based deep grader for borderline cases.

# validators.py
import json
from typing import List, Tuple
from langchain_openai import ChatOpenAI
from langchain_core.documents import Document
from langchain_core.prompts import ChatPromptTemplate
from langchain_core.output_parsers import PydanticOutputParser
from models import RelevanceScore, HallucinationAssessment


RELEVANCE_GRADER_PROMPT = """You are evaluating whether a retrieved document contains 
information that would help answer a specific question. Your job is to be precise and 
critical — not optimistic. A document that mentions related topics but doesn't contain 
the specific information needed should be scored "no" or "partial", not "yes".

Question: {question}

Retrieved Document:
Title: {doc_title}
Content: {doc_content}

Evaluate the document on these criteria:
1. Does it directly address what the question is asking?
2. Does it contain specific, usable information (not just tangential references)?
3. Is the information current enough to be reliable (if temporal context is relevant)?

{format_instructions}

Be strict. A "yes" means this document substantially helps answer the question. 
A "partial" means it has some useful content but is incomplete or only marginally relevant.
A "no" means this document would not help generate an accurate answer."""


HALLUCINATION_DETECTOR_PROMPT = """You are a fact-checking system. Your task is to 
determine whether a generated answer is fully supported by the provided source documents.

Question that was asked: {question}

Source Documents Used:
{documents}

Generated Answer:
{answer}

Instructions:
1. Identify every factual claim in the generated answer
2. Check each claim against the source documents
3. A claim is "unsupported" if:
   - It states a specific fact (number, date, name, policy detail) not present in the sources
   - It draws a conclusion that goes beyond what the sources state
   - It contradicts something in the sources

Do NOT mark as unsupported:
   - Reasonable logical inferences explicitly supported by the text
   - Paraphrases that accurately represent source content
   - Common knowledge statements used as framing

{format_instructions}"""


class RelevanceGrader:
    """
    Two-stage relevance grader.
    
    Stage 1: Fast embedding similarity pre-filter
    Stage 2: LLM-based deep grading for borderline documents
    """
    
    def __init__(
        self,
        llm: ChatOpenAI,
        fast_threshold: float = 0.85,
        reject_threshold: float = 0.40,
    ):
        self.llm = llm
        self.fast_threshold = fast_threshold  # Above this: auto-accept
        self.reject_threshold = reject_threshold  # Below this: auto-reject
        
        self.parser = PydanticOutputParser(pydantic_object=RelevanceScore)
        self.prompt = ChatPromptTemplate.from_template(RELEVANCE_GRADER_PROMPT)
        self.chain = self.prompt | self.llm.with_structured_output(RelevanceScore)
    
    def grade_with_similarity(
        self, 
        doc: Document, 
        similarity_score: float,
        question: str
    ) -> Tuple[RelevanceScore, bool]:
        """
        Grade a document, using similarity as a fast-path decision when possible.
        Returns (RelevanceScore, used_llm_grader).
        """
        # Fast path: very high similarity — do LLM grading anyway for these,
        # because high similarity doesn't guarantee answer quality
        # Fast reject: very low similarity — skip the expensive LLM call
        if similarity_score < self.reject_threshold:
            return RelevanceScore(
                score="no",
                reasoning=f"Embedding similarity {similarity_score:.3f} below rejection threshold",
                confidence=0.90
            ), False
        
        # For everything above rejection threshold, use LLM grading
        # This is more expensive but dramatically more accurate
        return self._llm_grade(doc, question), True
    
    def _llm_grade(self, doc: Document, question: str) -> RelevanceScore:
        """Full LLM-based relevance grading."""
        title = doc.metadata.get("title", doc.metadata.get("source", "Unknown"))
        # Truncate content to avoid token explosion — use first 1500 chars
        content = doc.page_content[:1500]
        
        try:
            result = self.chain.invoke({
                "question": question,
                "doc_title": title,
                "doc_content": content,
                "format_instructions": self.parser.get_format_instructions()
            })
            return result
        except Exception as e:
            # Fail safe: if grading fails, treat as partial
            return RelevanceScore(
                score="partial",
                reasoning=f"Grading failed: {str(e)}",
                confidence=0.3
            )
    
    def grade_documents(
        self,
        documents: List[Document],
        similarity_scores: List[float],
        question: str
    ) -> List[Tuple[Document, RelevanceScore]]:
        """Grade all retrieved documents and return paired (doc, score) tuples."""
        results = []
        for doc, sim_score in zip(documents, similarity_scores):
            relevance, _ = self.grade_with_similarity(doc, sim_score, question)
            results.append((doc, relevance))
        return results

Design decision: Notice we don't auto-accept high-similarity documents. This is intentional and differs from the naive approach. High embedding similarity means the document talks about similar things, but a question about "what's the deadline to submit a FMLA claim" might have 0.88 similarity with a document about "FMLA eligibility requirements" — same topic, completely different informational content. The LLM grader catches this distinction.

---

Making the Retrieval Decision

After grading all retrieved documents, you need to aggregate the scores into a single routing decision. This is where architectural judgment matters.

# validators.py (continued)

from models import GraphState
from typing import Literal


def compute_retrieval_decision(
    scored_docs: List[Tuple[Document, RelevanceScore]],
    high_quality_threshold: int = 2,
    partial_threshold: int = 1
) -> Tuple[Literal["proceed", "fallback", "augment"], List[Document]]:
    """
    Given a list of (document, relevance_score) pairs, decide:
    
    - "proceed": enough high-quality documents exist, use them
    - "fallback": documents are too poor, go to web search entirely
    - "augment": some useful documents, but supplement with web search
    
    Also returns the filtered list of documents worth keeping.
    """
    yes_docs = [(doc, score) for doc, score in scored_docs if score.score == "yes"]
    partial_docs = [(doc, score) for doc, score in scored_docs if score.score == "partial"]
    
    # Weight partials: require more of them to compensate
    effective_quality = len(yes_docs) + (len(partial_docs) * 0.5)
    
    if len(yes_docs) >= high_quality_threshold:
        # We have enough high-quality docs — proceed with confidence
        useful_docs = [doc for doc, _ in yes_docs]
        return "proceed", useful_docs
    
    elif effective_quality >= partial_threshold:
        # We have some signal but it's weak — augment with web search
        useful_docs = [doc for doc, _ in yes_docs + partial_docs]
        return "augment", useful_docs
    
    else:
        # Nothing useful found locally — fall back entirely to web search
        return "fallback", []

The three-way routing decision is one of the key contributions of the CRAG paper, and it's worth thinking through the implications carefully:

• Proceed means you trust your corpus. This is the happy path.
• Augment means you're mixing local knowledge with fresh web results. This requires careful prompt design because you're blending sources of different quality and provenance.
• Fallback means your corpus failed completely. You're now doing real-time web RAG, which has completely different latency characteristics and trust assumptions.

Warning: The "fallback" path has significant security implications. You're now introducing arbitrary web content into your LLM context. If you're deploying this in an enterprise setting, "fallback" should route to a curated external source (your own documentation site, an approved knowledge base API) rather than an open web search. We'll implement both options below.

---

Implementing the Web Search Fallback

# retrievers.py
from typing import List
from langchain_community.tools.tavily_search import TavilySearchResults
from langchain_core.documents import Document
import os


class WebSearchRetriever:
    """
    Wraps Tavily web search and converts results to Documents
    that are compatible with our pipeline's document format.
    """
    
    def __init__(self, max_results: int = 4, search_depth: str = "advanced"):
        self.search_tool = TavilySearchResults(
            max_results=max_results,
            search_depth=search_depth,  # "advanced" gets full page content
            include_answer=True,         # Tavily's own summary as a bonus doc
            include_raw_content=False,   # Raw HTML is noise
        )
    
    def retrieve(self, query: str) -> List[Document]:
        """
        Execute web search and return results as Documents.
        Includes metadata about the web source for transparency.
        """
        try:
            results = self.search_tool.invoke({"query": query})
        except Exception as e:
            return []
        
        documents = []
        for result in results:
            if isinstance(result, dict):
                content = result.get("content", "")
                url = result.get("url", "")
                title = result.get("title", url)
                score = result.get("score", 0.0)
                
                doc = Document(
                    page_content=content,
                    metadata={
                        "source": url,
                        "title": title,
                        "relevance_score": score,
                        "source_type": "web_search",
                        "query": query
                    }
                )
                documents.append(doc)
        
        return documents
    
    def retrieve_with_query_expansion(self, question: str, llm) -> List[Document]:
        """
        For low-confidence fallback scenarios, expand the query before searching.
        This often recovers documents that a literal query would miss.
        """
        expansion_prompt = f"""Generate 2 alternative search queries for the following question.
Return only the queries, one per line, no numbering or explanation.

Original question: {question}

Alternative queries:"""
        
        response = llm.invoke(expansion_prompt)
        alternative_queries = response.content.strip().split('\n')
        
        all_docs = self.retrieve(question)  # Original query
        seen_urls = {doc.metadata.get("source", "") for doc in all_docs}
        
        for alt_query in alternative_queries[:2]:  # Limit to 2 expansions
            alt_query = alt_query.strip()
            if alt_query:
                alt_docs = self.retrieve(alt_query)
                for doc in alt_docs:
                    url = doc.metadata.get("source", "")
                    if url not in seen_urls:
                        all_docs.append(doc)
                        seen_urls.add(url)
        
        return all_docs

---

Building the Hallucination Detector

The hallucination detector is the second validation checkpoint. It runs after generation and checks whether the LLM's output is grounded in the actual documents used.

# validators.py (continued)

class HallucinationDetector:
    """
    Validates that a generated answer is factually grounded in source documents.
    
    This is NOT a general fact-checker — it only checks consistency between
    the answer and the provided documents, not ground truth accuracy.
    Important distinction: a document can be wrong, and an answer consistent
    with a wrong document will pass this check. The retrieval validator
    handles source quality; this handles generation faithfulness.
    """
    
    def __init__(self, llm: ChatOpenAI):
        self.llm = llm
        self.chain = (
            ChatPromptTemplate.from_template(HALLUCINATION_DETECTOR_PROMPT) 
            | self.llm.with_structured_output(HallucinationAssessment)
        )
    
    def assess(
        self, 
        question: str,
        answer: str, 
        documents: List[Document]
    ) -> HallucinationAssessment:
        """
        Check whether the answer is grounded in the provided documents.
        """
        # Format documents for the prompt
        doc_text = self._format_documents(documents)
        
        try:
            result = self.chain.invoke({
                "question": question,
                "documents": doc_text,
                "answer": answer,
                "format_instructions": ""  # Handled by with_structured_output
            })
            return result
        except Exception as e:
            # Conservative fail: if detection fails, assume hallucination risk
            return HallucinationAssessment(
                is_grounded=False,
                unsupported_claims=[f"Assessment failed: {str(e)}"],
                confidence=0.1
            )
    
    def _format_documents(self, documents: List[Document]) -> str:
        """Format documents for inclusion in the hallucination check prompt."""
        formatted = []
        for i, doc in enumerate(documents, 1):
            source = doc.metadata.get("source", f"Document {i}")
            title = doc.metadata.get("title", source)
            # Use full content here — we need accuracy, not speed
            formatted.append(f"[Document {i}: {title}]\n{doc.page_content}")
        return "\n\n---\n\n".join(formatted)

A crucial architectural note: The hallucination detector is checking faithfulness to sources, not factual accuracy. These are different things. If your retrieved document says "the policy was updated in 2021" but the real update was 2022, an answer that says "per policy updated in 2021" is faithful (passes the hallucination check) but factually wrong. The only way to catch this is to have accurate source documents — which is why retrieval quality validation is the more important checkpoint of the two.

---

Building the Answer Generator

The generator itself needs some design care. When we're mixing local documents with web results (augment mode), the prompt needs to handle source provenance explicitly.

# From graph.py — the generation node logic

from langchain_openai import ChatOpenAI
from langchain_core.prompts import ChatPromptTemplate

GENERATION_PROMPT_STANDARD = """You are a knowledgeable assistant. Answer the question 
using ONLY the information provided in the source documents below. 

If the documents don't contain enough information to fully answer the question, 
say so explicitly — do not invent details or fill gaps with general knowledge.

Source Documents:
{documents}

Question: {question}

Answer:"""

GENERATION_PROMPT_MIXED_SOURCES = """You are a knowledgeable assistant. Answer the 
question using the provided source documents. These documents come from two sources:
1. Internal knowledge base documents (marked [INTERNAL])  
2. Web search results (marked [WEB])

Prioritize internal documents when they conflict with web results, as they represent
authoritative organizational knowledge. Use web results to supplement gaps.

If you cannot find sufficient information, state that explicitly.

Source Documents:
{documents}

Question: {question}

Answer:"""


def format_docs_for_generation(
    documents: List[Document], 
    include_source_type: bool = False
) -> str:
    """Format documents for the generation prompt."""
    parts = []
    for i, doc in enumerate(documents, 1):
        source = doc.metadata.get("source", f"Doc {i}")
        title = doc.metadata.get("title", source)
        source_type = doc.metadata.get("source_type", "internal")
        
        if include_source_type:
            tag = "[WEB]" if source_type == "web_search" else "[INTERNAL]"
            header = f"{tag} {title}"
        else:
            header = title
            
        parts.append(f"[{i}] {header}\n{doc.page_content}")
    
    return "\n\n".join(parts)

---

Assembling the LangGraph State Machine

Now comes the integration. LangGraph is ideal for this because the correction loop has genuine conditional branching, and managing that with plain Python callbacks quickly becomes unmaintainable.

# graph.py
from typing import Any, Dict
from langgraph.graph import StateGraph, END
from langchain_openai import ChatOpenAI
from langchain_community.vectorstores import Chroma
from langchain_openai import OpenAIEmbeddings

from models import GraphState, RelevanceScore
from validators import RelevanceGrader, HallucinationDetector, compute_retrieval_decision
from retrievers import WebSearchRetriever


class CRAGSystem:
    """
    Complete Corrective RAG system implemented as a LangGraph state machine.
    
    The graph has the following flow:
    
    retrieve → grade_documents → [route based on decision] 
        → proceed: generate → check_hallucination → [route]
        → augment: web_search → generate → check_hallucination → [route]  
        → fallback: web_search → generate → check_hallucination → [route]
    
    check_hallucination routes to:
        → consistent: finalize
        → inconsistent: regenerate (max 2 attempts) or flag_for_review
    """
    
    def __init__(
        self,
        vector_store: Chroma,
        llm_model: str = "gpt-4o",
        grader_model: str = "gpt-4o-mini",  # Cheaper model for grading tasks
        retrieval_k: int = 5,
        max_generation_attempts: int = 2
    ):
        self.vector_store = vector_store
        self.retrieval_k = retrieval_k
        self.max_generation_attempts = max_generation_attempts
        
        # Use a more capable model for generation, cheaper for validation
        self.generation_llm = ChatOpenAI(model=llm_model, temperature=0)
        self.grader_llm = ChatOpenAI(model=grader_model, temperature=0)
        
        self.relevance_grader = RelevanceGrader(llm=self.grader_llm)
        self.hallucination_detector = HallucinationDetector(llm=self.grader_llm)
        self.web_retriever = WebSearchRetriever()
        
        self.graph = self._build_graph()
    
    # ─────────────────────────────────────────────
    # Graph Nodes
    # ─────────────────────────────────────────────
    
    def retrieve(self, state: GraphState) -> Dict[str, Any]:
        """Node: Retrieve documents from vector store with similarity scores."""
        print(f"[RETRIEVE] Fetching top-{self.retrieval_k} documents...")
        
        # Use similarity_search_with_score to get both docs and scores
        results = self.vector_store.similarity_search_with_score(
            state.question, 
            k=self.retrieval_k
        )
        
        documents = []
        for doc, score in results:
            # Store the similarity score in metadata for the grader
            doc.metadata["embedding_similarity"] = float(score)
            documents.append(doc)
        
        return {"documents": documents}
    
    def grade_documents(self, state: GraphState) -> Dict[str, Any]:
        """Node: Grade retrieved documents for relevance and make routing decision."""
        print(f"[GRADE] Evaluating {len(state.documents)} documents...")
        
        similarity_scores = [
            doc.metadata.get("embedding_similarity", 0.5) 
            for doc in state.documents
        ]
        
        scored_docs = self.relevance_grader.grade_documents(
            state.documents, 
            similarity_scores, 
            state.question
        )
        
        relevance_scores = [score for _, score in scored_docs]
        decision, filtered_docs = compute_retrieval_decision(scored_docs)
        
        print(f"[GRADE] Decision: {decision} | "
              f"Useful docs: {len(filtered_docs)}/{len(state.documents)}")
        
        for doc, score in scored_docs:
            title = doc.metadata.get("title", "Unknown")[:50]
            print(f"  → {title}: {score.score} (confidence: {score.confidence:.2f})")
        
        return {
            "documents": filtered_docs,
            "relevance_scores": relevance_scores,
            "retrieval_decision": decision
        }
    
    def web_search(self, state: GraphState) -> Dict[str, Any]:
        """Node: Retrieve documents from web search."""
        print(f"[WEB SEARCH] Querying web for: {state.question[:80]}...")
        
        if state.retrieval_decision == "fallback":
            # For full fallback, use query expansion for better recall
            web_docs = self.web_retriever.retrieve_with_query_expansion(
                state.question, 
                self.grader_llm
            )
        else:
            # For augmentation, a direct query is fine
            web_docs = self.web_retriever.retrieve(state.question)
        
        print(f"[WEB SEARCH] Retrieved {len(web_docs)} web documents")
        return {"web_results": web_docs}
    
    def generate(self, state: GraphState) -> Dict[str, Any]:
        """Node: Generate answer from available documents."""
        # Combine local docs and web results based on the retrieval decision
        all_docs = state.documents + state.web_results
        has_mixed_sources = len(state.web_results) > 0 and len(state.documents) > 0
        
        print(f"[GENERATE] Generating from {len(all_docs)} documents "
              f"(attempt {state.generation_attempts + 1})...")
        
        if has_mixed_sources:
            prompt_template = GENERATION_PROMPT_MIXED_SOURCES
        else:
            prompt_template = GENERATION_PROMPT_STANDARD
        
        prompt = ChatPromptTemplate.from_template(prompt_template)
        formatted_docs = format_docs_for_generation(
            all_docs, 
            include_source_type=has_mixed_sources
        )
        
        chain = prompt | self.generation_llm
        response = chain.invoke({
            "documents": formatted_docs,
            "question": state.question
        })
        
        return {
            "generation": response.content,
            "generation_attempts": state.generation_attempts + 1
        }
    
    def check_hallucination(self, state: GraphState) -> Dict[str, Any]:
        """Node: Check whether the generated answer is grounded in sources."""
        all_docs = state.documents + state.web_results
        print(f"[HALLUCINATION CHECK] Verifying answer against {len(all_docs)} sources...")
        
        assessment = self.hallucination_detector.assess(
            question=state.question,
            answer=state.generation,
            documents=all_docs
        )
        
        if assessment.is_grounded:
            print(f"[HALLUCINATION CHECK] ✓ Answer is grounded "
                  f"(confidence: {assessment.confidence:.2f})")
        else:
            print(f"[HALLUCINATION CHECK] ✗ Found {len(assessment.unsupported_claims)} "
                  f"unsupported claims:")
            for claim in assessment.unsupported_claims:
                print(f"  → {claim[:100]}")
        
        return {"hallucination_assessment": assessment}
    
    def finalize(self, state: GraphState) -> Dict[str, Any]:
        """Node: Accept the generation as the final answer."""
        return {"final_answer": state.generation}
    
    def flag_for_review(self, state: GraphState) -> Dict[str, Any]:
        """
        Node: Answer couldn't be verified after max attempts.
        Returns a safe response rather than a potentially hallucinated one.
        """
        print("[FLAG] Answer could not be verified. Returning safe fallback.")
        safe_response = (
            f"I wasn't able to find sufficiently reliable information to answer "
            f"your question about: '{state.question}'. "
            f"Please consult an authoritative source directly."
        )
        return {"final_answer": safe_response, "error": "hallucination_unresolved"}
    
    # ─────────────────────────────────────────────
    # Routing Functions (Conditional Edges)
    # ─────────────────────────────────────────────
    
    def route_after_grading(self, state: GraphState) -> str:
        """Route based on retrieval decision."""
        if state.retrieval_decision == "proceed":
            return "generate"
        elif state.retrieval_decision in ("fallback", "augment"):
            return "web_search"
        else:
            return "generate"  # Safe default
    
    def route_after_hallucination_check(self, state: GraphState) -> str:
        """Route based on hallucination assessment."""
        assessment = state.hallucination_assessment
        
        if assessment.is_grounded:
            return "finalize"
        
        if state.generation_attempts < self.max_generation_attempts:
            print(f"[ROUTE] Hallucination detected, retrying generation "
                  f"(attempt {state.generation_attempts + 1}/{self.max_generation_attempts})")
            return "generate"  # Retry generation
        
        return "flag_for_review"
    
    # ─────────────────────────────────────────────
    # Graph Assembly
    # ─────────────────────────────────────────────
    
    def _build_graph(self) -> Any:
        """Assemble the LangGraph state machine."""
        workflow = StateGraph(GraphState)
        
        # Add nodes
        workflow.add_node("retrieve", self.retrieve)
        workflow.add_node("grade_documents", self.grade_documents)
        workflow.add_node("web_search", self.web_search)
        workflow.add_node("generate", self.generate)
        workflow.add_node("check_hallucination", self.check_hallucination)
        workflow.add_node("finalize", self.finalize)
        workflow.add_node("flag_for_review", self.flag_for_review)
        
        # Define edges
        workflow.set_entry_point("retrieve")
        workflow.add_edge("retrieve", "grade_documents")
        
        # Conditional routing after grading
        workflow.add_conditional_edges(
            "grade_documents",
            self.route_after_grading,
            {
                "generate": "generate",
                "web_search": "web_search"
            }
        )
        
        # Web search always flows to generation
        workflow.add_edge("web_search", "generate")
        
        # Generation always flows to hallucination check
        workflow.add_edge("generate", "check_hallucination")
        
        # Conditional routing after hallucination check
        workflow.add_conditional_edges(
            "check_hallucination",
            self.route_after_hallucination_check,
            {
                "finalize": "finalize",
                "generate": "generate",
                "flag_for_review": "flag_for_review"
            }
        )
        
        # Terminal nodes
        workflow.add_edge("finalize", END)
        workflow.add_edge("flag_for_review", END)
        
        return workflow.compile()
    
    def run(self, question: str) -> Dict[str, Any]:
        """Execute the CRAG pipeline for a given question."""
        initial_state = GraphState(question=question)
        final_state = self.graph.invoke(initial_state)
        return {
            "answer": final_state.get("final_answer"),
            "retrieval_decision": final_state.get("retrieval_decision"),
            "generation_attempts": final_state.get("generation_attempts"),
            "is_grounded": final_state.get("hallucination_assessment", {}).get("is_grounded"),
            "error": final_state.get("error")
        }

---

Performance Considerations and Latency Optimization

Let's be direct: CRAG adds latency. A standard RAG pipeline might take 1-2 seconds. A CRAG pipeline with full LLM-based document grading and hallucination checking can take 8-15 seconds for the "proceed" path, and 15-25 seconds for the "fallback" path.

Here's where you can recover latency without sacrificing quality:

1. Run document grading in parallel:

import asyncio
from typing import List, Tuple


class AsyncRelevanceGrader(RelevanceGrader):
    """Async version that grades all documents concurrently."""
    
    async def _llm_grade_async(
        self, 
        doc: Document, 
        question: str
    ) -> RelevanceScore:
        """Async LLM grading."""
        title = doc.metadata.get("title", "Unknown")
        content = doc.page_content[:1500]
        
        try:
            result = await self.chain.ainvoke({
                "question": question,
                "doc_title": title,
                "doc_content": content,
                "format_instructions": ""
            })
            return result
        except Exception:
            return RelevanceScore(score="partial", reasoning="Async grading failed", confidence=0.3)
    
    async def grade_documents_async(
        self,
        documents: List[Document],
        similarity_scores: List[float],
        question: str
    ) -> List[Tuple[Document, RelevanceScore]]:
        """Grade all documents concurrently using asyncio.gather."""
        
        # Fast-reject low-similarity docs synchronously (no LLM needed)
        tasks = []
        fast_rejected = []
        
        for doc, sim_score in zip(documents, similarity_scores):
            if sim_score < self.reject_threshold:
                fast_rejected.append((doc, RelevanceScore(
                    score="no",
                    reasoning=f"Similarity {sim_score:.3f} below threshold",
                    confidence=0.9
                )))
                tasks.append(None)
            else:
                tasks.append(self._llm_grade_async(doc, question))
        
        # Execute all LLM gradings concurrently
        llm_tasks = [t for t in tasks if t is not None]
        if llm_tasks:
            llm_results = await asyncio.gather(*llm_tasks, return_exceptions=True)
        else:
            llm_results = []
        
        # Reconstruct results in order
        result = []
        llm_idx = 0
        for i, (doc, task) in enumerate(zip(documents, tasks)):
            if task is None:
                result.append(fast_rejected[i - llm_idx])
                llm_idx += 1
            else:
                score = llm_results[llm_idx] if not isinstance(llm_results[llm_idx], Exception) \
                    else RelevanceScore(score="partial", reasoning="Error", confidence=0.3)
                result.append((doc, score))
                llm_idx += 1
        
        return result

2. Cache grading decisions for repeated queries:

import hashlib
from functools import lru_cache


def make_cache_key(doc_content: str, question: str) -> str:
    """Create a stable cache key for a (document, question) pair."""
    combined = f"{question[:200]}||{doc_content[:500]}"
    return hashlib.sha256(combined.encode()).hexdigest()

3. Use a cheaper grader model strategically:

In our CRAGSystem constructor, we already use gpt-4o-mini for grading. This is deliberate. The relevance grading task requires judgment but not deep reasoning — gpt-4o-mini performs nearly as well as gpt-4o for binary/ternary classification tasks with well-structured prompts. Save the expensive model for generation.

Benchmark context: In practice, grading 5 documents in parallel with gpt-4o-mini adds roughly 1.5-2.5 seconds of latency (depending on prompt length and API load). The hallucination check adds another 1.5-3 seconds. The total CRAG overhead is 3-6 seconds on top of standard retrieval and generation time. Whether that's acceptable depends entirely on your use case — a real-time chatbot has different requirements than an overnight research report pipeline.

---

Hands-On Exercise

Now you'll implement CRAG against a realistic document corpus. We'll use financial compliance documents as the domain.

Setup:

# main.py — exercise scaffold
from langchain_openai import OpenAIEmbeddings
from langchain_community.vectorstores import Chroma
from langchain_core.documents import Document

# Sample corpus: mix of relevant and tangentially-related documents
# In practice, load these from your actual document store
sample_documents = [
    Document(
        page_content="""The Know Your Customer (KYC) verification process requires 
        all new account holders to submit a government-issued photo ID and proof of 
        address within 30 days of account opening. Failure to complete KYC verification 
        will result in account suspension. Accepted IDs include passport, driver's 
        license, and national identity card.""",
        metadata={"title": "KYC Requirements v2.3", "source": "compliance/kyc.pdf", "year": 2023}
    ),
    Document(
        page_content="""Anti-Money Laundering (AML) screening is performed on all 
        transactions exceeding $10,000. Suspicious Activity Reports (SARs) must be 
        filed within 30 days of detecting suspicious activity. All customer-facing 
        staff must complete annual AML training.""",
        metadata={"title": "AML Policy", "source": "compliance/aml.pdf", "year": 2023}
    ),
    Document(
        page_content="""Quarterly financial statements are prepared in accordance with 
        GAAP standards. Revenue recognition follows ASC 606 guidelines. The fiscal year 
        ends December 31st. External audits are conducted by PricewaterhouseCoopers.""",
        metadata={"title": "Financial Reporting Standards", "source": "finance/reporting.pdf", "year": 2022}
    ),
    Document(
        page_content="""Employee expense reimbursements must be submitted within 60 days 
        of the expense date. Receipts are required for all expenses over $25. Travel 
        expenses require pre-approval for amounts over $500. Reimbursements are 
        processed in the next payroll cycle.""",
        metadata={"title": "Expense Policy", "source": "hr/expenses.pdf", "year": 2023}
    ),
]

# Initialize vector store
embeddings = OpenAIEmbeddings()
vector_store = Chroma.from_documents(sample_documents, embeddings)

# Initialize CRAG system
crag = CRAGSystem(vector_store=vector_store)

# Test queries — one well-served by the corpus, one that requires web search
queries = [
    "What documents are required for KYC verification?",  # Should: proceed
    "What is the current Basel III capital adequacy ratio requirement?",  # Should: fallback
    "What are the AML reporting requirements?",  # Should: proceed  
]

for query in queries:
    print(f"\n{'='*60}")
    print(f"QUERY: {query}")
    print('='*60)
    result = crag.run(query)
    print(f"\nANSWER: {result['answer']}")
    print(f"Decision: {result['retrieval_decision']}")
    print(f"Grounded: {result['is_grounded']}")
    print(f"Generation attempts: {result['generation_attempts']}")

Your challenge: Extend the system with these two additions:

1. Add a source citation requirement: Modify the generation prompt so the LLM must cite specific document numbers (e.g., "[1]", "[3]") in its answer. Update the hallucination detector to verify that cited documents actually contain the claimed information.

2. Add a confidence threshold: Implement a mechanism where if hallucination_assessment.confidence < 0.6, the system routes to "flag_for_review" even if is_grounded is True. This handles the case where the detector itself is uncertain.

---

Common Mistakes & Troubleshooting

Mistake 1: Using ChromaDB distance scores as probability scores

ChromaDB returns L2 distances or cosine distances depending on configuration. A lower L2 distance means higher similarity. If you're comparing these to a threshold without checking which metric your store uses, your fast-reject logic will be inverted — you'll reject good documents and keep bad ones.

# Check your collection's distance metric
collection = vector_store._collection
metadata = collection.metadata
print(metadata)  # Look for "hnsw:space" key

# Chroma uses cosine similarity by default when created with OpenAI embeddings
# But similarity_search_with_score returns: higher score = more similar for cosine
# For IP (inner product): same
# For L2: LOWER score = more similar (it's a distance, not similarity)

Mistake 2: Grading documents against the question embedding instead of the question text

Some implementations try to be clever and compare document embeddings to query embeddings directly in the grader. But the LLM grader needs the actual text of the question to understand semantic nuances that embeddings lose. Always pass the raw question string to the LLM grader.

Mistake 3: Not setting a maximum generation retry limit

Without the max_generation_attempts guard, a question that the available documents genuinely cannot answer will cause an infinite loop: generate → hallucination detected → generate → hallucination detected... The LLM will keep trying to answer from inadequate sources and keep failing.

# This silent bug is hard to find in production
# Always verify this in your state machine
if state.generation_attempts >= self.max_generation_attempts:
    # Must route to flag_for_review, not back to generate

Mistake 4: Running hallucination detection on web search content with the same threshold as corpus content

Web content has different reliability characteristics than your curated corpus. You should apply stricter thresholds when the generation is based partially or entirely on web results.

def route_after_hallucination_check(self, state: GraphState) -> str:
    assessment = state.hallucination_assessment
    
    # Apply stricter threshold when web results were used
    threshold = 0.85 if state.web_results else 0.70
    
    if assessment.is_grounded and assessment.confidence >= threshold:
        return "finalize"
    # ...rest of routing logic

Mistake 5: Forgetting that the grader LLM itself can hallucinate

The RelevanceScore and HallucinationAssessment are generated by an LLM. That LLM can make mistakes. This is why we capture confidence in the structured output and why we default to the conservative option (treat as partial/not grounded) when the grader is uncertain. Never assume 100% accuracy from your validators.

Debugging tip: Add a debug_mode flag to your CRAGSystem that saves the full intermediate state at each node to a log file. When a query produces a wrong answer, you can trace exactly which document got graded as "yes" that shouldn't have, or which claim the hallucination detector missed.

def grade_documents(self, state: GraphState) -> Dict[str, Any]:
    # ... grading logic ...
    
    if self.debug_mode:
        import json
        debug_entry = {
            "question": state.question,
            "documents": [
                {
                    "title": doc.metadata.get("title"),
                    "score": score.score,
                    "reasoning": score.reasoning,
                    "confidence": score.confidence
                }
                for doc, score in scored_docs
            ],
            "decision": decision
        }
        with open("crag_debug.jsonl", "a") as f:
            f.write(json.dumps(debug_entry) + "\n")

---

Summary & Next Steps

You've built a production-capable Corrective RAG system that addresses both of the fundamental RAG failure modes: retrieving irrelevant documents and generating hallucinated answers. Let's take stock of what we built and the key architectural decisions that make it robust.

The key architectural principles:

• Two-checkpoint validation: Retrieval quality is validated before generation, output faithfulness is validated after. Neither checkpoint can be skipped without weakening the system.
• Three-way routing: "Proceed," "augment," and "fallback" give the system proportional responses to different levels of retrieval confidence. Binary routing (proceed or fallback) either wastes resources or loses useful context.
• Model stratification: Use cheap models (gpt-4o-mini) for classification tasks (grading, hallucination detection) and capable models (gpt-4o) for generation. The grading tasks have clear rubrics; the generation task requires judgment.
• Conservative failure modes: When uncertain, flag for review rather than returning a potentially wrong answer. A "I couldn't find reliable information" response is far less harmful than a confident hallucination.
• Bounded retry loops: Always cap generation retry attempts. Infinite loops in production are catastrophic.

What to explore next:

• Adaptive retrieval: Rather than fixed k=5 retrieval, use the relevance scores from the first retrieval attempt to decide how many more documents to fetch. If the top 5 all score "no," retrieve 15 and grade the new batch before deciding.

• Fine-tuned graders: The relevance grading task is ideal for fine-tuning on domain-specific examples. A fine-tuned gpt-4o-mini grader on your specific document corpus will outperform a general-purpose prompted grader and cost a fraction of the inference price.

• Retrieval reranking: Integrate a cross-encoder reranker (like Cohere Rerank or a local model) between the vector search and the LLM grader. Cross-encoders consider the query and document together and are significantly more accurate than bi-encoder similarity alone.

• Evaluation harness: Build an offline evaluation suite using LangSmith or RAGAS to measure your CRAG pipeline's performance on a golden dataset. Track precision/recall at the retrieval decision boundary and faithfulness scores over time as your corpus changes.

• Graph-level observability: Integrate LangSmith tracing into your LangGraph nodes to get a full timeline of each query execution, including which node took how long and what intermediate state looked like. This is invaluable for production debugging.

The CRAG architecture we've built here is a solid foundation. Production deployments will want to add caching layers, async execution throughout, rate limiting on the web search fallback, and governance logging for regulated industries. But the core logic — retrieve, grade, route, generate, validate — is the pattern you'll build on.