Module 8: Capstone RAG Project with ChromaDB

Capsule 05: Testing and Evaluation

Capsule description

A RAG without evaluation can look correct in development and fail spectacularly in production. Users don't forgive hallucinated answers, 10-second latencies, or sporadic errors without diagnosis.

In this capsule you'll define and run a complete testing strategy: unit tests for ChromaDB operations, integration tests for the API endpoints, performance tests for latency and throughput, and accuracy validation with a golden set. You'll use pytest with fixtures and parametrization to keep tests maintainable and reusable.

At the end you'll have objective evidence that your system meets production-ready thresholds: p95 latency < 2s, throughput > 20 QPS, accuracy > 90%.

Why Testing in RAG is Critical

The risk of not testing

Typical scenario:
├── Ingestion works
├── Retrieval returns documents
├── Generation responds with something coherent
└── But is it correct? Is it fast? Does it scale?

Without tests: you don't know until the user complains.

RAG combines several components with subtle failures: chunking can cut off critical context, embeddings can degrade with new data, the LLM can make things up when retrieval fails. Tests give you confidence to deploy and evolve.

Testing pyramid for RAG

                    ▲
                   / \
                  / E2E \         Few, slow: full flow
                 /───────\
                /  Integ  \       More: API + ChromaDB + LLM mocks
               /───────────\
              /    Unit     \     Many, fast: chunking, filters, utilities
             /───────────────\

Unit: pure functions (chunking, filters, formatting)
Integration: real /ask flow with ChromaDB (real embeddings or mock)
E2E/Performance: latency, throughput, accuracy with a golden set

Recommended Test Types

Unit

Goal: validate isolated logic without external dependencies.

What to test	Example
Chunking	Long text → chunks of the expected size with overlap
Filters	Function that applies `where` over metadata
Utilities	ID normalization, source formatting
Input validation	Maximum question length, forbidden characters

Integration

Goal: validate the real flow between components.

What to test	Example
`/ingest`	Documents are inserted and are queryable
`/search`	Query returns results with scores and metadata
`/ask`	Question → retrieval → generation → answer with sources
`/health`	Responds OK when ChromaDB is available

Performance

Goal: ensure latency and throughput within thresholds.

Metric	Suggested threshold	How to measure
p95 latency `/ask`	< 2s	`locust` or `pytest-benchmark`
Throughput	> 20 QPS	sustained concurrent requests
p95 latency `/search`	< 500ms	retrieval only, no LLM

Quality (Accuracy)

Goal: validate that answers are correct and grounded.

What to measure	Method
Keyword match	The answer contains expected terms
Relevance	Retrieved documents are relevant to the question
Hallucination	The answer doesn't make things up when there is no evidence

Project Structure for Tests

project/
├── app/
│   ├── api/
│   ├── retrieval/
│   ├── generation/
│   └── ingestion/
├── tests/
│   ├── conftest.py          # Shared fixtures
│   ├── unit/
│   │   ├── test_chunking.py
│   │   ├── test_filters.py
│   │   └── test_utils.py
│   ├── integration/
│   │   ├── test_ingest.py
│   │   ├── test_search.py
│   │   └── test_ask.py
│   ├── performance/
│   │   └── test_latency_throughput.py
│   └── evaluation/
│       ├── golden_set.json
│       └── test_accuracy.py
├── pytest.ini
└── requirements-dev.txt

pytest Configuration

pytest.ini

[pytest]
testpaths = tests
python_files = test_*.py
python_functions = test_*
addopts = -v --tb=short -x
markers =
    unit: Unit tests (fast)
    integration: Integration tests (need ChromaDB)
    performance: Performance tests (slow)
    accuracy: Accuracy evaluation (needs LLM or mock)

requirements-dev.txt

pytest>=7.4.0
pytest-asyncio>=0.21.0
pytest-cov>=4.1.0
pytest-benchmark>=4.0.0
httpx>=0.24.0
chromadb>=0.4.0

Reusable Fixtures (conftest.py)

# tests/conftest.py
import pytest
import chromadb
from chromadb.config import Settings
import tempfile
import os

# ========== ChromaDB ==========

@pytest.fixture(scope="session")
def chroma_client():
    """Persistent ChromaDB client for session tests."""
    path = tempfile.mkdtemp(prefix="chroma_test_")
    client = chromadb.PersistentClient(path=path)
    yield client
    # Optional cleanup: remove directory

@pytest.fixture
def collection(chroma_client):
    """Clean collection per test for isolation."""
    col_name = "test_collection"
    try:
        chroma_client.delete_collection(col_name)
    except Exception:
        pass
    return chroma_client.get_or_create_collection(
        name=col_name,
        metadata={"hnsw:space": "cosine", "hnsw:M": 16}
    )

# ========== Golden Set ==========

@pytest.fixture
def golden_set():
    """Set of questions with expected keywords for accuracy."""
    return [
        {"question": "What is a vector database?", "expected_keywords": ["vector", "embedding", "search"]},
        {"question": "How does HNSW work?", "expected_keywords": ["graph", "approximate", "neighbor"]},
        {"question": "When to use ChromaDB?", "expected_keywords": ["local", "development", "RAG"]},
        # ... 20+ items
    ]

# ========== API Client ==========

@pytest.fixture
def api_client():
    """HTTP client for integration tests against the API."""
    import httpx
    base_url = os.getenv("API_BASE_URL", "http://localhost:8000")
    with httpx.Client(base_url=base_url, timeout=30.0) as client:
        yield client

Unit Tests: ChromaDB Operations

# tests/unit/test_chromadb_operations.py
import pytest
from app.ingestion.chunking import chunk_text

class TestChunking:
    """Unit tests for the chunking logic."""

    @pytest.mark.parametrize("chunk_size,overlap,expected_count", [
        (100, 0, 10),
        (100, 20, 13),
        (512, 64, 3),
    ])
    def test_chunk_count(self, chunk_size, overlap, expected_count):
        text = " ".join(["word"] * 500)
        chunks = chunk_text(text, chunk_size=chunk_size, overlap=overlap)
        assert len(chunks) >= expected_count

    def test_chunk_overlap_preserves_context(self):
        text = "This is a text with important context in the middle."
        chunks = chunk_text(text, chunk_size=20, overlap=10)
        # The overlap must ensure "important" is not lost between chunks
        full_joined = " ".join(chunks)
        assert "important" in full_joined

    def test_empty_text_returns_empty_list(self):
        assert chunk_text("") == []
        assert chunk_text("   ") == []

# tests/unit/test_filters.py
import pytest
from app.retrieval.filters import build_where_clause

class TestFilters:
    def test_build_where_empty(self):
        assert build_where_clause({}) is None

    def test_build_where_single(self):
        where = build_where_clause({"category": "science"})
        assert where == {"category": "science"}

    def test_build_where_multiple(self):
        where = build_where_clause({"category": "tech", "language": "en"})
        assert where == {"$and": [{"category": "tech"}, {"language": "en"}]}

Integration Tests: API Endpoints

# tests/integration/test_api_endpoints.py
import pytest
from httpx import AsyncClient
from app.main import app

@pytest.mark.integration
@pytest.mark.asyncio
class TestHealthEndpoint:
    async def test_health_returns_200(self):
        async with AsyncClient(app=app, base_url="http://test") as client:
            response = await client.get("/health")
        assert response.status_code == 200
        data = response.json()
        assert data.get("status") == "ok"
        assert "chromadb" in data or "version" in data

@pytest.mark.integration
@pytest.mark.asyncio
class TestSearchEndpoint:
    async def test_search_returns_documents(self):
        async with AsyncClient(app=app, base_url="http://test") as client:
            response = await client.get(
                "/search",
                params={"q": "vector database", "top_k": 5}
            )
        assert response.status_code == 200
        data = response.json()
        assert "documents" in data or "results" in data
        results = data.get("documents", data.get("results", []))
        assert len(results) <= 5

    async def test_search_empty_query_returns_400(self):
        async with AsyncClient(app=app, base_url="http://test") as client:
            response = await client.get("/search", params={"q": ""})
        assert response.status_code in [400, 422]

@pytest.mark.integration
@pytest.mark.asyncio
class TestAskEndpoint:
    async def test_ask_returns_answer_and_sources(self):
        async with AsyncClient(app=app, base_url="http://test") as client:
            response = await client.post(
                "/ask",
                json={"question": "What is a vector database?"}
            )
        assert response.status_code == 200
        data = response.json()
        assert "answer" in data
        assert "sources" in data
        assert isinstance(data["sources"], list)

    async def test_ask_no_evidence_returns_explicit_message(self):
        async with AsyncClient(app=app, base_url="http://test") as client:
            response = await client.post(
                "/ask",
                json={"question": "xyznonexistent123456"}
            )
        assert response.status_code == 200
        # Must indicate that there is not enough evidence
        data = response.json()
        assert "evidence" in data.get("answer", "").lower() or "not found" in data.get("answer", "").lower()

Performance Tests

# tests/performance/test_latency_throughput.py
import pytest
import time
import statistics
import httpx

@pytest.mark.performance
class TestLatencyThroughput:
    """Performance tests: p95 latency < 2s, throughput > 20 QPS."""

    def test_ask_p95_latency_under_2s(self):
        base_url = "http://localhost:8000"
        latencies = []
        for _ in range(50):
            start = time.perf_counter()
            resp = httpx.post(f"{base_url}/ask", json={"question": "What is RAG?"}, timeout=10.0)
            assert resp.status_code == 200
            latencies.append(time.perf_counter() - start)
        p95 = sorted(latencies)[int(0.95 * len(latencies))]
        assert p95 < 2.0, f"p95 latency {p95:.2f}s exceeds 2s threshold"

    def test_search_throughput_over_20_qps(self):
        base_url = "http://localhost:8000"
        duration = 5  # seconds
        end = time.time() + duration
        count = 0
        while time.time() < end:
            resp = httpx.get(f"{base_url}/search", params={"q": "vector", "top_k": 5}, timeout=5.0)
            if resp.status_code == 200:
                count += 1
        qps = count / duration
        assert qps >= 20, f"Throughput {qps:.1f} QPS below 20 QPS target"

Accuracy Validation with a Golden Set

Evaluation template

# tests/evaluation/test_accuracy.py
import pytest
from app.retrieval.retriever import retrieve
from app.generation.generator import generate_answer

def evaluate_accuracy(system_ask_fn, test_set):
    """Evaluate accuracy: % of answers that contain the expected keywords."""
    hits = 0
    for item in test_set:
        answer = system_ask_fn(item["question"])
        keywords = item.get("expected_keywords", [])
        if any(kw.lower() in answer.lower() for kw in keywords):
            hits += 1
    return hits / len(test_set) if test_set else 0.0

@pytest.mark.accuracy
def test_accuracy_above_90_percent(golden_set, collection):
    """Validate that the system exceeds 90% accuracy on the golden set."""
    def ask(q):
        retrieved = retrieve(q, collection, top_k=5)
        return generate_answer(q, retrieved["documents"][0] if retrieved["documents"][0] else [])

    accuracy = evaluate_accuracy(ask, golden_set)
    assert accuracy >= 0.90, f"Accuracy {accuracy:.2%} below 90% threshold"

Minimum recommended test set

Include at least:

20 frequent questions — cover 80% of typical usage
10 difficult or ambiguous questions — measure robustness
10 out-of-coverage questions — validate "I don't know" instead of hallucinating

Example golden_set.json:

[
  {
    "question": "What is a vector database?",
    "expected_keywords": ["vector", "embedding", "search", "similarity"],
    "category": "frequent"
  },
  {
    "question": "What is the difference between HNSW and IVF?",
    "expected_keywords": ["graph", "clustering", "approximate"],
    "category": "difficult"
  },
  {
    "question": "What happened on March 15, 2030 on Mars?",
    "expected_keywords": ["i don't have", "evidence", "unknown"],
    "category": "out_of_coverage"
  }
]

Suggested Thresholds

Metric	Threshold	Action if it fails
Accuracy	≥ 0.85 (90% target)	Review chunking, top_k, prompt
p95 latency `/ask`	< 2.5s (2s target)	Reduce top_k, cache embeddings, faster LLM
Throughput	> 20 QPS	Scale horizontally, optimize retrieval
Error rate	< 1%	Review retries, timeouts, error handling

Adjust these values based on your stack (LLM, hardware, data).

Evaluation Report

Generate an automated report after each run:

# scripts/generate_evaluation_report.py
def generate_report(results: dict) -> str:
    return f"""
# RAG Evaluation Report

| Metric     | Result    | Threshold | Status   |
|------------|-----------|----------|----------|
| Accuracy   | {results.get('accuracy', 0):.2%} | 0.90     | {'✅' if results.get('accuracy', 0) >= 0.90 else '❌'} |
| p95 /ask   | {results.get('p95_ms', 0):.0f}ms | 2000ms   | {'✅' if results.get('p95_ms', 0) < 2000 else '❌'} |
| Throughput | {results.get('qps', 0):.1f} QPS | 20 QPS   | {'✅' if results.get('qps', 0) >= 20 else '❌'} |
| Error rate | {results.get('error_rate', 0):.2%} | 1%       | {'✅' if results.get('error_rate', 0) < 0.01 else '❌'} |
"""

Exercises with Detailed Solutions

Exercise 1: Parametrize chunking tests

Goal: Add 3 combinations of (chunk_size, overlap) that validate edge cases.

Solution:

@pytest.mark.parametrize("chunk_size,overlap,min_chunks", [
    (50, 0, 20),
    (200, 50, 5),
    (512, 128, 2),
])
def test_chunk_sizes(self, chunk_size, overlap, min_chunks):
    text = "a " * 1000
    chunks = chunk_text(text, chunk_size=chunk_size, overlap=overlap)
    assert len(chunks) >= min_chunks
    for c in chunks:
        assert len(c) <= chunk_size + overlap

Exercise 2: Ingestion idempotency test

Goal: Verify that re-ingesting the same documents does not duplicate records.

Solution:

@pytest.mark.integration
def test_ingest_idempotent(collection, sample_docs):
    from app.ingestion.pipeline import ingest_batch
    ids = [f"doc_{i}" for i in range(len(sample_docs))]
    ingest_batch(collection, sample_docs, ids)
    count_1 = collection.count()
    ingest_batch(collection, sample_docs, ids)  # Re-ingest
    count_2 = collection.count()
    assert count_2 == count_1, "Re-ingestion should not duplicate"

Exercise 3: Latency test with pytest-benchmark

Goal: Use pytest-benchmark to measure /search latency.

Solution:

@pytest.mark.performance
def test_search_latency_benchmark(benchmark, api_client):
    def _search():
        return api_client.get("/search", params={"q": "vector", "top_k": 5})

    result = benchmark(_search)
    assert result.stats["mean"] < 0.5  # 500ms

Exercise 4: Extend the golden set with negative cases

Goal: Add 5 questions that must return "I have no evidence" without hallucinating.

Solution:

NEGATIVE_CASES = [
    {"question": "How much does the moon weigh in exact kilograms?", "expected_keywords": ["no", "evidence", "unknown"]},
    {"question": "Give me your grandmother's cake recipe", "expected_keywords": ["i don't have", "outside"]},
]
# In test_accuracy: combine golden_set + NEGATIVE_CASES and verify there is no hallucination

Exercise 5: Fixture that mocks the LLM

Goal: Create a fixture that returns a fixed answer for fast tests without calling OpenAI.

Solution:

@pytest.fixture
def mock_llm(monkeypatch):
    def fake_generate(question, context):
        return f"Mock answer for: {question[:50]}"
    from app.generation import generator
    monkeypatch.setattr(generator, "generate_answer", fake_generate)

Exercise 6: Error handling test in `/ask`

Goal: Verify that when ChromaDB fails, the API returns 503 with a clear message.

Solution:

@pytest.mark.integration
async def test_ask_handles_chromadb_unavailable(monkeypatch):
    from app.main import app
    from app.retrieval import retriever
    def raise_err(*args, **kwargs):
        raise ConnectionError("ChromaDB unavailable")
    monkeypatch.setattr(retriever, "retrieve", raise_err)
    async with AsyncClient(app=app, base_url="http://test") as client:
        response = await client.post("/ask", json={"question": "test"})
    assert response.status_code == 503
    assert "unavailable" in response.json().get("detail", "").lower()

Test Troubleshooting

"Tests green, but users complain"

Add real user query cases to your golden set. The initial tests usually cover ideal cases; complaints come from ambiguous questions, typos, or specific domains not represented.

"Accuracy goes up, latency too"

Evaluate the trade-off: more context (higher top_k) improves accuracy but increases latency. Define which metric is the priority per phase (MVP: latency; maturity: accuracy). Consider embedding cache and selective re-ranking.

"We don't know why a question fails"

Save the retrieval output and final prompt for diagnosis. Add a debug mode that returns retrieved_docs and prompt_sent in responses when X-Debug: true.

"Integration tests are slow"

Use in-memory databases or temporary files. Run unit tests first (pytest tests/unit -m unit) and integration only when necessary. Parallelize with pytest-xdist.

"Performance tests fail in CI but pass locally"

CI usually has less CPU/memory. Increase thresholds for CI or run performance tests only in staging, not on every commit.

Command to Run Tests

# Unit tests only (fast)
pytest tests/unit -v -m unit

# Integration (requires the API running)
pytest tests/integration -v -m integration

# Performance (requires real load)
pytest tests/performance -v -m performance

# Accuracy (may require an LLM API key)
pytest tests/evaluation -v -m accuracy

# All with coverage
pytest tests/ -v --cov=app --cov-report=html

Summary

You defined objective acceptance metrics: accuracy ≥90%, p95 <2s, throughput >20 QPS, error rate <1%.
You implemented unit tests for chunking, filters, and utilities with pytest and @pytest.mark.parametrize.
You implemented integration tests for /health, /search, /ask with reusable fixtures.
You added performance tests for latency and throughput with automatic thresholds.
You configured accuracy validation with a golden set (20 frequent + 10 difficult + 10 out-of-coverage).
You created fixtures in conftest.py for ChromaDB, the golden set, and the API client.
You have an evaluation report that documents the system's status before deploy.

Next step: Instrument and deploy to operate the system with observability (Capsule 06).

Additional Resources

pytest Documentation — fixtures, parametrize, markers
pytest-asyncio — async tests for FastAPI
pytest-benchmark — performance measurement
ChromaDB Testing Guide — best practices with ChromaDB
FastAPI Testing — TestClient, dependency override
RAGAS: RAG Evaluation — RAG evaluation metrics
LLM-as-Judge — LLM-based evaluation for quality
Locust — load testing for throughput and latency

Estimated time: 45-55 minutes
Next: 06-deployment-observability.md

← PreviousCapsule 04: Retrieval, Generation, and API Next →Capsule 06: Observability and Deployment

Capsule 05: Testing and Evaluation

Capsule description

Why Testing in RAG is Critical

The risk of not testing

Testing pyramid for RAG

Recommended Test Types

Unit

Integration

Performance

Quality (Accuracy)

Project Structure for Tests

pytest Configuration

pytest.ini

requirements-dev.txt

Reusable Fixtures (conftest.py)

Unit Tests: ChromaDB Operations

Integration Tests: API Endpoints

Performance Tests

Accuracy Validation with a Golden Set

Evaluation template

Minimum recommended test set

Suggested Thresholds

Evaluation Report

Exercises with Detailed Solutions

Exercise 1: Parametrize chunking tests

Exercise 2: Ingestion idempotency test

Exercise 3: Latency test with pytest-benchmark

Exercise 4: Extend the golden set with negative cases

Exercise 5: Fixture that mocks the LLM

Exercise 6: Error handling test in /ask

Test Troubleshooting

"Tests green, but users complain"

"Accuracy goes up, latency too"

"We don't know why a question fails"

"Integration tests are slow"

"Performance tests fail in CI but pass locally"

Command to Run Tests

Summary

Additional Resources

Exercise 6: Error handling test in `/ask`