Corrective RAG: Complete Implementation 2026

Overview

Corrective RAG is a specialized retrieval pattern that focuses on self-correcting retrieval with quality assessment. This guide shows you how to build a production-ready system using Weaviate.

Why Corrective RAG?

Standard RAG often struggles with complex queries, multi-hop reasoning, or domain-specific content. Corrective RAG addresses these limitations through self-correcting retrieval with quality assessment.

Architecture

Query → [Corrective Preprocessing] → Vector Search → [Context Processing] → LLM → Response
              ↓                                           ↑
         Query expansion                         Reranking + filtering

Implementation

Setup

bash
pip install langchain langchain-openai weaviate tiktoken

python
import os
from langchain_openai import ChatOpenAI, OpenAIEmbeddings
from langchain_core.documents import Document
from langchain_core.prompts import ChatPromptTemplate
from langchain_core.output_parsers import StrOutputParser
from langchain_core.runnables import RunnablePassthrough
Initialize
llm = ChatOpenAI(model="gpt-4o-mini", temperature=0)
embeddings = OpenAIEmbeddings(model="text-embedding-3-small")

Corrective Retriever

python
from langchain.retrievers import CorrectiveRetriever
from langchain_weaviate import WeaviateVectorStore
Build vector store
vectorstore = WeaviateVectorStore.from_documents(
    documents=your_documents,
    embedding=embeddings,
    index_name="my-rag-index"
)
Create specialized retriever for self-correcting retrieval with quality assessment
retriever = vectorstore.as_retriever(
    search_type="mmr",
    search_kwargs={
        "k": 6,
        "fetch_k": 25,
        "lambda_mult": 0.7
    }
)

Document Processing

python
from langchain.text_splitter import RecursiveCharacterTextSplitter
from langchain_community.document_loaders import DirectoryLoader
def load_and_process_documents(directory: str) -> list[Document]:
    """Load and process documents for Corrective RAG."""
    
    # Load documents
    loader = DirectoryLoader(directory, glob="**/*.txt")
    raw_docs = loader.load()
    
    # Split with overlap for context preservation
    splitter = RecursiveCharacterTextSplitter(
        chunk_size=800,
        chunk_overlap=150,
        separators=["\n\n", "\n", ". ", " ", ""]
    )
    
    chunks = splitter.split_documents(raw_docs)
    
    # Add metadata for self-correcting retrieval with quality assessment
    for i, chunk in enumerate(chunks):
        chunk.metadata.update({
            "chunk_id": i,
            "variant": "Corrective",
            "chunk_length": len(chunk.page_content)
        })
    
    print(f"Created {len(chunks)} chunks from {len(raw_docs)} documents")
    return chunks
chunks = load_and_process_documents("./documents/")

RAG Chain

python
def create_corrective_chain(retriever):
    """Create Corrective RAG chain optimized for self-correcting retrieval with quality assessment."""
    
    prompt = ChatPromptTemplate.from_messages([
        ("system", """You are a knowledgeable AI assistant.
        Use the following retrieved context to answer questions accurately.
        
        Context:
        {context}
        
        Guidelines for self-correcting retrieval with quality assessment:
        - Reference specific information from the context
        - If information is not in context, say so clearly
        - Cite sources when possible
        - Be concise but complete"""),
        ("human", "{question}")
    ])
    
    def format_context(docs: list[Document]) -> str:
        formatted = []
        for doc in docs:
            source = doc.metadata.get('source', 'Unknown')
            formatted.append(f"[Source: {source}]\n{doc.page_content}")
        return "\n\n---\n\n".join(formatted)
    
    chain = (
        {
            "context": retriever | format_context,
            "question": RunnablePassthrough()
        }
        | prompt
        | llm
        | StrOutputParser()
    )
    
    return chain
Build and use the chain
rag_chain = create_corrective_chain(retriever)
answer = rag_chain.invoke("Your question here")

Advanced: Streaming with Sources

python
from langchain_core.runnables import RunnableParallel
def create_rag_with_sources(retriever):
    """RAG that returns answer + source documents."""
    
    prompt = ChatPromptTemplate.from_messages([
        ("system", "Answer based on context. Be accurate and cite sources.\n\nContext: {context}"),
        ("human", "{question}")
    ])
    
    # Run retrieval and formatting in parallel
    setup = RunnableParallel(
        context=retriever | (lambda docs: "\n\n".join(d.page_content for d in docs)),
        question=RunnablePassthrough(),
        source_documents=retriever
    )
    
    chain = setup | {
        "answer": prompt | llm | StrOutputParser(),
        "sources": lambda x: [d.metadata.get('source') for d in x['source_documents']]
    }
    
    return chain
chain_with_sources = create_rag_with_sources(retriever)
result = chain_with_sources.invoke("What is the main topic?")
print(f"Answer: {result['answer']}")
print(f"Sources: {result['sources']}")

Evaluation

python
from ragas import evaluate
from ragas.metrics import faithfulness, answer_relevancy, context_precision, context_recall
from datasets import Dataset
def evaluate_rag(test_cases: list[dict]) -> dict:
    """Evaluate Corrective RAG quality with RAGAS."""
    
    dataset = Dataset.from_list(test_cases)
    
    result = evaluate(
        dataset,
        metrics=[
            faithfulness,
            answer_relevancy,
            context_precision,
            context_recall
        ]
    )
    
    print(f"Faithfulness: {result['faithfulness']:.3f}")
    print(f"Answer Relevancy: {result['answer_relevancy']:.3f}")
    print(f"Context Precision: {result['context_precision']:.3f}")
    print(f"Context Recall: {result['context_recall']:.3f}")
    
    return result
test_cases = [
    {
        "question": "What are the key features?",
        "answer": rag_chain.invoke("What are the key features?"),
        "contexts": [d.page_content for d in retriever.invoke("What are the key features?")],
        "ground_truth": "Expected answer..."
    }
]
evaluate_rag(test_cases)

Performance Tips

Embedding cache: Cache embeddings to avoid recomputing

Async retrieval: Use async for concurrent document retrieval

Batch indexing: Index documents in batches of 100

Model selection: Use gpt-4o-mini for cost, gpt-4o for quality

Conclusion

Corrective RAG with Weaviate provides an excellent foundation for self-correcting retrieval with quality assessment. The patterns shown here are production-tested and scalable.

Start with the basic implementation, measure quality with RAGAS, then iterate based on metrics.

---

*Corrective RAG implementation | Weaviate | May 2026*