RAG-Based Testing Series — Part 5: Building a RAG Test Framework from Scratch

rag_test_framework/
├── config/
│   └── settings.py          ← all configuration in one place
├── core/
│   ├── retriever.py         ← retrieval logic + scoring
│   ├── evaluator.py         ← RAGAS evaluation wrapper
│   └── rag_pipeline.py      ← end-to-end RAG call
├── tests/
│   ├── conftest.py          ← shared pytest fixtures
│   ├── test_retrieval.py    ← retrieval quality tests
│   ├── test_faithfulness.py ← faithfulness + hallucination tests
│   └── test_edge_cases.py   ← edge case tests
├── data/
│   └── test_cases.json      ← your ground truth dataset
├── reports/
│   └── (auto-generated)     ← test run reports saved here
├── run_tests.py             ← single entry point to run everything
└── requirements.txt

python run_tests.py

pytest tests/ -v --tb=short

pip install ragas
pip install openai
pip install chromadb
pip install datasets
pip install langchain-openai
pip install pytest
pip install pytest-json-report

ragas>=0.1.0
openai>=1.0.0
chromadb>=0.4.0
datasets>=2.0.0
langchain-openai>=0.0.1
pytest>=7.0.0
pytest-json-report>=1.5.0

# config/settings.py

import os

# ── API Keys ──────────────────────────────────────────────
OPENAI_API_KEY = os.getenv("OPENAI_API_KEY", "your-openai-api-key")

# ── Models ────────────────────────────────────────────────
EMBEDDING_MODEL = "text-embedding-ada-002"
EVALUATION_LLM  = "gpt-4o-mini"
RAG_LLM         = "gpt-4o-mini"

# ── Retrieval Settings ────────────────────────────────────
DEFAULT_TOP_K             = 3          # how many docs to retrieve
SIMILARITY_THRESHOLD      = 0.8        # max L2 distance to consider relevant
                                        # tune this for your embedding model

# ── Quality Thresholds ────────────────────────────────────
# These are the minimum acceptable scores for your system.
# Adjust based on your risk tolerance.
MIN_PRECISION_AT_K        = 0.6        # >= 60% of retrieved docs must be relevant
MIN_RECALL_AT_K           = 0.7        # >= 70% of relevant docs must be retrieved
MIN_MRR                   = 0.7        # first relevant doc should rank in top 2 on average
MIN_FAITHFULNESS          = 0.8        # >= 80% of answer claims must be grounded
CRITICAL_FAITHFULNESS     = 0.3        # below this = outright fabrication, always fail

# ── Reporting ─────────────────────────────────────────────
REPORTS_DIR               = "reports"
REPORT_FILENAME           = "rag_test_report.json"

# core/retriever.py

import chromadb
from chromadb.utils import embedding_functions
from config.settings import (
    OPENAI_API_KEY,
    EMBEDDING_MODEL,
    DEFAULT_TOP_K,
    SIMILARITY_THRESHOLD
)


def build_collection(collection_name: str, documents: list[str], doc_ids: list[str]):
    """
    Build a ChromaDB collection from a list of documents.
    Call this once during test setup.
    """
    client = chromadb.Client()
    embedding_fn = embedding_functions.OpenAIEmbeddingFunction(
        api_key=OPENAI_API_KEY,
        model_name=EMBEDDING_MODEL
    )

    # Delete collection if it already exists (clean slate for each test run)
    try:
        client.delete_collection(collection_name)
    except Exception:
        pass

    collection = client.create_collection(
        name=collection_name,
        embedding_function=embedding_fn
    )

    collection.add(documents=documents, ids=doc_ids)
    return collection


def retrieve(collection, query: str, n_results: int = DEFAULT_TOP_K) -> dict:
    """
    Retrieve top N documents for a query.
    Returns documents, IDs, and distance scores.
    """
    results = collection.query(
        query_texts=[query],
        n_results=n_results,
        include=["documents", "distances", "ids"]
    )
    return {
        "documents": results["documents"][0],
        "ids":       results["ids"][0],
        "distances": results["distances"][0]
    }


def filter_by_threshold(retrieval_result: dict, threshold: float = SIMILARITY_THRESHOLD) -> dict:
    """
    Filter out documents whose distance exceeds the similarity threshold.
    Returns only the documents considered relevant.
    """
    filtered = [
        (doc, doc_id, dist)
        for doc, doc_id, dist in zip(
            retrieval_result["documents"],
            retrieval_result["ids"],
            retrieval_result["distances"]
        )
        if dist < threshold
    ]

    if not filtered:
        return {"documents": [], "ids": [], "distances": []}

    docs, ids, dists = zip(*filtered)
    return {"documents": list(docs), "ids": list(ids), "distances": list(dists)}


def calculate_precision_at_k(retrieved_ids: list, relevant_ids: list, k: int) -> float:
    retrieved_at_k = retrieved_ids[:k]
    hits = [doc_id for doc_id in retrieved_at_k if doc_id in relevant_ids]
    return len(hits) / k if k > 0 else 0.0


def calculate_recall_at_k(retrieved_ids: list, relevant_ids: list, k: int) -> float:
    retrieved_at_k = retrieved_ids[:k]
    hits = [doc_id for doc_id in retrieved_at_k if doc_id in relevant_ids]
    return len(hits) / len(relevant_ids) if relevant_ids else 0.0


def calculate_mrr(collection, test_cases: list, k: int = 5) -> float:
    reciprocal_ranks = []
    for test in test_cases:
        result = retrieve(collection, test["query"], n_results=k)
        rr = 0.0
        for rank, doc_id in enumerate(result["ids"], start=1):
            if doc_id in test["relevant_doc_ids"]:
                rr = 1.0 / rank
                break
        reciprocal_ranks.append(rr)
    return round(sum(reciprocal_ranks) / len(reciprocal_ranks), 4) if reciprocal_ranks else 0.0

# core/evaluator.py

from datasets import Dataset
from ragas import evaluate
from ragas.metrics import faithfulness, answer_relevancy
from langchain_openai import ChatOpenAI, OpenAIEmbeddings
from config.settings import OPENAI_API_KEY, EVALUATION_LLM


def build_evaluator():
    """
    Build and return the LLM + embeddings used by RAGAS.
    Call once and reuse — avoids re-initialising on every test.
    """
    llm = ChatOpenAI(
        model=EVALUATION_LLM,
        openai_api_key=OPENAI_API_KEY
    )
    embeddings = OpenAIEmbeddings(
        openai_api_key=OPENAI_API_KEY
    )
    return llm, embeddings


def evaluate_faithfulness(test_data: list, llm, embeddings) -> list[dict]:
    """
    Run RAGAS faithfulness + answer_relevancy evaluation on a list of test cases.

    Each test case must have:
        - question  (str)
        - answer    (str)
        - contexts  (list of str)

    Returns a list of dicts with per-case scores added.
    """
    dataset = Dataset.from_list(test_data)

    results = evaluate(
        dataset=dataset,
        metrics=[faithfulness, answer_relevancy],
        llm=llm,
        embeddings=embeddings
    )

    df = results.to_pandas()

    scored = []
    for i, row in df.iterrows():
        scored.append({
            **test_data[i],
            "faithfulness":     round(float(row["faithfulness"]), 4),
            "answer_relevancy": round(float(row["answer_relevancy"]), 4)
        })

    return scored

# core/rag_pipeline.py

from openai import OpenAI
from core.retriever import retrieve, filter_by_threshold
from config.settings import OPENAI_API_KEY, RAG_LLM, DEFAULT_TOP_K

_openai_client = OpenAI(api_key=OPENAI_API_KEY)

SYSTEM_PROMPT = """You are a helpful customer support assistant.
Answer questions using ONLY the information provided in the context below.
If the context does not contain enough information to answer the question,
say so clearly — do not make up information.
If the question is outside the scope of the provided context, say so politely."""


def run_rag(collection, question: str, n_results: int = DEFAULT_TOP_K) -> dict:
    """
    Run a full RAG call: retrieve context, then generate an answer.

    Returns:
        question   — the original question
        contexts   — list of retrieved document strings (after threshold filter)
        answer     — the LLM's generated answer
        raw_distances — distances for all retrieved docs (for debugging)
    """
    raw_result = retrieve(collection, question, n_results=n_results)
    filtered   = filter_by_threshold(raw_result)

    contexts = filtered["documents"]

    if not contexts:
        # No relevant documents found — return a structured "I don't know"
        return {
            "question":      question,
            "contexts":      [],
            "answer":        "I don't have relevant information to answer this question.",
            "raw_distances": raw_result["distances"]
        }

    context_block = "\n\n".join(contexts)
    prompt = f"{SYSTEM_PROMPT}\n\nContext:\n{context_block}\n\nQuestion: {question}\n\nAnswer:"

    response = _openai_client.chat.completions.create(
        model=RAG_LLM,
        messages=[{"role": "user", "content": prompt}]
    )

    return {
        "question":      question,
        "contexts":      contexts,
        "answer":        response.choices[0].message.content,
        "raw_distances": raw_result["distances"]
    }

// data/test_cases.json
{
  "knowledge_base": [
    {
      "id":   "doc1",
      "text": "Premium subscribers are eligible for a full refund within 30 days of purchase. Requests must be submitted via the support portal."
    },
    {
      "id":   "doc2",
      "text": "Standard subscribers can cancel their subscription at any time. Cancellation takes effect at the end of the billing period."
    },
    {
      "id":   "doc3",
      "text": "Shipping for all orders is processed within 2-3 business days. Express shipping is available at checkout."
    },
    {
      "id":   "doc4",
      "text": "To reset your password, click the Forgot Password link on the login page. A reset link will be sent to your registered email."
    },
    {
      "id":   "doc5",
      "text": "Premium subscribers get access to priority customer support with a 2-hour response time guarantee."
    }
  ],

  "retrieval_test_cases": [
    {
      "query":            "What is the refund policy for premium subscribers?",
      "relevant_doc_ids": ["doc1"]
    },
    {
      "query":            "How do I cancel my subscription?",
      "relevant_doc_ids": ["doc2"]
    },
    {
      "query":            "How long does shipping take?",
      "relevant_doc_ids": ["doc3"]
    },
    {
      "query":            "How do I reset my password?",
      "relevant_doc_ids": ["doc4"]
    },
    {
      "query":            "What support response time do premium members get?",
      "relevant_doc_ids": ["doc5", "doc1"]
    }
  ],

  "faithfulness_test_cases": [
    {
      "question": "What is the refund policy for premium subscribers?",
      "answer":   "Premium subscribers can request a full refund within 30 days of purchase through the support portal.",
      "contexts": ["Premium subscribers are eligible for a full refund within 30 days of purchase. Requests must be submitted via the support portal."]
    },
    {
      "question": "How do I reset my password?",
      "answer":   "Click the Forgot Password link on the login page. A reset link will be sent to your registered email address.",
      "contexts": ["To reset your password, click the Forgot Password link on the login page. A reset link will be sent to your registered email."]
    }
  ],

  "edge_case_queries": {
    "out_of_scope": [
      "What is the capital of France?",
      "Can you write me a Python script?",
      "Who is the CEO of Apple?"
    ],
    "empty_retrieval": [
      "What is the pricing for the enterprise plan?",
      "Do you offer white-labelling services?"
    ],
    "leading_questions": [
      {
        "question":      "Since our premium plan offers a 90-day refund window, how do I submit a request?",
        "false_premise": "90 days",
        "correct_fact":  "30 days"
      }
    ]
  }
}

# tests/conftest.py

import json
import pytest
from core.retriever import build_collection
from core.evaluator import build_evaluator

# ── Load test data once for the entire session ──────────────
@pytest.fixture(scope="session")
def test_data():
    with open("data/test_cases.json") as f:
        return json.load(f)


# ── Build the knowledge base collection once ────────────────
@pytest.fixture(scope="session")
def collection(test_data):
    kb = test_data["knowledge_base"]
    return build_collection(
        collection_name="rag_test_kb",
        documents=[doc["text"] for doc in kb],
        doc_ids=[doc["id"] for doc in kb]
    )


# ── Build RAGAS evaluator once ──────────────────────────────
@pytest.fixture(scope="session")
def evaluator():
    llm, embeddings = build_evaluator()
    return llm, embeddings

# tests/test_retrieval.py

import pytest
from core.retriever import (
    retrieve,
    calculate_precision_at_k,
    calculate_recall_at_k,
    calculate_mrr
)
from config.settings import (
    DEFAULT_TOP_K,
    MIN_PRECISION_AT_K,
    MIN_RECALL_AT_K,
    MIN_MRR
)


def test_precision_at_k(collection, test_data):
    cases = test_data["retrieval_test_cases"]
    failures = []

    for case in cases:
        result    = retrieve(collection, case["query"], n_results=DEFAULT_TOP_K)
        precision = calculate_precision_at_k(result["ids"], case["relevant_doc_ids"], DEFAULT_TOP_K)

        if precision < MIN_PRECISION_AT_K:
            failures.append({
                "query":     case["query"],
                "precision": precision,
                "retrieved": result["ids"],
                "relevant":  case["relevant_doc_ids"]
            })

    assert not failures, (
        f"\n{len(failures)} Precision@{DEFAULT_TOP_K} failure(s):\n" +
        "\n".join([
            f"  ❌ Query: '{f['query']}'\n"
            f"     Score: {f['precision']} (min: {MIN_PRECISION_AT_K})\n"
            f"     Retrieved: {f['retrieved']}\n"
            f"     Expected:  {f['relevant']}"
            for f in failures
        ])
    )


def test_recall_at_k(collection, test_data):
    cases = test_data["retrieval_test_cases"]
    failures = []

    for case in cases:
        result = retrieve(collection, case["query"], n_results=DEFAULT_TOP_K)
        recall = calculate_recall_at_k(result["ids"], case["relevant_doc_ids"], DEFAULT_TOP_K)

        if recall < MIN_RECALL_AT_K:
            failures.append({
                "query":  case["query"],
                "recall": recall
            })

    assert not failures, (
        f"\n{len(failures)} Recall@{DEFAULT_TOP_K} failure(s):\n" +
        "\n".join([
            f"  ❌ Query: '{f['query']}'\n"
            f"     Score: {f['recall']} (min: {MIN_RECALL_AT_K})"
            for f in failures
        ])
    )


def test_mrr(collection, test_data):
    cases = test_data["retrieval_test_cases"]
    mrr   = calculate_mrr(collection, cases)

    assert mrr >= MIN_MRR, (
        f"\nMRR too low: {mrr} (min: {MIN_MRR})\n"
        f"Relevant documents are not ranking high enough in retrieval results."
    )

# tests/test_faithfulness.py

import pytest
from core.evaluator import evaluate_faithfulness
from config.settings import MIN_FAITHFULNESS, CRITICAL_FAITHFULNESS


def test_faithfulness_above_threshold(evaluator, test_data):
    llm, embeddings = evaluator
    cases   = test_data["faithfulness_test_cases"]
    scored  = evaluate_faithfulness(cases, llm, embeddings)
    failures = [s for s in scored if s["faithfulness"] < MIN_FAITHFULNESS]

    assert not failures, (
        f"\n{len(failures)} faithfulness failure(s):\n" +
        "\n".join([
            f"  ❌ Question: '{f['question']}'\n"
            f"     Score: {f['faithfulness']} (min: {MIN_FAITHFULNESS})\n"
            f"     Answer: {f['answer'][:100]}..."
            for f in failures
        ])
    )


def test_no_critical_hallucinations(evaluator, test_data):
    """
    Any answer below CRITICAL_FAITHFULNESS is an outright fabrication.
    This test should never be allowed to pass in any environment.
    """
    llm, embeddings = evaluator
    cases   = test_data["faithfulness_test_cases"]
    scored  = evaluate_faithfulness(cases, llm, embeddings)
    critical = [s for s in scored if s["faithfulness"] < CRITICAL_FAITHFULNESS]

    assert not critical, (
        f"\n🚨 {len(critical)} CRITICAL hallucination(s) detected:\n" +
        "\n".join([
            f"  🔴 Question: '{f['question']}'\n"
            f"     Score: {f['faithfulness']} (critical threshold: {CRITICAL_FAITHFULNESS})\n"
            f"     Answer: {f['answer'][:100]}..."
            for f in critical
        ])
    )

# tests/test_edge_cases.py

import pytest
from core.retriever import retrieve, filter_by_threshold
from core.rag_pipeline import run_rag
from config.settings import SIMILARITY_THRESHOLD


def test_empty_retrieval_detected(collection, test_data):
    """
    Queries with no relevant documents should return zero docs
    after threshold filtering.
    """
    out_of_kb_queries = test_data["edge_case_queries"]["empty_retrieval"]

    for query in out_of_kb_queries:
        raw      = retrieve(collection, query)
        filtered = filter_by_threshold(raw, threshold=SIMILARITY_THRESHOLD)

        assert len(filtered["documents"]) == 0, (
            f"\n❌ Expected no relevant docs for: '{query}'\n"
            f"   Got {len(filtered['documents'])} doc(s) below threshold.\n"
            f"   Distances: {raw['distances']}"
        )


def test_rag_response_on_empty_retrieval(collection, test_data):
    """
    When no relevant documents exist, the pipeline should return
    an uncertainty response — not a fabricated answer.
    """
    uncertainty_indicators = [
        "don't have", "no information", "unable to find",
        "not available", "contact", "cannot find"
    ]

    out_of_kb_queries = test_data["edge_case_queries"]["empty_retrieval"]

    for query in out_of_kb_queries:
        result       = run_rag(collection, query)
        answer_lower = result["answer"].lower()
        is_uncertain = any(ind in answer_lower for ind in uncertainty_indicators)

        assert is_uncertain, (
            f"\n❌ RAG pipeline did not express uncertainty for: '{query}'\n"
            f"   Answer: {result['answer']}\n"
            f"   Expected one of: {uncertainty_indicators}"
        )


def test_out_of_scope_handling(collection, test_data):
    """
    Out-of-scope queries should produce a decline or redirect — not a fabricated answer.
    """
    decline_indicators = [
        "outside", "scope", "unable to help", "don't have information",
        "not able to assist", "please contact", "beyond", "not within"
    ]

    for query in test_data["edge_case_queries"]["out_of_scope"]:
        result       = run_rag(collection, query)
        answer_lower = result["answer"].lower()
        declined     = any(ind in answer_lower for ind in decline_indicators)

        assert declined, (
            f"\n❌ System did not decline out-of-scope query: '{query}'\n"
            f"   Answer: {result['answer']}"
        )


def test_leading_question_correction(collection, evaluator, test_data):
    """
    When a query contains a false premise, the generated answer should
    contradict the false premise — not accept it.
    """
    from core.evaluator import evaluate_faithfulness
    from config.settings import MIN_FAITHFULNESS

    llm, embeddings = evaluator
    leading_cases   = test_data["edge_case_queries"]["leading_questions"]

    for case in leading_cases:
        result = run_rag(collection, case["question"])

        # The answer should NOT contain the false premise
        assert case["false_premise"] not in result["answer"], (
            f"\n❌ LLM accepted false premise in leading question.\n"
            f"   Question:      {case['question']}\n"
            f"   False premise: {case['false_premise']}\n"
            f"   Answer:        {result['answer']}"
        )

        # The answer SHOULD contain the correct fact
        assert case["correct_fact"] in result["answer"], (
            f"\n❌ LLM did not correct the false premise.\n"
            f"   Question:     {case['question']}\n"
            f"   Correct fact: {case['correct_fact']}\n"
            f"   Answer:       {result['answer']}"
        )

# run_tests.py

import os
import json
import subprocess
from datetime import datetime
from config.settings import REPORTS_DIR, REPORT_FILENAME

def run():
    os.makedirs(REPORTS_DIR, exist_ok=True)

    timestamp   = datetime.now().strftime("%Y%m%d_%H%M%S")
    report_path = os.path.join(REPORTS_DIR, f"{timestamp}_{REPORT_FILENAME}")

    print("=" * 65)
    print("  RAG Test Framework — Full Suite")
    print(f"  Started: {datetime.now().strftime('%Y-%m-%d %H:%M:%S')}")
    print("=" * 65)

    result = subprocess.run(
        [
            "pytest", "tests/",
            "-v",
            "--tb=short",
            f"--json-report",
            f"--json-report-file={report_path}"
        ],
        capture_output=False  # stream output to terminal in real time
    )

    print("\n" + "=" * 65)

    # Parse and print summary from the JSON report
    if os.path.exists(report_path):
        with open(report_path) as f:
            report = json.load(f)

        summary  = report.get("summary", {})
        passed   = summary.get("passed", 0)
        failed   = summary.get("failed", 0)
        total    = summary.get("total", 0)
        duration = round(report.get("duration", 0), 2)

        print(f"  Results : {passed}/{total} passed | {failed} failed")
        print(f"  Duration: {duration}s")
        print(f"  Report  : {report_path}")

        if failed > 0:
            print("\n  ❌ Failed tests:")
            for test in report.get("tests", []):
                if test["outcome"] == "failed":
                    print(f"     • {test['nodeid']}")

        print("=" * 65)

    return result.returncode


if __name__ == "__main__":
    exit(run())

# Run the full suite
python run_tests.py

# Run only retrieval tests
pytest tests/test_retrieval.py -v

# Run only faithfulness tests
pytest tests/test_faithfulness.py -v

# Run only edge case tests
pytest tests/test_edge_cases.py -v

# Run with full failure output
pytest tests/ -v --tb=long

=================================================================
  RAG Test Framework — Full Suite
  Started: 2025-06-01 14:32:01
=================================================================

tests/test_retrieval.py::test_precision_at_k    PASSED
tests/test_retrieval.py::test_recall_at_k       PASSED
tests/test_retrieval.py::test_mrr               PASSED
tests/test_faithfulness.py::test_faithfulness_above_threshold  PASSED
tests/test_faithfulness.py::test_no_critical_hallucinations    PASSED
tests/test_edge_cases.py::test_empty_retrieval_detected        PASSED
tests/test_edge_cases.py::test_rag_response_on_empty_retrieval PASSED
tests/test_edge_cases.py::test_out_of_scope_handling           PASSED
tests/test_edge_cases.py::test_leading_question_correction     PASSED

=================================================================
  Results : 9/9 passed | 0 failed
  Duration: 42.3s
  Report  : reports/20250601_143201_rag_test_report.json
=================================================================

python run_tests.py

Layer 1 — RETRIEVAL QUALITY ✅
          tests/test_retrieval.py
          → Precision@K, Recall@K, MRR

Layer 2 — FAITHFULNESS & HALLUCINATION ✅
          tests/test_faithfulness.py
          → Faithfulness score, Answer Relevancy score

Layer 3 — EDGE CASES ✅
          tests/test_edge_cases.py
          → Empty retrieval, out-of-scope, leading questions

Layer 4 — FULL FRAMEWORK ✅  ← You are here
          All layers combined, one command to run

Layer 5 — CI/CD AUTOMATION ← Up next (Part 6)
          Running automatically on every knowledge base change

Part 1 — What Is RAG & Why It Needs Different Testing       ✅ Done
Part 2 — Testing Retrieval Quality: Are You Fetching Right? ✅ Done
Part 3 — Faithfulness & Hallucination Detection             ✅ Done
Part 4 — Edge Cases: What Breaks RAG & How to Catch It      ✅ Done
Part 5 — Building a RAG Test Framework from Scratch         ← You are here
Part 6 — Automating RAG Quality Checks in CI/CD             ← Up next