Módulo 8: Proyecto Integrador RAG con ChromaDB

Cápsula 05: Testing y Evaluación

Descripción de la cápsula

Un RAG sin evaluación puede parecer correcto en desarrollo y fallar estrepitosamente en producción. Los usuarios no perdonan respuestas alucinadas, latencias de 10 segundos o errores esporádicos sin diagnóstico.

En esta cápsula definirás y ejecutarás una estrategia de pruebas completa: unit tests para operaciones de ChromaDB, integration tests para los endpoints de la API, performance tests para latencia y throughput, y accuracy validation con un golden set. Usarás pytest con fixtures y parametrización para mantener tests mantenibles y reutilizables.

Al final tendrás evidencia objetiva de que tu sistema cumple umbrales production-ready: latencia p95 < 2s, throughput > 20 QPS, accuracy > 90%.

Por qué Testing en RAG es Crítico

El riesgo de no testear

Escenario típico:
├── Ingestion funciona
├── Retrieval devuelve documentos
├── Generation responde algo coherente
└── ¿Pero es correcto? ¿Es rápido? ¿Escala?

Sin tests: no lo sabes hasta que el usuario se queja.

RAG combina varios componentes con fallos sutiles: chunking puede cortar contexto crítico, embeddings pueden degradarse con datos nuevos, el LLM puede inventar cuando retrieval falla. Los tests te dan confianza para desplegar y evolucionar.

Pirámide de testing para RAG

                    ▲
                   / \
                  / E2E \         Pocos, lentos: flujo completo
                 /───────\
                /  Integ  \       Más: API + ChromaDB + mocks LLM
               /───────────\
              /    Unit     \     Muchos, rápidos: chunking, filtros, utilidades
             /───────────────\

Unit: funciones puras (chunking, filtros, formateo)
Integration: flujo real /ask con ChromaDB (embeddings reales o mock)
E2E/Performance: latencia, throughput, accuracy con golden set

Tipos de Prueba Recomendados

Unitarias

Objetivo: validar lógica aislada sin dependencias externas.

Qué testear	Ejemplo
Chunking	Texto largo → chunks de tamaño esperado con overlap
Filtros	Función que aplica `where` sobre metadata
Utilidades	Normalización de IDs, formateo de fuentes
Validación de inputs	Longitud máxima de pregunta, caracteres prohibidos

Integración

Objetivo: validar flujo real entre componentes.

Qué testear	Ejemplo
`/ingest`	Documentos se insertan y son consultables
`/search`	Query retorna resultados con scores y metadata
`/ask`	Pregunta → retrieval → generación → respuesta con fuentes
`/health`	Responde OK cuando ChromaDB está disponible

Performance

Objetivo: asegurar latencia y throughput dentro de umbrales.

Métrica	Umbral sugerido	Cómo medir
Latencia p95 `/ask`	< 2s	`locust` o `pytest-benchmark`
Throughput	> 20 QPS	requests concurrentes sostenidos
Latencia p95 `/search`	< 500ms	solo retrieval, sin LLM

Calidad (Accuracy)

Objetivo: validar que las respuestas son correctas y fundamentadas.

Qué medir	Método
Keyword match	Respuesta contiene términos esperados
Relevance	Documentos recuperados son relevantes para la pregunta
Hallucination	Respuesta no inventa cuando no hay evidencia

Estructura de Proyecto para Tests

project/
├── app/
│   ├── api/
│   ├── retrieval/
│   ├── generation/
│   └── ingestion/
├── tests/
│   ├── conftest.py          # Fixtures compartidos
│   ├── 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

Configuración de pytest

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

Fixtures Reutilizables (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():
    """Cliente ChromaDB persistente para tests de sesión."""
    path = tempfile.mkdtemp(prefix="chroma_test_")
    client = chromadb.PersistentClient(path=path)
    yield client
    # Cleanup opcional: eliminar directorio

@pytest.fixture
def collection(chroma_client):
    """Colección limpia por test para aislamiento."""
    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 de preguntas con keywords esperados para accuracy."""
    return [
        {"question": "¿Qué es un vector database?", "expected_keywords": ["vector", "embedding", "búsqueda"]},
        {"question": "¿Cómo funciona HNSW?", "expected_keywords": ["grafo", "aproximado", "vecino"]},
        {"question": "¿Cuándo usar ChromaDB?", "expected_keywords": ["local", "desarrollo", "RAG"]},
        # ... 20+ items
    ]

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

@pytest.fixture
def api_client():
    """Cliente HTTP para tests de integración contra la 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:
    """Tests unitarios de la lógica de chunking."""

    @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(["palabra"] * 500)
        chunks = chunk_text(text, chunk_size=chunk_size, overlap=overlap)
        assert len(chunks) >= expected_count

    def test_chunk_overlap_preserves_context(self):
        text = "Este es un texto con contexto importante en el medio."
        chunks = chunk_text(text, chunk_size=20, overlap=10)
        # El overlap debe garantizar que "importante" no se pierda entre chunks
        full_joined = " ".join(chunks)
        assert "importante" 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": "es"})
        assert where == {"$and": [{"category": "tech"}, {"language": "es"}]}

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": "¿Qué es un 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
        # Debe indicar que no hay evidencia suficiente
        data = response.json()
        assert "evidencia" in data.get("answer", "").lower() or "no encontrado" 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:
    """Tests de performance: latencia p95 < 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": "¿Qué es 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  # segundos
        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 con Golden Set

Plantilla de evaluación

# 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):
    """Evalúa accuracy: % de respuestas que contienen keywords esperados."""
    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):
    """Valida que el sistema supera 90% de accuracy en 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"

Set de prueba mínimo recomendado

Incluye al menos:

20 preguntas frecuentes — cubren el 80% del uso típico
10 preguntas difíciles o ambiguas — miden robustez
10 preguntas fuera de cobertura — validan “no sé” en lugar de alucinar

Ejemplo golden_set.json:

[
  {
    "question": "¿Qué es un vector database?",
    "expected_keywords": ["vector", "embedding", "búsqueda", "similaridad"],
    "category": "frequent"
  },
  {
    "question": "¿Cuál es la diferencia entre HNSW e IVF?",
    "expected_keywords": ["grafo", "clustering", "aproximado"],
    "category": "difficult"
  },
  {
    "question": "¿Qué pasó el 15 de marzo de 2030 en Marte?",
    "expected_keywords": ["no tengo", "evidencia", "desconocido"],
    "category": "out_of_coverage"
  }
]

Umbrales Sugeridos

Métrica	Umbral	Acción si falla
Accuracy	≥ 0.85 (90% objetivo)	Revisar chunking, top_k, prompt
Latencia p95 `/ask`	< 2.5s (2s objetivo)	Reducir top_k, cachear embeddings, LLM más rápido
Throughput	> 20 QPS	Escalar horizontalmente, optimizar retrieval
Error rate	< 1%	Revisar retries, timeouts, manejo de errores

Ajusta estos valores según tu stack (LLM, hardware, datos).

Reporte de Evaluación

Genera un reporte automatizado después de cada run:

# scripts/generate_evaluation_report.py
def generate_report(results: dict) -> str:
    return f"""
# Reporte de Evaluación RAG

| Métrica    | Resultado | Umbral   | Estado   |
|------------|-----------|----------|----------|
| 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 '❌'} |
"""

Ejercicios con Soluciones Detalladas

Ejercicio 1: Parametrizar tests de chunking

Objetivo: Añadir 3 combinaciones de (chunk_size, overlap) que validen bordes.

Solución:

@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

Ejercicio 2: Test de idempotencia de ingestion

Objetivo: Verificar que re-ingestar los mismos documentos no duplica registros.

Solución:

@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-ingesta
    count_2 = collection.count()
    assert count_2 == count_1, "Re-ingestion should not duplicate"

Ejercicio 3: Test de latency con pytest-benchmark

Objetivo: Usar pytest-benchmark para medir latencia de /search.

Solución:

@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

Ejercicio 4: Extender golden set con casos negativos

Objetivo: Añadir 5 preguntas que deben devolver “no tengo evidencia” sin alucinar.

Solución:

NEGATIVE_CASES = [
    {"question": "¿Cuánto pesa la luna en kilogramos exactos?", "expected_keywords": ["no", "evidencia", "desconocido"]},
    {"question": "Dame la receta del pastel de tu abuela", "expected_keywords": ["no tengo", "fuera"]},
]
# En test_accuracy: combinar golden_set + NEGATIVE_CASES y verificar que no hay alucinación

Ejercicio 5: Fixture que mockea el LLM

Objetivo: Crear fixture que devuelve respuesta fija para tests rápidos sin llamar a OpenAI.

Solución:

@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)

Ejercicio 6: Test de error handling en `/ask`

Objetivo: Verificar que cuando ChromaDB falla, la API devuelve 503 con mensaje claro.

Solución:

@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()

Troubleshooting de Pruebas

"Tests verdes, pero usuarios se quejan"

Agrega casos reales de consultas de usuarios a tu golden set. Los tests iniciales suelen cubrir casos ideales; las quejas vienen de preguntas ambiguas, typos o dominio específico no representado.

"Accuracy sube, latencia también"

Evalúa el trade-off: más contexto (top_k mayor) mejora accuracy pero aumenta latencia. Define cuál métrica es prioritaria por fase (MVP: latencia; madurez: accuracy). Considera cache de embeddings y re-ranking selectivo.

"No sabemos por qué falla una pregunta"

Guarda retrieval output y prompt final para diagnóstico. Añade un modo debug que devuelva retrieved_docs y prompt_sent en respuestas cuando X-Debug: true.

"Tests de integración son lentos"

Usa bases de datos en memoria o archivos temporales. Ejecuta unit tests primero (pytest tests/unit -m unit) y integration solo cuando sea necesario. Paralleliza con pytest-xdist.

"Performance tests fallan en CI pero pasan local"

CI suele tener menos CPU/memoria. Aumenta umbrales para CI o ejecuta performance tests solo en staging, no en cada commit.

Comando para Ejecutar Tests

# Solo unit tests (rápidos)
pytest tests/unit -v -m unit

# Integration (requiere API levantada)
pytest tests/integration -v -m integration

# Performance (requiere carga real)
pytest tests/performance -v -m performance

# Accuracy (puede requerir API key de LLM)
pytest tests/evaluation -v -m accuracy

# Todos con coverage
pytest tests/ -v --cov=app --cov-report=html

Resumen

Definiste métricas objetivas de aceptación: accuracy ≥90%, p95 <2s, throughput >20 QPS, error rate <1%.
Implementaste unit tests para chunking, filtros y utilidades con pytest y @pytest.mark.parametrize.
Implementaste integration tests para /health, /search, /ask con fixtures reutilizables.
Añadiste performance tests para latencia y throughput con umbrales automáticos.
Configuraste accuracy validation con golden set (20 frecuentes + 10 difíciles + 10 fuera de cobertura).
Creaste fixtures en conftest.py para ChromaDB, golden set y API client.
Tienes un reporte de evaluación que documenta el estado del sistema antes de deploy.

Próximo paso: Instrumentar y desplegar para operar el sistema con observabilidad (Cápsula 06).

Recursos Adicionales

pytest Documentation — fixtures, parametrize, markers
pytest-asyncio — tests async para FastAPI
pytest-benchmark — medición de performance
ChromaDB Testing Guide — mejores prácticas con ChromaDB
FastAPI Testing — TestClient, dependency override
RAGAS: RAG Evaluation — métricas de evaluación RAG
LLM-as-Judge — evaluación con LLM para calidad
Locust — load testing para throughput y latencia

Tiempo estimado: 45-55 minutos
Siguiente: 06-observabilidad-deployment.md

← AnteriorCápsula 04: Retrieval, Generation y API Siguiente →Cápsula 06: Observabilidad y Deployment

Cápsula 05: Testing y Evaluación

Descripción de la cápsula

Por qué Testing en RAG es Crítico

El riesgo de no testear

Pirámide de testing para RAG

Tipos de Prueba Recomendados

Unitarias

Integración

Performance

Calidad (Accuracy)

Estructura de Proyecto para Tests

Configuración de pytest

pytest.ini

requirements-dev.txt

Fixtures Reutilizables (conftest.py)

Unit Tests: ChromaDB Operations

Integration Tests: API Endpoints

Performance Tests

Accuracy Validation con Golden Set

Plantilla de evaluación

Set de prueba mínimo recomendado

Umbrales Sugeridos

Reporte de Evaluación

Ejercicios con Soluciones Detalladas

Ejercicio 1: Parametrizar tests de chunking

Ejercicio 2: Test de idempotencia de ingestion

Ejercicio 3: Test de latency con pytest-benchmark

Ejercicio 4: Extender golden set con casos negativos

Ejercicio 5: Fixture que mockea el LLM

Ejercicio 6: Test de error handling en /ask

Troubleshooting de Pruebas

"Tests verdes, pero usuarios se quejan"

"Accuracy sube, latencia también"

"No sabemos por qué falla una pregunta"

"Tests de integración son lentos"

"Performance tests fallan en CI pero pasan local"

Comando para Ejecutar Tests

Resumen

Recursos Adicionales

Ejercicio 6: Test de error handling en `/ask`