Self-Correcting RAG with Hallucination Detection

Introduction

30 Second Summary

You've probably asked an AI a question and gotten a confident, completely wrong answer. In production systems, these hallucinations can break user trust and cause real damage.

In this project, you will build a self-correcting RAG pipeline that automatically detects poor retrieval, rewrites its own queries to recover, and scores every response for faithfulness before it reaches the user.

What You'll Build

A multi-agent corrective RAG system powered by LangGraph, Qdrant, DeepEval, and Chainlit that answers questions from a company policy knowledge base while self-healing bad retrieval and evaluating its own outputs.

By the end of this project, you'll have:

• A corrective RAG graph in LangGraph with retriever, critic, rewriter, and generator agents that automatically rewrite queries when retrieval quality is poor.
• A Qdrant vector database with indexed company policy documents and a DeepEval evaluation harness using Groq's Llama 3.3 70B as the judge model.
• A Chainlit chat UI with Chain of Thought visualization and real-time evaluation scores, deployed via Docker Compose.
• Secret Mission: Calibrate your auto-judge against human judgment using Cohen's kappa and tune evaluation prompts to improve agreement.

Are there any prerequisites?

You should be comfortable writing Python and familiar with basic RAG concepts (embeddings, vector search, retrieval-augmented generation).

You will also need Docker Desktop installed on your machine. Step 1 walks you through everything else.

Set Up Your Environment

Before building any agents or evaluation gates, you need a working foundation. This means a Python environment with the right packages, a running vector database, and a connection to your LLM provider.

In this step, you'll install your dependencies, spin up Qdrant in a Docker container, and verify that both Qdrant and Groq are responding before you write a single line of application code.

In this step, get ready to:

• Install Python dependencies and configure environment files.
• Start a Qdrant vector database with Docker.
• Verify the full setup with a smoke test.

Install Python dependencies

This project uses several libraries that work together. LangGraph orchestrates the agents, DeepEval evaluates responses for hallucination, and Chainlit provides the chat UI.

• Press Cmd+Space (macOS) or the Windows key (Windows) to open your search bar.
• Type Terminal and press Enter to open it.
• Move to your Desktop by running this command:

cd ~/Desktop

• Create your project folder and move into it by running these commands:

mkdir rag-pipeline
cd rag-pipeline

• Create a Python virtual environment and activate it by running:

python3 -m venv venv
source venv/bin/activate

• Confirm your virtual environment is active by running:

which python3

You should see a path that includes rag-pipeline/venv/bin/python3. This confirms you are working inside the virtual environment.

Seeing your system Python path instead?

If which python3 shows something like /usr/bin/python3, the virtual environment did not activate. Make sure you ran source venv/bin/activate (not just venv/bin/activate without source).

Help me activate my Python virtual environment.

• Press Cmd+Space (macOS) or the Windows key (Windows) to open your search bar.
• Type Visual Studio Code and press Enter to open it.
• Click File in the top menu bar.
• Click Open Folder.
• Select the rag-pipeline folder on your Desktop, then click Open.

Now you'll create the requirements.txt file that lists every package the project needs.

• In VS Code's Explorer sidebar, click the New File icon (a page with a plus sign) at the top of the file tree.
• Name the file requirements.txt and press Enter.
• Paste the following content into the file:

langgraph>=0.3.0
langchain-groq>=0.3.0
langchain-huggingface>=0.1.0
langchain-qdrant>=0.2.0
langchain-core>=0.3.0
deepeval>=2.0.0
chainlit>=2.0.0
qdrant-client>=1.12.0
sentence-transformers>=3.0.0
python-dotenv>=1.0.0
pydantic>=2.0.0

What are all these packages?

• langgraph builds the multi-agent graph that orchestrates retrieval, grading, rewriting, and generation.
• langchain-groq connects to the Groq API for fast LLM inference.
• langchain-huggingface and sentence-transformers run the local embedding model on your CPU.
• deepeval provides the faithfulness and hallucination metrics that score every response.
• chainlit serves the chat UI with built-in Chain of Thought visualization.

• Save requirements.txt by pressing Cmd+S (macOS) or Ctrl+S (Windows).

Now install all the packages. Switch back to your terminal window (the one where you activated the virtual environment).

• Install all project dependencies by running:

pip install -r requirements.txt

This will take a few minutes

The sentence-transformers package downloads PyTorch and several ML libraries. Expect the install to take 2-5 minutes depending on your internet speed.

You should see Successfully installed followed by a long list of package names at the end of the output.

Seeing a permissions error or install failure?

Make sure your virtual environment is active. Your terminal prompt should show (venv) at the beginning. If it does not, run source venv/bin/activate again before retrying the install.

Help me fix my pip install error.

Next, you need a Groq API key. Groq provides free access to fast LLM inference with no credit card required.

✔️ I already have a Groq API key

Great. Keep your API key ready. You will paste it into a .env file in the next few steps.

ⓧ I need to sign up for Groq

• Open console.groq.com in your browser.
• Click Sign up and create an account using Google, GitHub, or email. No credit card is required.
• Verify your email if prompted.
• Click API Keys in the left sidebar, or go directly to console.groq.com/keys.
• Click Create API Key.
• Enter a name like rag-pipeline and click Submit.
• Copy the key immediately and save it somewhere safe. Groq shows the full key exactly once. If you close the dialog without copying, you will need to create a new key.

Now create the environment files that store your configuration.

• In VS Code, click the New File icon in the Explorer sidebar.
• Name the file .env.example and press Enter.
• Paste the following content into the file:

GROQ_API_KEY=your-groq-api-key-here
QDRANT_HOST=localhost
QDRANT_PORT=6333

What is this file for?

The .env.example file is a template that shows what environment variables the project needs, without containing real secrets. It is safe to commit to version control. Your actual .env file (created next) holds your real API key and should never be committed.

• Save .env.example.
• Create another new file in the Explorer sidebar.
• Name it .env and press Enter.
• Paste the following content into the file:

GROQ_API_KEY=your-groq-api-key-here
QDRANT_HOST=localhost
QDRANT_PORT=6333

• Replace your-groq-api-key-here with the actual Groq API key you copied earlier.
• Save .env.

Keep your API key safe

Never commit your .env file to version control. If you use Git, add .env to your .gitignore file.

• Confirm your files exist by running this in your terminal:

ls -a

You should see .env, .env.example, requirements.txt, and the venv folder in the output.

✔️ Awesome, I've got everything!

Great. Double check you have saved all three files.

ⓧ I'd like to double check the full code

Here is what each file should look like at this point.

langgraph>=0.3.0
langchain-groq>=0.3.0
langchain-huggingface>=0.1.0
langchain-qdrant>=0.2.0
langchain-core>=0.3.0
deepeval>=2.0.0
chainlit>=2.0.0
qdrant-client>=1.12.0
sentence-transformers>=3.0.0
python-dotenv>=1.0.0
pydantic>=2.0.0

GROQ_API_KEY=your-groq-api-key-here
QDRANT_HOST=localhost
QDRANT_PORT=6333

GROQ_API_KEY=your-actual-api-key
QDRANT_HOST=localhost
QDRANT_PORT=6333

Start Qdrant with Docker

Qdrant is a high-performance vector database built in Rust. You will use it to store and search document embeddings. It runs as a Docker container with a built-in web dashboard.

Why Qdrant over ChromaDB?

ChromaDB is popular for RAG prototypes, but Qdrant is a better fit for production-style work. Qdrant is Docker-native with persistent storage, has a built-in dashboard for inspecting collections, and returns query results significantly faster.

Since this project focuses on building production-grade infrastructure, Qdrant is the right choice.

• Press Cmd+Space (macOS) or the Windows key (Windows) to open your search bar.
• Type Docker Desktop and press Enter to open it.
• Wait until the Docker Desktop status bar in the bottom left shows Engine running.

• Pull and start the Qdrant container by running this command in your terminal:

docker run -d \
  --name rag-qdrant \
  -p 6333:6333 \
  -p 6334:6334 \
  -v qdrant_data:/qdrant/storage \
  qdrant/qdrant:latest

What does this command do?

• -d runs the container in the background so you get your terminal back.
• -p 6333:6333 exposes the REST API port. -p 6334:6334 exposes the gRPC port.
• -v qdrant_data:/qdrant/storage creates a named Docker volume so your data persists across container restarts.

Docker will pull the Qdrant image if this is your first time running it. This may take a minute.

• Verify the container is running by executing:

docker ps

You should see a row for rag-qdrant with status Up and ports 6333 and 6334 listed.

• Open your browser and go to http://localhost:6333/dashboard.

You should see the Qdrant web dashboard. It shows zero collections for now, which is expected.

Container not starting or port conflict?

If you see an error about port 6333 already being in use, another process is occupying that port. Stop it with docker stop rag-qdrant && docker rm rag-qdrant and try running the container again.

If Docker says the daemon is not running, make sure Docker Desktop is open and shows Engine running.

Help me troubleshoot my Qdrant Docker container.

Verify the setup

You have packages installed, Qdrant running, and a Groq API key configured. A quick smoke test confirms everything is connected before you start building.

• In VS Code, create a new file called test_setup.py in the Explorer sidebar.
• Paste the following code into the file:

from dotenv import load_dotenv
from qdrant_client import QdrantClient
from langchain_groq import ChatGroq

# Load environment variables from .env
load_dotenv()

# Test Qdrant connection
client = QdrantClient(host="localhost", port=6333)
collections = client.get_collections()
print(f"Qdrant connected! Collections: {collections}")

# Test Groq API
llm = ChatGroq(model_name="llama-3.1-8b-instant", temperature=0)
response = llm.invoke("Say hello in one word.")
print(f"Groq response: {response.content}")

print("All systems go!")

What does this code do?

• load_dotenv() reads your .env file so GROQ_API_KEY is available as an environment variable.
• QdrantClient connects to the Qdrant container on port 6333 and lists existing collections.
• ChatGroq sends a simple prompt to the Llama 3.1 8B model via Groq to verify the API key works.

• Save test_setup.py.
• Run the smoke test in your terminal by executing:

python3 test_setup.py

You should see three lines of output: a Qdrant connection confirmation showing zero collections, a one-word greeting from Groq, and All systems go! at the end.

Seeing an authentication or connection error?

If you see AuthenticationError or Invalid API Key, open your .env file and verify you replaced your-groq-api-key-here with your actual Groq API key. Make sure there are no extra spaces around the = sign.

If you see Connection refused, your Qdrant container may have stopped. Run docker ps to check if rag-qdrant is listed. If not, start it again with docker start rag-qdrant.

Help me troubleshoot my smoke test.

Once the test passes, you can delete test_setup.py. It was only needed to confirm your setup.

Your environment is locked and loaded. Next up, you will create the company policy documents and index them into Qdrant so your RAG agents have something to search.

Index Documents into Qdrant

Your environment is ready and both Groq and Qdrant are confirmed working. Now it's time to give your RAG system something to work with.

Every RAG pipeline starts with a knowledge base. In this step, you'll create a set of company policy documents, convert them into vector embeddings using a local model, and store them in Qdrant for fast similarity search.

In this step, get ready to:

• Create a document corpus of 8 company policy documents.
• Generate vector embeddings using a local HuggingFace model.
• Index all documents into Qdrant and verify retrieval works.

Create the document corpus and embedding pipeline

RAG systems retrieve relevant documents by comparing vector representations. You need a set of documents and an embedding model to generate those vectors.

The all-MiniLM-L6-v2 model from sentence-transformers produces 384-dimensional vectors. It runs entirely on your CPU with no API calls required.

• Create a new file called indexer.py in your project folder.
• Add the following code to define your document corpus and embedding pipeline:

import os
from dotenv import load_dotenv
from qdrant_client import QdrantClient
from qdrant_client.models import Distance, VectorParams, PointStruct
from langchain_huggingface import HuggingFaceEmbeddings

# Load environment variables from .env
load_dotenv()

# Company policy documents that form the RAG knowledge base
DOCUMENTS = [
    {
        "content": "Our company offers a 30-day return policy for all electronics purchased in-store or online. Items must be in original packaging with receipt. Refunds are processed within 5-7 business days after the item is received at our warehouse.",
        "metadata": {"source": "return_policy.md", "category": "returns"},
    },
    {
        "content": "Employee vacation policy: Full-time employees receive 15 days of paid vacation per year. Vacation days do not roll over to the next calendar year. Part-time employees receive vacation days on a pro-rated basis.",
        "metadata": {"source": "hr_policy.md", "category": "hr"},
    },
    {
        "content": "IT security policy: All employees must use two-factor authentication for company systems. Passwords must be at least 12 characters and changed every 90 days. Personal devices are not permitted on the corporate network without MDM enrollment.",
        "metadata": {"source": "it_security.md", "category": "security"},
    },
    {
        "content": "Data privacy policy: Customer data is encrypted at rest using AES-256 and in transit using TLS 1.3. We retain customer data for a maximum of 3 years after the last interaction. Customers may request data deletion at any time through our privacy portal.",
        "metadata": {"source": "data_privacy.md", "category": "privacy"},
    },
    {
        "content": "Remote work policy: Employees may work remotely up to 3 days per week with manager approval. A dedicated home office space with reliable internet is required. The company provides a one-time $500 stipend for home office equipment.",
        "metadata": {"source": "remote_work.md", "category": "hr"},
    },
    {
        "content": "Expense reimbursement policy: Business travel expenses are reimbursed within 14 business days of submitting receipts. Maximum hotel rate is $200 per night for domestic travel. International travel requires VP-level pre-approval.",
        "metadata": {"source": "expense_policy.md", "category": "finance"},
    },
    {
        "content": "Software procurement policy: All software purchases over $1000 require IT department review. SaaS subscriptions must go through the vendor security assessment process. Open source software may be used freely if it has an approved license (MIT, Apache 2.0, BSD).",
        "metadata": {"source": "procurement.md", "category": "it"},
    },
    {
        "content": "Incident response policy: Security incidents must be reported to the IT security team within 1 hour of discovery. The incident response team will triage within 4 hours. Post-incident reviews are mandatory within 5 business days of resolution.",
        "metadata": {"source": "incident_response.md", "category": "security"},
    },
]

# Collection and embedding configuration
COLLECTION_NAME = "company_policies"
EMBEDDING_MODEL = "all-MiniLM-L6-v2"
EMBEDDING_DIM = 384


# Factory function for the local embedding model
def create_embeddings():
    return HuggingFaceEmbeddings(model_name=EMBEDDING_MODEL)

What does this code do?

This sets up the foundation for your indexing pipeline:

• DOCUMENTS is a list of 8 company policy documents, each with text content and metadata (source file name and category). These simulate a real company knowledge base.
• COLLECTION_NAME, EMBEDDING_MODEL, and EMBEDDING_DIM configure the Qdrant collection name and the embedding model's output dimensions (384).
• create_embeddings() returns a HuggingFaceEmbeddings instance that runs the all-MiniLM-L6-v2 model locally on CPU. No API calls or API keys needed for embedding.

• Save indexer.py.

You won't see a change yet. The next substep adds the function that indexes these documents into Qdrant.

Store embeddings in Qdrant

With documents and embeddings defined, you need a function that connects to Qdrant, creates a vector collection, and uploads all the embedded documents. The function also runs a test query at the end to confirm everything works.

• In indexer.py, add the following code below the create_embeddings() function:

def index_documents():
    # Connect to the Qdrant instance using environment variables
    client = QdrantClient(
        host=os.getenv("QDRANT_HOST", "localhost"),
        port=int(os.getenv("QDRANT_PORT", 6333)),
    )

    # Reset the collection if it already exists
    if client.collection_exists(COLLECTION_NAME):
        client.delete_collection(COLLECTION_NAME)

    # Create a new collection with cosine similarity
    client.create_collection(
        collection_name=COLLECTION_NAME,
        vectors_config=VectorParams(size=EMBEDDING_DIM, distance=Distance.COSINE),
    )

    # Generate embeddings for all document texts
    embeddings = create_embeddings()
    texts = [doc["content"] for doc in DOCUMENTS]
    vectors = embeddings.embed_documents(texts)

    # Build Qdrant points with content and metadata payloads
    points = [
        PointStruct(
            id=i,
            vector=vectors[i],
            payload={"content": doc["content"], **doc["metadata"]},
        )
        for i, doc in enumerate(DOCUMENTS)
    ]

    # Upload all points to Qdrant
    client.upsert(collection_name=COLLECTION_NAME, points=points)
    print(f"Indexed {len(points)} documents into '{COLLECTION_NAME}'")

    # Verify retrieval with a test query
    results = client.query_points(
        collection_name=COLLECTION_NAME,
        query=embeddings.embed_query("What is the return policy?"),
        limit=2,
    )
    print(f"Test query returned {len(results.points)} results:")
    for point in results.points:
        print(f"  Score: {point.score:.3f} | {point.payload['content'][:80]}...")


if __name__ == "__main__":
    index_documents()

What does this code do?

• QdrantClient connects to the Qdrant instance using the host and port from your .env file.
• collection_exists() and delete_collection() reset the collection if you re-run the script. This makes the indexer safely re-runnable.
• create_collection() creates a new Qdrant collection configured for 384-dimension vectors with cosine distance. Cosine distance measures how similar two vectors are by the angle between them.
• embed_documents() converts all 8 document texts into 384-dimensional vectors in one batch.
• Each PointStruct bundles a vector with its payload (the original text, source filename, and category). The payload is what you'll retrieve later during RAG.
• The test query at the end searches for "What is the return policy?" and prints the top 2 results with similarity scores.

Why cosine distance?

Cosine distance compares the direction of two vectors, ignoring their magnitude. Two documents about the same topic will point in a similar direction even if one is much longer than the other. This makes it the standard choice for text similarity search.

• Save indexer.py.

Seeing a connection error?

• Make sure your Qdrant Docker container is still running. Check with docker ps and look for the Qdrant container.
• Verify that QDRANT_HOST is set to localhost and QDRANT_PORT is set to 6333 in your .env file.
• Why can't my script connect to Qdrant?

Run the indexer and test retrieval

Time to run the script and see your documents land in Qdrant.

• Run the indexer by executing this command:

python indexer.py

What should I see?

The first run takes a moment because sentence-transformers downloads the all-MiniLM-L6-v2 model (~90 MB). Subsequent runs are faster because the model is cached locally.

You should see output similar to this:

Indexed 8 documents into 'company_policies'
Test query returned 2 results:
  Score: 0.8XX | Our company offers a 30-day return policy for all electronics purchased in-...
  Score: 0.3XX | Expense reimbursement policy: Business travel expenses are reimbursed wit...

The top result should be the return policy document with a high similarity score. The exact scores will vary slightly, but the return policy document should clearly rank first.

• Open the Qdrant dashboard in your browser at http://localhost:6333/dashboard.
• You should see a collection called company_policies with 8 points listed.

Not seeing 8 documents?

• Check your terminal output for error messages. A Connection refused error means Qdrant is not running.
• If you see an import error, make sure your virtual environment is activated and all dependencies from requirements.txt are installed.
• If the dashboard shows 0 points, re-run python indexer.py and check for errors in the output.
• My indexer script runs but documents are not appearing in Qdrant.

✔️ Awesome, I've got everything!

Great. Double check you've saved indexer.py and that your Qdrant dashboard shows 8 indexed points.

ⓧ I'd like to double check the full code

Here is the complete indexer.py file. Compare it against yours line by line.

import os
from dotenv import load_dotenv
from qdrant_client import QdrantClient
from qdrant_client.models import Distance, VectorParams, PointStruct
from langchain_huggingface import HuggingFaceEmbeddings

load_dotenv()

DOCUMENTS = [
    {
        "content": "Our company offers a 30-day return policy for all electronics purchased in-store or online. Items must be in original packaging with receipt. Refunds are processed within 5-7 business days after the item is received at our warehouse.",
        "metadata": {"source": "return_policy.md", "category": "returns"},
    },
    {
        "content": "Employee vacation policy: Full-time employees receive 15 days of paid vacation per year. Vacation days do not roll over to the next calendar year. Part-time employees receive vacation days on a pro-rated basis.",
        "metadata": {"source": "hr_policy.md", "category": "hr"},
    },
    {
        "content": "IT security policy: All employees must use two-factor authentication for company systems. Passwords must be at least 12 characters and changed every 90 days. Personal devices are not permitted on the corporate network without MDM enrollment.",
        "metadata": {"source": "it_security.md", "category": "security"},
    },
    {
        "content": "Data privacy policy: Customer data is encrypted at rest using AES-256 and in transit using TLS 1.3. We retain customer data for a maximum of 3 years after the last interaction. Customers may request data deletion at any time through our privacy portal.",
        "metadata": {"source": "data_privacy.md", "category": "privacy"},
    },
    {
        "content": "Remote work policy: Employees may work remotely up to 3 days per week with manager approval. A dedicated home office space with reliable internet is required. The company provides a one-time $500 stipend for home office equipment.",
        "metadata": {"source": "remote_work.md", "category": "hr"},
    },
    {
        "content": "Expense reimbursement policy: Business travel expenses are reimbursed within 14 business days of submitting receipts. Maximum hotel rate is $200 per night for domestic travel. International travel requires VP-level pre-approval.",
        "metadata": {"source": "expense_policy.md", "category": "finance"},
    },
    {
        "content": "Software procurement policy: All software purchases over $1000 require IT department review. SaaS subscriptions must go through the vendor security assessment process. Open source software may be used freely if it has an approved license (MIT, Apache 2.0, BSD).",
        "metadata": {"source": "procurement.md", "category": "it"},
    },
    {
        "content": "Incident response policy: Security incidents must be reported to the IT security team within 1 hour of discovery. The incident response team will triage within 4 hours. Post-incident reviews are mandatory within 5 business days of resolution.",
        "metadata": {"source": "incident_response.md", "category": "security"},
    },
]

COLLECTION_NAME = "company_policies"
EMBEDDING_MODEL = "all-MiniLM-L6-v2"
EMBEDDING_DIM = 384


def create_embeddings():
    return HuggingFaceEmbeddings(model_name=EMBEDDING_MODEL)


def index_documents():
    client = QdrantClient(
        host=os.getenv("QDRANT_HOST", "localhost"),
        port=int(os.getenv("QDRANT_PORT", 6333)),
    )

    if client.collection_exists(COLLECTION_NAME):
        client.delete_collection(COLLECTION_NAME)

    client.create_collection(
        collection_name=COLLECTION_NAME,
        vectors_config=VectorParams(size=EMBEDDING_DIM, distance=Distance.COSINE),
    )

    embeddings = create_embeddings()
    texts = [doc["content"] for doc in DOCUMENTS]
    vectors = embeddings.embed_documents(texts)

    points = [
        PointStruct(
            id=i,
            vector=vectors[i],
            payload={"content": doc["content"], **doc["metadata"]},
        )
        for i, doc in enumerate(DOCUMENTS)
    ]

    client.upsert(collection_name=COLLECTION_NAME, points=points)
    print(f"Indexed {len(points)} documents into '{COLLECTION_NAME}'")

    results = client.query_points(
        collection_name=COLLECTION_NAME,
        query=embeddings.embed_query("What is the return policy?"),
        limit=2,
    )
    print(f"Test query returned {len(results.points)} results:")
    for point in results.points:
        print(f"  Score: {point.score:.3f} | {point.payload['content'][:80]}...")


if __name__ == "__main__":
    index_documents()

Your knowledge base is indexed and searchable. Next up, you'll build the multi-agent corrective RAG graph that retrieves from this collection, grades document quality, and automatically rewrites bad queries.

Build the Corrective RAG Graph

Your Qdrant vector store is loaded with eight company policy documents, ready to serve search results. Now it's time to build the intelligence layer that decides what to do with those results.

A basic RAG pipeline retrieves documents and generates an answer. The problem is that retrieval sometimes returns irrelevant context, and the generator happily produces an answer anyway. In this step, you'll build a corrective RAG graph in LangGraph where a critic agent grades retrieval quality and triggers automatic query rewriting when the context is poor.

In this step, get ready to:

• Define a graph state and a retriever node that searches Qdrant.
• Build critic, rewriter, and generator nodes with conditional routing.
• Assemble all nodes into a compiled LangGraph and test the self-correcting loop.

Define the graph state and retriever node

Every LangGraph graph needs a state object that flows through each node. Each node reads from the state, does its work, and writes its output back.

• Create a new file called rag_graph.py in your project folder.
• Add the imports, constants, state definition, and helper functions by copying this code:

import os
from typing import TypedDict, Literal
from dotenv import load_dotenv
from pydantic import BaseModel, Field
from langchain_groq import ChatGroq
from langchain_huggingface import HuggingFaceEmbeddings
from qdrant_client import QdrantClient
from langgraph.graph import StateGraph, START, END

load_dotenv()

# Must match the collection and model used in indexer.py
EMBEDDING_MODEL = "all-MiniLM-L6-v2"
COLLECTION_NAME = "company_policies"
# Maximum number of query rewrites before forcing a generation
MAX_RETRIES = 1


# The state that flows through every node in the graph
class RAGState(TypedDict):
    question: str
    documents: list[str]
    generation: str
    retries: int


# Pydantic model for structured grading output
class RelevanceGrade(BaseModel):
    binary_score: str = Field(
        description="Relevance score: 'yes' if relevant, 'no' if not relevant"
    )


# Factory functions for shared dependencies
def get_qdrant_client():
    return QdrantClient(
        host=os.getenv("QDRANT_HOST", "localhost"),
        port=int(os.getenv("QDRANT_PORT", 6333)),
    )


def get_embeddings():
    return HuggingFaceEmbeddings(model_name=EMBEDDING_MODEL)


def get_llm():
    return ChatGroq(
        model_name="llama-3.1-8b-instant",
        temperature=0,
    )


def get_grader_llm():
    return ChatGroq(
        model_name="llama-3.1-8b-instant",
        temperature=0,
    )

What does this code do?

RAGState is a TypedDict that carries four fields through the graph: the user's question, the retrieved documents, the final generation, and a retries counter that prevents infinite rewrite loops.

RelevanceGrade is a Pydantic model that forces the LLM grader to return a structured yes/no score instead of freeform text. The four factory functions centralize how the graph creates its Qdrant client, embedding model, and Groq LLM instances.

Why two separate LLM factories?

get_llm() and get_grader_llm() use the same model right now, but separating them makes it easy to swap one model for another later. For example, you might want a faster model for generation and a more capable one for grading.

You won't see any output yet. The next block adds the retriever node that uses these building blocks.

• Add the retrieve function below get_grader_llm() in rag_graph.py:

def retrieve(state: RAGState) -> dict:
    # Connect to Qdrant and embed the user's question
    client = get_qdrant_client()
    embeddings = get_embeddings()
    query_vector = embeddings.embed_query(state["question"])

    # Search for the top 3 most relevant documents
    results = client.query_points(
        collection_name=COLLECTION_NAME,
        query=query_vector,
        limit=3,
    )

    # Extract the text content from each result
    documents = [point.payload["content"] for point in results.points]
    return {"documents": documents}

What does this code do?

The retrieve function is the first node in the graph. It takes the current question from the state, embeds it, and searches Qdrant for the 3 most similar documents. It returns a dictionary with the documents key, which LangGraph merges back into the state automatically.

• Save rag_graph.py.

You won't see output from this node until the graph is assembled and tested in the final substep.

Build the critic, rewriter, and generator nodes

The retriever fetches documents, but it has no way to judge whether those documents actually answer the question. The next three functions form the self-correcting loop: a critic grades retrieval quality, a rewriter reformulates poor queries, and a generator produces the final answer.

• Add the grade_documents function below retrieve in rag_graph.py:

def grade_documents(state: RAGState) -> Literal["generate_answer", "rewrite_question"]:
    # Use structured output to force a yes/no relevance score
    grader = get_grader_llm().with_structured_output(RelevanceGrade)

    question = state["question"]
    context = "\n\n".join(state["documents"])

    prompt = (
        "You are a grader assessing relevance of retrieved documents to a user question.\n"
        f"Here is the retrieved context:\n\n{context}\n\n"
        f"Here is the user question: {question}\n"
        "If the documents contain information related to the question, grade as relevant.\n"
        "Give a binary score 'yes' or 'no'."
    )

    result = grader.invoke([{"role": "user", "content": prompt}])

    # Route based on grading result and retry budget
    if result.binary_score == "yes":
        return "generate_answer"
    elif state["retries"] >= MAX_RETRIES:
        return "generate_answer"
    else:
        return "rewrite_question"

What does this code do?

grade_documents is the routing function in the graph. It uses .with_structured_output(RelevanceGrade) to force the LLM to return a Pydantic object with a binary_score field instead of freeform text.

The function returns a string that is the name of the next node to run. If the documents are relevant, it routes to generate_answer. If they are irrelevant and retries are left, it routes to rewrite_question. If retries are exhausted, it forces generation anyway to avoid an infinite loop.

• Add the rewrite_question and generate_answer functions below grade_documents in rag_graph.py:

def rewrite_question(state: RAGState) -> dict:
    # Ask the LLM to reformulate the question for better retrieval
    llm = get_llm()
    prompt = (
        "You are a question rewriter. Look at the input question and try to reason "
        "about the underlying intent to produce a better search query.\n\n"
        f"Original question: {state['question']}\n\n"
        "Rewritten question:"
    )
    response = llm.invoke([{"role": "user", "content": prompt}])
    return {
        "question": response.content.strip(),
        "retries": state["retries"] + 1,
    }


def generate_answer(state: RAGState) -> dict:
    # Produce a grounded answer from the retrieved documents
    llm = get_llm()
    context = "\n\n".join(state["documents"])
    prompt = (
        "You are an assistant for question-answering tasks. Use the following "
        "retrieved context to answer the question. If you don't know the answer, "
        "say that you don't know. Keep the answer concise (3 sentences max).\n\n"
        f"Context: {context}\n\n"
        f"Question: {state['question']}\n\n"
        "Answer:"
    )
    response = llm.invoke([{"role": "user", "content": prompt}])
    return {"generation": response.content.strip()}

What do these functions do?

rewrite_question asks the LLM to reformulate a poorly-worded query into something the vector search is more likely to match. It increments retries so the graph can track how many rewrite attempts have been made.

generate_answer concatenates all retrieved documents into a single context string and asks the LLM to produce a concise answer grounded in that context. The prompt explicitly tells the model to say "I don't know" if the context is insufficient.

• Save rag_graph.py.

You won't see output yet. The next substep wires everything into a graph and runs it.

Assemble and test the graph

You have four functions: retrieve, grade, rewrite, and generate. Now you need to tell LangGraph how they connect. The StateGraph class lets you register nodes and define edges between them, including conditional edges that route to different nodes based on a function's return value.

• Add the build_graph function below generate_answer in rag_graph.py:

def build_graph() -> StateGraph:
    workflow = StateGraph(RAGState)

    # Register each function as a named node
    workflow.add_node("retrieve", retrieve)
    workflow.add_node("rewrite_question", rewrite_question)
    workflow.add_node("generate_answer", generate_answer)

    # Define how nodes connect
    workflow.add_edge(START, "retrieve")
    # grade_documents returns the name of the next node to run
    workflow.add_conditional_edges("retrieve", grade_documents)
    workflow.add_edge("rewrite_question", "retrieve")
    workflow.add_edge("generate_answer", END)

    return workflow.compile()

What does this code do?

add_node registers each function as a named node. add_edge creates a fixed connection between two nodes. add_conditional_edges takes a routing function (grade_documents) that returns the name of the next node to run.

The flow is: START goes to retrieve, then grade_documents decides whether to go to generate_answer or rewrite_question. If the question is rewritten, it loops back to retrieve for a second try. compile() finalizes the graph into a runnable object.

Now add a test block at the bottom of the file to run two queries through the graph. One is a direct question and the other is an ambiguous query that should trigger the corrective loop.

• Add the test block below build_graph in rag_graph.py:

if __name__ == "__main__":
    graph = build_graph()

    # Test 1: A clear, direct question should generate without rewriting
    print("--- Test 1: Direct question ---")
    result = graph.invoke({"question": "What is the return policy?", "documents": [], "generation": "", "retries": 0})
    print(f"Answer: {result['generation']}")
    print(f"Retries: {result['retries']}")

    # Test 2: A vague query should trigger the rewrite loop
    print("\n--- Test 2: Ambiguous question ---")
    result = graph.invoke({"question": "stuff about working from home equipment money", "documents": [], "generation": "", "retries": 0})
    print(f"Answer: {result['generation']}")
    print(f"Retries: {result['retries']}")

What do these tests check?

Test 1 asks a clear question that should match the return policy document directly. The critic should grade the retrieval as relevant and skip the rewrite loop, leaving retries at 0.

Test 2 uses a vague, poorly-worded query. The critic is more likely to grade the retrieval as irrelevant, triggering a rewrite before retrying retrieval. You should see retries at 1 if the rewrite loop fired.

• Save rag_graph.py.
• Run the graph tests by running this command:

python rag_graph.py

You should see two test results printed. Test 1 shows an answer about the 30-day return policy with Retries: 0. Test 2 shows an answer about the remote work stipend, potentially with Retries: 1 if the corrective loop triggered.

Retries may vary

The grading decision depends on the LLM's judgment. The ambiguous query in Test 2 might show Retries: 0 if the critic deems the initial retrieval relevant enough, or Retries: 1 if it triggers a rewrite. Both outcomes confirm the graph is working correctly.

Seeing an error?

• If you see Connection refused, make sure your Qdrant Docker container is still running. Check with docker ps and restart it if needed.
• If you see AuthenticationError or Invalid API key, double-check that your .env file has the correct GROQ_API_KEY value.
• If you see a RateLimitError, you may have hit the Groq free tier limit of 30 requests per minute. Wait a minute and try again.

Help me debug my RAG graph test output

✔️ Awesome, I've got everything!

Double check you've saved rag_graph.py and both tests printed answers successfully.

ⓧ I'd like to double check the full code

Here is the complete rag_graph.py file. Compare it against yours to spot any differences.

import os
from typing import TypedDict, Literal
from dotenv import load_dotenv
from pydantic import BaseModel, Field
from langchain_groq import ChatGroq
from langchain_huggingface import HuggingFaceEmbeddings
from qdrant_client import QdrantClient
from langgraph.graph import StateGraph, START, END

load_dotenv()

EMBEDDING_MODEL = "all-MiniLM-L6-v2"
COLLECTION_NAME = "company_policies"
MAX_RETRIES = 1


class RAGState(TypedDict):
    question: str
    documents: list[str]
    generation: str
    retries: int


class RelevanceGrade(BaseModel):
    binary_score: str = Field(
        description="Relevance score: 'yes' if relevant, 'no' if not relevant"
    )


def get_qdrant_client():
    return QdrantClient(
        host=os.getenv("QDRANT_HOST", "localhost"),
        port=int(os.getenv("QDRANT_PORT", 6333)),
    )


def get_embeddings():
    return HuggingFaceEmbeddings(model_name=EMBEDDING_MODEL)


def get_llm():
    return ChatGroq(
        model_name="llama-3.1-8b-instant",
        temperature=0,
    )


def get_grader_llm():
    return ChatGroq(
        model_name="llama-3.1-8b-instant",
        temperature=0,
    )


def retrieve(state: RAGState) -> dict:
    client = get_qdrant_client()
    embeddings = get_embeddings()
    query_vector = embeddings.embed_query(state["question"])

    results = client.query_points(
        collection_name=COLLECTION_NAME,
        query=query_vector,
        limit=3,
    )

    documents = [point.payload["content"] for point in results.points]
    return {"documents": documents}


def grade_documents(state: RAGState) -> Literal["generate_answer", "rewrite_question"]:
    grader = get_grader_llm().with_structured_output(RelevanceGrade)

    question = state["question"]
    context = "\n\n".join(state["documents"])

    prompt = (
        "You are a grader assessing relevance of retrieved documents to a user question.\n"
        f"Here is the retrieved context:\n\n{context}\n\n"
        f"Here is the user question: {question}\n"
        "If the documents contain information related to the question, grade as relevant.\n"
        "Give a binary score 'yes' or 'no'."
    )

    result = grader.invoke([{"role": "user", "content": prompt}])

    if result.binary_score == "yes":
        return "generate_answer"
    elif state["retries"] >= MAX_RETRIES:
        return "generate_answer"
    else:
        return "rewrite_question"


def rewrite_question(state: RAGState) -> dict:
    llm = get_llm()
    prompt = (
        "You are a question rewriter. Look at the input question and try to reason "
        "about the underlying intent to produce a better search query.\n\n"
        f"Original question: {state['question']}\n\n"
        "Rewritten question:"
    )
    response = llm.invoke([{"role": "user", "content": prompt}])
    return {
        "question": response.content.strip(),
        "retries": state["retries"] + 1,
    }


def generate_answer(state: RAGState) -> dict:
    llm = get_llm()
    context = "\n\n".join(state["documents"])
    prompt = (
        "You are an assistant for question-answering tasks. Use the following "
        "retrieved context to answer the question. If you don't know the answer, "
        "say that you don't know. Keep the answer concise (3 sentences max).\n\n"
        f"Context: {context}\n\n"
        f"Question: {state['question']}\n\n"
        "Answer:"
    )
    response = llm.invoke([{"role": "user", "content": prompt}])
    return {"generation": response.content.strip()}


def build_graph() -> StateGraph:
    workflow = StateGraph(RAGState)

    workflow.add_node("retrieve", retrieve)
    workflow.add_node("rewrite_question", rewrite_question)
    workflow.add_node("generate_answer", generate_answer)

    workflow.add_edge(START, "retrieve")
    workflow.add_conditional_edges("retrieve", grade_documents)
    workflow.add_edge("rewrite_question", "retrieve")
    workflow.add_edge("generate_answer", END)

    return workflow.compile()


if __name__ == "__main__":
    graph = build_graph()

    print("--- Test 1: Direct question ---")
    result = graph.invoke({"question": "What is the return policy?", "documents": [], "generation": "", "retries": 0})
    print(f"Answer: {result['generation']}")
    print(f"Retries: {result['retries']}")

    print("\n--- Test 2: Ambiguous question ---")
    result = graph.invoke({"question": "stuff about working from home equipment money", "documents": [], "generation": "", "retries": 0})
    print(f"Answer: {result['generation']}")
    print(f"Retries: {result['retries']}")

Your corrective RAG graph can now retrieve, judge, rewrite, and generate. Next up, you'll add an automated hallucination evaluation gate that scores every response for faithfulness before it reaches the user.

Add the Hallucination Evaluation Gate

Your corrective RAG graph can now retrieve, grade, rewrite, and generate answers. The next challenge is knowing whether those answers are actually trustworthy.

In production AI systems, you need an automated evaluation gate that checks every response before it reaches the user. You'll build that gate using DeepEval with a stronger LLM acting as the judge.

In this step, get ready to:

• Build a custom Groq judge wrapper for DeepEval using Llama 3.3 70B.
• Wire up faithfulness and hallucination metrics to score every response.
• Test the gate on both a faithful and a fabricated answer.

Build the Groq judge wrapper for DeepEval

DeepEval defaults to OpenAI for its judge model, but you can plug in any provider by subclassing DeepEvalBaseLLM. You'll create a GroqJudge that uses Llama 3.3 70B via Groq as the evaluation model.

• Create a new file called evaluator.py in your project folder and add the imports and GroqJudge class by copying the following code:

import os
from dotenv import load_dotenv
from deepeval.models import DeepEvalBaseLLM
from deepeval.metrics import FaithfulnessMetric, HallucinationMetric
from deepeval.test_case import LLMTestCase
from langchain_groq import ChatGroq

load_dotenv()


# Custom judge that uses Groq's Llama 3.3 70B for evaluation
class GroqJudge(DeepEvalBaseLLM):
    def __init__(self, model_name="llama-3.3-70b-versatile"):
        self.model_name = model_name
        self._model = None

    # Lazy-load the ChatGroq client on first use
    def load_model(self):
        if self._model is None:
            self._model = ChatGroq(
                model_name=self.model_name,
                temperature=0,
            )
        return self._model

    # Synchronous generation for DeepEval metrics
    def generate(self, prompt: str, **kwargs) -> str:
        model = self.load_model()
        response = model.invoke(prompt)
        return response.content

    # Async generation for DeepEval metrics
    async def a_generate(self, prompt: str, **kwargs) -> str:
        model = self.load_model()
        response = await model.ainvoke(prompt)
        return response.content

    def get_model_name(self) -> str:
        return self.model_name

What does this code do?

GroqJudge subclasses DeepEvalBaseLLM, which is the abstract base class DeepEval uses for custom LLM judges. You implement four required methods.

• load_model() creates and caches a ChatGroq client pointed at Llama 3.3 70B. Lazy loading means it only connects when the first evaluation runs.
• generate() and a_generate() handle synchronous and async LLM calls. DeepEval invokes these internally when running metrics.
• get_model_name() returns a string identifier for logging and display.

• You won't see a change yet. The next chunk adds the function that uses this class.

Why use a stronger model as the judge?

Your RAG pipeline generates answers with Llama 3.1 8B and judges them with Llama 3.3 70B. This is called model asymmetry. A weaker model cannot reliably catch errors made by a stronger one, so the judge must always be more capable than the generator.

This pattern is standard in production evaluation systems. The 70B model has the reasoning depth to decompose claims, check them against source documents, and explain its verdict.

Wire up FaithfulnessMetric and HallucinationMetric

Two metrics catch different problems. FaithfulnessMetric checks whether the answer contradicts its source documents. HallucinationMetric checks whether the answer fabricates information not present in the context.

• In evaluator.py, add the evaluate_response function below the GroqJudge class by copying the following code:

def evaluate_response(question: str, answer: str, retrieval_context: list[str]) -> dict:
    judge = GroqJudge()

    # Faithfulness: does the answer contradict its source documents?
    faithfulness_metric = FaithfulnessMetric(
        threshold=0.7,
        model=judge,
        include_reason=True,
        async_mode=False,
    )

    # Hallucination: does the answer fabricate information?
    hallucination_metric = HallucinationMetric(
        threshold=0.5,
        model=judge,
        include_reason=True,
        async_mode=False,
    )

    # Faithfulness uses retrieval_context (RAG pipeline output)
    faithfulness_case = LLMTestCase(
        input=question,
        actual_output=answer,
        retrieval_context=retrieval_context,
    )

    # Hallucination uses context (ground truth reference)
    hallucination_case = LLMTestCase(
        input=question,
        actual_output=answer,
        context=retrieval_context,
    )

    # Run both metrics and collect results
    faithfulness_metric.measure(faithfulness_case)
    hallucination_metric.measure(hallucination_case)

    return {
        "faithfulness": {
            "score": faithfulness_metric.score,
            "reason": faithfulness_metric.reason,
            "passed": faithfulness_metric.score >= 0.7,
        },
        "hallucination": {
            "score": hallucination_metric.score,
            "reason": hallucination_metric.reason,
            "passed": hallucination_metric.score <= 0.5,
        },
    }

What does this code do?

• FaithfulnessMetric extracts claims from the answer, then checks each one against the retrieval_context. The score equals supported claims divided by total claims. A threshold of 0.7 means at least 70% of claims must be supported to pass.
• HallucinationMetric checks whether the answer contradicts the provided context. A score at or below 0.5 means the answer is acceptably grounded.
• Notice the two different LLMTestCase objects. Faithfulness requires retrieval_context (what the RAG pipeline actually retrieved). Hallucination requires context (the ground truth reference). In this project, you pass the same documents for both.
• async_mode=False forces synchronous execution. This keeps the evaluation straightforward and avoids event loop conflicts when you integrate with Chainlit later.

• You won't see a change yet. The next chunk adds tests so you can run the evaluator.

Seeing an import error?

Make sure your virtual environment is activated. The deepeval package should already be installed from Step 1. If not, run pip install deepeval to install it.

Help me debug a DeepEval import error in evaluator.py

Test the evaluation gate

You'll run the evaluator on two test cases. The first is a faithful answer that closely matches the source document. The second deliberately fabricates details to confirm the gate catches hallucinations.

• In evaluator.py, add the test block at the bottom of the file by copying the following code:

if __name__ == "__main__":
    # Test 1: Answer that faithfully reflects the source document
    print("--- Test 1: Faithful response ---")
    result = evaluate_response(
        question="What is the return policy?",
        answer="The company offers a 30-day return policy for electronics. Items must be in original packaging with receipt. Refunds take 5-7 business days.",
        retrieval_context=[
            "Our company offers a 30-day return policy for all electronics purchased in-store or online. Items must be in original packaging with receipt. Refunds are processed within 5-7 business days after the item is received at our warehouse."
        ],
    )
    print(f"Faithfulness: {result['faithfulness']['score']:.2f} | {result['faithfulness']['reason']}")
    print(f"Hallucination: {result['hallucination']['score']:.2f} | {result['hallucination']['reason']}")

    # Test 2: Answer with fabricated details (60 days, free shipping, no receipt, instant refund)
    print("\n--- Test 2: Hallucinated response ---")
    result = evaluate_response(
        question="What is the return policy?",
        answer="The company offers a 60-day return policy with free shipping on all returns. No receipt needed. Refunds are instant.",
        retrieval_context=[
            "Our company offers a 30-day return policy for all electronics purchased in-store or online. Items must be in original packaging with receipt. Refunds are processed within 5-7 business days after the item is received at our warehouse."
        ],
    )
    print(f"Faithfulness: {result['faithfulness']['score']:.2f} | {result['faithfulness']['reason']}")
    print(f"Hallucination: {result['hallucination']['score']:.2f} | {result['hallucination']['reason']}")

What does this code do?

• Test 1 passes an answer that closely paraphrases the source document. Every claim (30 days, original packaging, 5-7 business days) is supported by the context.
• Test 2 passes an answer where every detail is wrong. It says 60 days instead of 30, claims free shipping (never mentioned), drops the receipt requirement, and says refunds are instant instead of 5-7 days.
• The printed output shows the numeric score and the judge's reasoning for each metric. This reasoning is what your Chainlit UI will display later.

• Save evaluator.py.
• Run the evaluator by running this command:

python evaluator.py

What should I see?

The evaluation makes several Groq API calls, so it may take 15-30 seconds per test case. The Llama 3.3 70B model is doing real reasoning work behind the scenes.

For Test 1 (faithful), you should see a high faithfulness score (close to 1.0) and a low hallucination score. For Test 2 (hallucinated), you should see a low faithfulness score and a high hallucination score. Both tests should also print the judge's reasoning explaining why each claim was or was not supported.

Seeing a rate limit or timeout error?

The Groq free tier allows 30 requests per minute and 1,000 requests per day on Llama 3.3 70B. Each evaluation makes 2-4 judge calls per test case. If you hit a rate limit, wait 60 seconds and try again.

Make sure your GROQ_API_KEY is set correctly in your .env file. An invalid key produces an authentication error, not a rate limit error.

Help me debug a Groq API error when running evaluator.py

✔️ Awesome, I've got everything!

Double check you've saved your file before moving on.

ⓧ I'd like to double check the full code

Here is the complete evaluator.py file for comparison.

import os
from dotenv import load_dotenv
from deepeval.models import DeepEvalBaseLLM
from deepeval.metrics import FaithfulnessMetric, HallucinationMetric
from deepeval.test_case import LLMTestCase
from langchain_groq import ChatGroq

load_dotenv()


class GroqJudge(DeepEvalBaseLLM):
    def __init__(self, model_name="llama-3.3-70b-versatile"):
        self.model_name = model_name
        self._model = None

    def load_model(self):
        if self._model is None:
            self._model = ChatGroq(
                model_name=self.model_name,
                temperature=0,
            )
        return self._model

    def generate(self, prompt: str, **kwargs) -> str:
        model = self.load_model()
        response = model.invoke(prompt)
        return response.content

    async def a_generate(self, prompt: str, **kwargs) -> str:
        model = self.load_model()
        response = await model.ainvoke(prompt)
        return response.content

    def get_model_name(self) -> str:
        return self.model_name


def evaluate_response(question: str, answer: str, retrieval_context: list[str]) -> dict:
    judge = GroqJudge()

    faithfulness_metric = FaithfulnessMetric(
        threshold=0.7,
        model=judge,
        include_reason=True,
        async_mode=False,
    )

    hallucination_metric = HallucinationMetric(
        threshold=0.5,
        model=judge,
        include_reason=True,
        async_mode=False,
    )

    faithfulness_case = LLMTestCase(
        input=question,
        actual_output=answer,
        retrieval_context=retrieval_context,
    )

    hallucination_case = LLMTestCase(
        input=question,
        actual_output=answer,
        context=retrieval_context,
    )

    faithfulness_metric.measure(faithfulness_case)
    hallucination_metric.measure(hallucination_case)

    return {
        "faithfulness": {
            "score": faithfulness_metric.score,
            "reason": faithfulness_metric.reason,
            "passed": faithfulness_metric.score >= 0.7,
        },
        "hallucination": {
            "score": hallucination_metric.score,
            "reason": hallucination_metric.reason,
            "passed": hallucination_metric.score <= 0.5,
        },
    }


if __name__ == "__main__":
    print("--- Test 1: Faithful response ---")
    result = evaluate_response(
        question="What is the return policy?",
        answer="The company offers a 30-day return policy for electronics. Items must be in original packaging with receipt. Refunds take 5-7 business days.",
        retrieval_context=[
            "Our company offers a 30-day return policy for all electronics purchased in-store or online. Items must be in original packaging with receipt. Refunds are processed within 5-7 business days after the item is received at our warehouse."
        ],
    )
    print(f"Faithfulness: {result['faithfulness']['score']:.2f} | {result['faithfulness']['reason']}")
    print(f"Hallucination: {result['hallucination']['score']:.2f} | {result['hallucination']['reason']}")

    print("\n--- Test 2: Hallucinated response ---")
    result = evaluate_response(
        question="What is the return policy?",
        answer="The company offers a 60-day return policy with free shipping on all returns. No receipt needed. Refunds are instant.",
        retrieval_context=[
            "Our company offers a 30-day return policy for all electronics purchased in-store or online. Items must be in original packaging with receipt. Refunds are processed within 5-7 business days after the item is received at our warehouse."
        ],
    )
    print(f"Faithfulness: {result['faithfulness']['score']:.2f} | {result['faithfulness']['reason']}")
    print(f"Hallucination: {result['hallucination']['score']:.2f} | {result['hallucination']['reason']}")

Your evaluation gate is ready to catch hallucinations before they reach users. Next up, you'll build a Chainlit chat UI that surfaces these scores and agent reasoning steps in real time.

Build the Chainlit Chat UI

Your RAG pipeline retrieves, corrects, and generates answers. Your evaluation gate scores every response for faithfulness and hallucination. Both pieces work independently in the terminal.

Now you need a way for users to actually interact with the system and see what is happening under the hood. In this step, you will build a Chainlit chat UI that surfaces every agent reasoning step and trust score in real time.

In this step, get ready to:

• Create a Chainlit app that runs your LangGraph pipeline on every user message.
• Add real-time evaluation scores with pass/fail indicators below each answer.
• Test the full flow with straightforward and out-of-scope questions.

Create the Chainlit app with graph integration

Chainlit is purpose-built for conversational AI applications. It has a built-in Step API that displays intermediate agent reasoning steps without custom UI code.

Why Chainlit instead of Streamlit or Gradio?

Streamlit and Gradio are great for general data apps, but they require significant custom code to show Chain of Thought reasoning steps from a multi-agent pipeline.

Chainlit's Step API lets you nest intermediate outputs (retrieval, grading, rewriting) as collapsible steps inside the chat. This gives you full observability into your RAG pipeline with just a few lines of code.

• Create a new file called app.py in your project folder.
• Add the imports and the chat start handler by copying this code:

import os
import chainlit as cl
from dotenv import load_dotenv
from rag_graph import build_graph, RAGState
from evaluator import evaluate_response

load_dotenv()


@cl.on_chat_start
async def start():
    # Build the RAG graph once per chat session
    graph = build_graph()
    cl.user_session.set("graph", graph)
    await cl.Message(
        content=(
            "**Self-Correcting RAG with Hallucination Detection**\n\n"
            "Ask me anything about company policies (returns, HR, IT security, "
            "data privacy, remote work, expenses, procurement, incident response).\n\n"
            "I will show you my reasoning steps and evaluate my own answer for "
            "faithfulness and hallucination."
        )
    ).send()

What does this code do?

• The imports pull in your build_graph function, RAGState type, and evaluate_response from the files you already created.
• The @cl.on_chat_start decorator runs once when a user opens the chat. It compiles the LangGraph pipeline and stores it in the session so every message reuses the same graph.
• The welcome message tells the user what topics the system covers and that it will show reasoning steps.

You won't see a change yet. The next chunk adds the message handler that runs the pipeline.

Each time a user sends a message, the app needs to stream the question through the RAG graph and surface every agent step as a collapsible element in the chat.

• In app.py, add the message handler below the start() function by copying this code:

@cl.on_message
async def main(message: cl.Message):
    graph = cl.user_session.get("graph")
    user_question = message.content

    # Initialize the graph state with the user's question
    initial_state: RAGState = {
        "question": user_question,
        "documents": [],
        "generation": "",
        "retries": 0,
    }

    retrieved_docs = []
    final_generation = ""
    total_retries = 0

    # Stream through the graph and display each agent step
    async with cl.Step(name="Running RAG Pipeline") as pipeline_step:
        for event in graph.stream(initial_state):
            for node_name, output in event.items():
                if node_name == "retrieve":
                    docs = output.get("documents", [])
                    retrieved_docs = docs
                    async with cl.Step(name="\ud83d\udd0d Retrieved Documents") as step:
                        doc_text = "\n\n---\n\n".join(
                            [f"**Doc {i+1}**: {d[:150]}..." for i, d in enumerate(docs)]
                        )
                        step.output = doc_text

                elif node_name == "rewrite_question":
                    new_q = output.get("question", "")
                    total_retries = output.get("retries", 0)
                    async with cl.Step(name="\ud83d\udd04 Query Rewritten") as step:
                        step.output = f"Original query was not well-matched. Rewritten to:\n\n**{new_q}**"

                elif node_name == "generate_answer":
                    final_generation = output.get("generation", "")
                    async with cl.Step(name="\u270d\ufe0f Answer Generated") as step:
                        step.output = final_generation

        pipeline_step.output = f"Pipeline complete. Retries: {total_retries}"

    await cl.Message(content=f"**Answer**: {final_generation}").send()

What does this code do?

• graph.stream(initial_state) runs the graph and yields events as each node completes, instead of waiting for the entire pipeline to finish.
• Each node type (retrieve, rewrite_question, generate_answer) gets its own cl.Step element. These appear as collapsible sections in the chat UI so the user can inspect each reasoning step.
• The outer cl.Step(name="Running RAG Pipeline") wraps all the inner steps. It reports the total retry count once the pipeline finishes.
• retrieved_docs and final_generation are captured outside the loop so the evaluation gate (next chunk) can access them.

• Save app.py.

You won't run the app yet. The next chunk adds the evaluation scores that complete the UI.

Seeing import errors in your editor?

• Make sure your virtual environment is activated. Your editor may not pick up packages installed in the venv unless it is selected as the Python interpreter.
• Check that rag_graph.py and evaluator.py are in the same folder as app.py.

Help me fix import errors in my Chainlit app

Add real-time evaluation scores to the UI

Right now the app sends an answer, but the user has no way to know if that answer is trustworthy. You will add the DeepEval evaluation gate as a live quality check that runs after every response.

• In app.py, find this line inside the main() function:

    await cl.Message(content=f"**Answer**: {final_generation}").send()

• Add the evaluation display code directly below that line, still inside the main() function:

    # Send a loading message while evaluation runs
    eval_msg = cl.Message(content="\u23f3 Evaluating response for hallucination and faithfulness...")
    await eval_msg.send()

    try:
        # Run the DeepEval evaluation gate on the generated answer
        scores = evaluate_response(
            question=user_question,
            answer=final_generation,
            retrieval_context=retrieved_docs,
        )

        # Extract faithfulness scores and reasoning
        faith_score = scores["faithfulness"]["score"]
        faith_passed = "\u2705" if scores["faithfulness"]["passed"] else "\u274c"
        faith_reason = scores["faithfulness"]["reason"] or "No reason provided"

        # Extract hallucination scores and reasoning
        halluc_score = scores["hallucination"]["score"]
        halluc_passed = "\u2705" if scores["hallucination"]["passed"] else "\u274c"
        halluc_reason = scores["hallucination"]["reason"] or "No reason provided"

        # Format and display the evaluation results
        eval_text = (
            "**Evaluation Scores**\n\n"
            f"**Faithfulness**: {faith_score:.2f} {faith_passed}\n"
            f"_{faith_reason}_\n\n"
            f"**Hallucination**: {halluc_score:.2f} {halluc_passed}\n"
            f"_{halluc_reason}_"
        )
        eval_msg.content = eval_text
        await eval_msg.update()

    except Exception as e:
        eval_msg.content = f"\u26a0\ufe0f Evaluation failed: {str(e)}"
        await eval_msg.update()

What does this code do?

• A loading message appears immediately so the user knows evaluation is in progress. The Llama 3.3 70B judge model takes a few seconds to score each response.
• evaluate_response() runs both FaithfulnessMetric and HallucinationMetric from your evaluator.py file.
• Pass/fail indicators use emoji to make trust scores scannable at a glance. The judge's reasoning is displayed in italics so users understand why a score was given.
• The try/except block catches evaluation failures (like rate limits on the Groq judge) without crashing the entire chat.

Finally, add the entry point at the very bottom of the file, outside the main() function.

• Add this block at the end of app.py:

if __name__ == "__main__":
    from chainlit.cli import run_chainlit
    run_chainlit(__file__)

• Save app.py.

Indentation issues?

The evaluation code must be indented inside the main() function (one level of indentation). The if __name__ block must be at the top level (no indentation). Python will raise an IndentationError if either is at the wrong level.

Help me fix indentation in my Chainlit app.py

✔️ Awesome, I've got everything!

Great. Double check you have saved app.py before moving on.

ⓧ I'd like to double check the full code

import os
import chainlit as cl
from dotenv import load_dotenv
from rag_graph import build_graph, RAGState
from evaluator import evaluate_response

load_dotenv()


@cl.on_chat_start
async def start():
    graph = build_graph()
    cl.user_session.set("graph", graph)
    await cl.Message(
        content=(
            "**Self-Correcting RAG with Hallucination Detection**\n\n"
            "Ask me anything about company policies (returns, HR, IT security, "
            "data privacy, remote work, expenses, procurement, incident response).\n\n"
            "I will show you my reasoning steps and evaluate my own answer for "
            "faithfulness and hallucination."
        )
    ).send()


@cl.on_message
async def main(message: cl.Message):
    graph = cl.user_session.get("graph")
    user_question = message.content

    initial_state: RAGState = {
        "question": user_question,
        "documents": [],
        "generation": "",
        "retries": 0,
    }

    retrieved_docs = []
    final_generation = ""
    total_retries = 0

    async with cl.Step(name="Running RAG Pipeline") as pipeline_step:
        for event in graph.stream(initial_state):
            for node_name, output in event.items():
                if node_name == "retrieve":
                    docs = output.get("documents", [])
                    retrieved_docs = docs
                    async with cl.Step(name="\ud83d\udd0d Retrieved Documents") as step:
                        doc_text = "\n\n---\n\n".join(
                            [f"**Doc {i+1}**: {d[:150]}..." for i, d in enumerate(docs)]
                        )
                        step.output = doc_text

                elif node_name == "rewrite_question":
                    new_q = output.get("question", "")
                    total_retries = output.get("retries", 0)
                    async with cl.Step(name="\ud83d\udd04 Query Rewritten") as step:
                        step.output = f"Original query was not well-matched. Rewritten to:\n\n**{new_q}**"

                elif node_name == "generate_answer":
                    final_generation = output.get("generation", "")
                    async with cl.Step(name="\u270d\ufe0f Answer Generated") as step:
                        step.output = final_generation

        pipeline_step.output = f"Pipeline complete. Retries: {total_retries}"

    await cl.Message(content=f"**Answer**: {final_generation}").send()

    eval_msg = cl.Message(content="\u23f3 Evaluating response for hallucination and faithfulness...")
    await eval_msg.send()

    try:
        scores = evaluate_response(
            question=user_question,
            answer=final_generation,
            retrieval_context=retrieved_docs,
        )

        faith_score = scores["faithfulness"]["score"]
        faith_passed = "\u2705" if scores["faithfulness"]["passed"] else "\u274c"
        faith_reason = scores["faithfulness"]["reason"] or "No reason provided"

        halluc_score = scores["hallucination"]["score"]
        halluc_passed = "\u2705" if scores["hallucination"]["passed"] else "\u274c"
        halluc_reason = scores["hallucination"]["reason"] or "No reason provided"

        eval_text = (
            "**Evaluation Scores**\n\n"
            f"**Faithfulness**: {faith_score:.2f} {faith_passed}\n"
            f"_{faith_reason}_\n\n"
            f"**Hallucination**: {halluc_score:.2f} {halluc_passed}\n"
            f"_{halluc_reason}_"
        )
        eval_msg.content = eval_text
        await eval_msg.update()

    except Exception as e:
        eval_msg.content = f"\u26a0\ufe0f Evaluation failed: {str(e)}"
        await eval_msg.update()


if __name__ == "__main__":
    from chainlit.cli import run_chainlit
    run_chainlit(__file__)

Test the full flow

Time to see everything working together. You will launch the Chainlit server and test with two types of questions to exercise both the happy path and the corrective loop.

• Make sure your Qdrant Docker container is still running. If you stopped it, start it again.
• Make sure your virtual environment is activated.
• Start the Chainlit app by running this command:

chainlit run app.py

What should I see?

The terminal outputs a URL (usually http://localhost:8000). Open it in your browser. You should see a chat interface with the welcome message about company policies.

The first load may take 10-20 seconds while the embedding model downloads. Subsequent starts are faster.

• Type What is the return policy? in the chat box and press Enter.

You should see the pipeline steps expand: Retrieved Documents, then Answer Generated. The answer appears below, followed by evaluation scores after a few seconds. Faithfulness should score high for this question since it maps directly to the return policy document.

• Now type an ambiguous question like stuff about working from home equipment money and press Enter.

This time you may see a Query Rewritten step appear before the answer, meaning the critic flagged the initial retrieval as low relevance and triggered the corrective loop. The evaluation scores may also differ.

What about Groq rate limits?

Each question makes 5-8 Groq API calls across the generator (Llama 3.1 8B) and judge (Llama 3.3 70B). The free tier allows 30 requests per minute on each model. If you ask questions rapidly, you may hit a rate limit. Wait 30-60 seconds and try again.

App won't start or page is blank?

• If you see ModuleNotFoundError, make sure your virtual environment is activated and all packages from requirements.txt are installed.
• If the browser shows a blank page, check the terminal for Python errors. A syntax or indentation mistake in app.py will prevent the server from starting.
• If you see Connection refused errors in the chat, your Qdrant container is not running. Start it with docker run or your previous Docker command.

Help me debug my Chainlit app startup

Your self-correcting RAG system now has a full chat interface with live observability. Next up, you will containerize the entire stack with Docker Compose for one-command deployment.

Deploy with Docker Compose

Your self-correcting RAG pipeline works end-to-end on your local machine. Agents retrieve, grade, rewrite, and generate. The evaluation gate catches hallucinations before they reach the user.

Running the pipeline requires starting Qdrant separately, managing environment variables, and launching Chainlit manually. Docker Compose lets you define the entire multi-container stack in one file and launch everything with a single command.

In this step, get ready to:

• Create a Dockerfile and Docker Compose configuration for the full stack.
• Build and launch both services with one command.
• Verify data persistence and self-healing behavior in the containerized environment.

Create the Dockerfile and Docker Compose configuration

You need two configuration files to containerize the stack. A Dockerfile describes how to build the Python app image. A docker-compose.yml defines both services and wires them together.

• Create a new file called Dockerfile (no file extension) in your project folder, then paste in the following code:

FROM python:3.11-slim

WORKDIR /app

# Copy and install Python dependencies first (caches this layer)
COPY requirements.txt .
RUN pip install --no-cache-dir -r requirements.txt

# Copy all project source files into the container
COPY . .

# Attempt to pre-index documents (will fail during build since Qdrant isn't running yet)
RUN python indexer.py || true

EXPOSE 8000

# Start the Chainlit server on all interfaces
CMD ["chainlit", "run", "app.py", "--host", "0.0.0.0", "--port", "8000"]

What does this Dockerfile do?

• FROM python:3.11-slim starts from a lightweight Python 3.11 base image.
• COPY requirements.txt . is copied before the rest of the code so Docker can cache the dependency install layer. If your code changes but dependencies don't, Docker skips the slow pip install step.
• RUN python indexer.py || true attempts to index documents during the build. It will fail because Qdrant isn't running yet, but || true prevents the build from stopping. You'll run the indexer manually after the stack is up.
• CMD launches the Chainlit server on port 8000, binding to all network interfaces so Docker can forward traffic into the container.

• Save Dockerfile.

• Now create a file called docker-compose.yml in the same project folder, then paste in the following configuration:

services:
  # Vector database for document storage and retrieval
  qdrant:
    image: qdrant/qdrant:latest
    container_name: rag-qdrant
    restart: unless-stopped
    ports:
      - "6333:6333"
      - "6334:6334"
    volumes:
      - qdrant_data:/qdrant/storage
    environment:
      - QDRANT__SERVICE__GRPC_PORT=6334

  # Chainlit app with RAG pipeline and evaluation gate
  app:
    build: .
    container_name: rag-app
    restart: unless-stopped
    ports:
      - "8000:8000"
    environment:
      - GROQ_API_KEY=${GROQ_API_KEY}
      - QDRANT_HOST=qdrant
      - QDRANT_PORT=6333
    depends_on:
      - qdrant

# Named volume persists Qdrant data across container restarts
volumes:
  qdrant_data:

What does this configuration do?

• The qdrant service runs the Qdrant vector database with a named volume (qdrant_data) so your indexed documents survive container restarts.
• The app service builds from your Dockerfile and depends on qdrant, so Docker starts Qdrant first.
• GROQ_API_KEY=${GROQ_API_KEY} reads your API key from the .env file. The key never gets baked into the Docker image.
• QDRANT_HOST=qdrant overrides the localhost value from your .env file. Inside a Docker network, services talk to each other by service name, not localhost.

• Save docker-compose.yml.
• Confirm both files exist by running:

ls Dockerfile docker-compose.yml

You should see both file names printed in the output.

Don't see both files?

• Make sure Dockerfile has no file extension. It should not be Dockerfile.txt or Dockerfile.dockerfile.
• Make sure both files are in your project root folder, the same folder that contains app.py and requirements.txt.
• Help me find my Dockerfile and docker-compose.yml

✔️ Awesome, I've got everything!

Great. Double check you've saved both files before moving on.

ⓧ I'd like to double check the full code

FROM python:3.11-slim

WORKDIR /app

COPY requirements.txt .
RUN pip install --no-cache-dir -r requirements.txt

COPY . .

RUN python indexer.py || true

EXPOSE 8000

CMD ["chainlit", "run", "app.py", "--host", "0.0.0.0", "--port", "8000"]

services:
  qdrant:
    image: qdrant/qdrant:latest
    container_name: rag-qdrant
    restart: unless-stopped
    ports:
      - "6333:6333"
      - "6334:6334"
    volumes:
      - qdrant_data:/qdrant/storage
    environment:
      - QDRANT__SERVICE__GRPC_PORT=6334

  app:
    build: .
    container_name: rag-app
    restart: unless-stopped
    ports:
      - "8000:8000"
    environment:
      - GROQ_API_KEY=${GROQ_API_KEY}
      - QDRANT_HOST=qdrant
      - QDRANT_PORT=6333
    depends_on:
      - qdrant

volumes:
  qdrant_data:

Build and launch the stack

Before launching Docker Compose, you need to stop the standalone Qdrant container from earlier. Both would try to bind the same ports, causing a conflict.

• Stop the standalone Qdrant container by running:

docker stop $(docker ps -q --filter ancestor=qdrant/qdrant:latest)

What does this command do?

docker ps -q --filter ancestor=qdrant/qdrant:latest finds the container ID of any running Qdrant container. docker stop then shuts it down, freeing up ports 6333 and 6334 for Docker Compose.

Seeing "docker stop requires at least 1 argument"?

This means no Qdrant container is currently running. That's fine. You can safely continue to the next step.

• Build and launch the full stack by running:

docker compose up --build

What should I see?

Docker builds the app image first (this takes a few minutes the first time because it installs all Python dependencies and downloads the embedding model). Then it starts both containers.

You'll see logs from both rag-qdrant and rag-app interleaved in your terminal. Look for a line like Your app is available at http://localhost:8000 from the Chainlit server.

The Qdrant collection is empty right now because the indexer couldn't reach Qdrant during the Docker build. You need to index the documents into the running Qdrant container.

• Open a new terminal window.
• Run the indexer inside the app container by running:

docker exec rag-app python indexer.py

You should see output confirming 8 documents were indexed and a test query returned results.

Seeing "Error: No such container"?

• Make sure docker compose up --build is still running in your other terminal window. The containers must be up before you can exec into them.
• Check that the container name matches by running docker ps. You should see rag-app and rag-qdrant listed.
• Help me get my Docker containers running

• Open your browser and go to http://localhost:8000.

You should see the Chainlit chat interface with the welcome message about company policies.

• Type a question like What is the return policy? and press Enter.

You should see the full pipeline in action: retrieved documents, the generated answer, and then the faithfulness and hallucination evaluation scores. The entire flow runs inside Docker containers.

Page not loading at localhost:8000?

• Check the terminal running docker compose up for error messages. A common issue is a missing or invalid GROQ_API_KEY in your .env file.
• Make sure no other process is using port 8000. If something else is running on that port, stop it first.
• Wait 30 seconds after both containers start. The Chainlit server and embedding model need time to initialize.
• Help me debug my Docker Compose deployment

Verify persistence and self-healing

A production-grade deployment needs to survive restarts without losing data. The named volume in your Docker Compose file should preserve the Qdrant collection across stop/start cycles.

• In the terminal running Docker Compose, press Ctrl+C to stop both containers.
• Restart the stack by running:

docker compose up

Notice you don't need the --build flag this time. Docker reuses the cached image since nothing changed.

• Open http://localhost:8000 in your browser.
• Ask the same policy question as before (e.g. What is the return policy?).

You should get the same answer with evaluation scores. The indexed documents survived the restart because Qdrant's data is stored in the qdrant_data named volume.

• Now test the corrective loop. Type an intentionally vague query like stuff about computer rules and passwords and press Enter.

Watch the Chain of Thought steps. If the critic grades the initial retrieval as irrelevant, you'll see a query rewrite step before the answer is generated. This confirms the self-healing loop works end-to-end in the containerized environment.

Seeing the corrective loop in action

The vague query might get rewritten to something more specific like "IT security password policy requirements." Whether the rewrite triggers depends on how the critic grades the initial results. Try different levels of vagueness to see the threshold.

Qdrant returning empty results after restart?

• Make sure you used docker compose up (not docker compose down) to stop the stack. docker compose down -v deletes named volumes, which erases all indexed data.
• If the data is gone, re-index by running docker exec rag-app python indexer.py again.
• Help me fix Qdrant data persistence in Docker

Secret mission

Add Human-AI Judge Calibration

Your auto-judge scores every response for faithfulness and hallucination. But how do you know those scores are correct? In this extension, you will label a test set yourself, measure agreement between your judgment and the AI judge using Cohen's kappa, and tune the evaluation prompts to improve that agreement.

Before you go

Clean Up Your Resources

Decide whether to keep your resources running, pause them to come back later, or delete them entirely. This project runs entirely on free resources. Qdrant is self-hosted via Docker and the Groq API free tier has no charges, so there are no ongoing costs.

Resources you used:

• Qdrant Docker container running on port 6333 with a persistent named volume (qdrant_data).
• Chainlit app container running on port 8000 via Docker Compose.
• Groq API account (free tier). 14,400 requests/day on Llama 3.1 8B and 1,000 requests/day on Llama 3.3 70B.
• Python virtual environment and project files (indexer.py, rag_graph.py, evaluator.py, app.py).
• sentence-transformers model cache (~90MB, stored locally by HuggingFace).

Keep everything running

No action needed. Choose this if you are still actively building or experimenting with the pipeline.

• Your Docker Compose stack stays running in the background. Qdrant and the Chainlit app continue serving requests.
• Everything is free. Qdrant is self-hosted and the Groq API free tier has no charges.
• Your indexed documents persist in the qdrant_data volume.

Pause - I'll come back to this later

Stop the running containers to free up memory and CPU without losing any data.

• Stop the Docker Compose stack by running:

docker compose stop

• This stops the containers without removing them. Your Qdrant data persists in the named volume.
• When you are ready to come back, restart everything by running:

docker compose start

• Your Groq API key stays in your .env file. No action needed for the Groq account.

Delete - I don't want to use this again

Remove all project resources and start fresh.

• Stop and remove all containers and volumes by running:

docker compose down -v

• The -v flag removes the qdrant_data volume. You will lose your indexed documents, but you can re-run the indexer if you change your mind later.
• Delete your project folder from the Desktop by running:

cd ~/Desktop
rm -rf corrective-rag

• Your Groq account is free and does not need to be deleted. If you want to revoke your API key, visit console.groq.com and delete the key from your dashboard.
• To remove the cached sentence-transformers model, delete the ~/.cache/huggingface directory. This frees up about 90MB of disk space.

Mission accomplished

Nice Work!

You just built a production-grade, self-healing RAG system with automated hallucination detection. Every response your system generates is retrieved, graded, self-corrected if needed, and evaluated by a stronger judge model before it reaches the user.

Here's what you learned:

• Built a multi-agent corrective RAG pipeline in LangGraph with a retriever, critic, rewriter, and generator that automatically rewrites queries when retrieval quality is poor.
• Wired up DeepEval's FaithfulnessMetric and HallucinationMetric with a Groq-powered LLM judge (Llama 3.3 70B) to score every response before it reaches the user.
• Containerized the entire stack with Docker Compose for one-command deployment, packaging Qdrant, a Chainlit chat UI with real-time evaluation scores, and the full RAG pipeline into a portable stack.
• Secret Mission: Calibrated your auto-judge against human judgment using Cohen's kappa and tuned evaluation prompts to improve agreement between your labels and the LLM judge.

Ready to quiz yourself?