🎯 El Desafío de Medir Calidad en RAG
Imagina que tu sistema RAG está en producción:
- ✅ API responde en 1-3 segundos
- ✅ 100+ requests por día
- ✅ Usuarios parecen contentos
Pero hay preguntas críticas sin respuesta:
- ¿Las respuestas son correctas?
- ¿El LLM está inventando información (hallucinations)?
- ¿Los documentos recuperados son relevantes?
- ¿Qué % de queries tienen buena calidad?
- ¿Cómo identificar y fix problemas sistemáticos?
Sin evaluación: Volando a ciegas, optimizando por intuición, descubriendo problemas cuando usuarios se quejan.
📊 La Magnitud del Problema
Dimensiones de Calidad en RAG
Un sistema RAG tiene múltiples puntos de falla:
Query → Embedding → Qdrant Search → Reranking → LLM Generation → Answer
Cada paso puede fallar:
├─ Embedding: Query mal entendido
├─ Search: Documentos irrelevantes recuperados
├─ Reranking: Orden incorrecto
└─ LLM: Hallucination, respuesta incompleta, tono inadecuado
Tipos de Métricas Necesarias
-
📊 Retrieval Metrics: ¿Recuperamos los documentos correctos?
- Precision@k
- Recall@k
- MRR (Mean Reciprocal Rank)
- nDCG (Normalized Discounted Cumulative Gain)
-
🤖 Generation Metrics: ¿El LLM genera buenas respuestas?
- Relevance (pregunta ↔ respuesta)
- Correctness (contexto ↔ respuesta)
- Completeness
- Hallucination rate
- Grounding (respuesta basada en contexto)
-
⚡ System Metrics: ¿El sistema es rápido y confiable?
- Latency (p50, p95, p99)
- Throughput (requests/s)
- Error rate
- Availability (uptime)
-
💰 Cost Metrics: ¿Es sostenible económicamente?
- Tokens consumidos por query
- Costo por query
- Costo mensual proyectado
💡 La Solución: Evaluación Multi-Nivel
Arquitectura de Evaluación
┌───────────────────────────────────────────────────────┐
│ User Query (Production) │
└─────────────────┬─────────────────────────────────────┘
│
┌────────▼────────┐
│ RAG Pipeline │
│ (Fast path) │
└────────┬────────┘
│
┌─────────────┴──────────────┐
│ │
Response to User Enqueue Evaluation
(1-3 seconds) (Non-blocking, async)
│
┌─────────────▼──────────────┐
│ Evaluation Worker │
│ (Background thread) │
└─────────────┬──────────────┘
│
┌─────────────▼──────────────┐
│ Phoenix Evals │
│ (LLM-as-a-Judge) │
├────────────────────────────┤
│ - Relevance │
│ - Hallucination │
│ - Toxicity │
│ - Grounding │
└─────────────┬──────────────┘
│
┌─────────────▼──────────────┐
│ Phoenix UI │
│ - Dashboards │
│ - Metrics │
│ - Drill-down │
└────────────────────────────┘
Key insight: Evaluación asíncrona no bloquea respuesta al usuario.
🤖 LLM-as-a-Judge Evaluation
EvaluationService
Archivo src/lus_laboris_api/api/services/evaluation_service.py:
"""
Asynchronous evaluation service using Phoenix Evals
"""
import asyncio
import logging
import queue
from concurrent.futures import ThreadPoolExecutor
from typing import Any
from phoenix.evals import (
HALLUCINATION_PROMPT_TEMPLATE,
RAG_RELEVANCY_PROMPT_TEMPLATE,
OpenAIModel,
)
from ..config import settings
from .phoenix_service import phoenix_service
logger = logging.getLogger(__name__)
class EvaluationService:
"""Service for asynchronous RAG response evaluation"""
def __init__(self):
self.enabled = settings.api_phoenix_enabled
self.evaluation_queue = queue.Queue()
self.executor = ThreadPoolExecutor(
max_workers=2,
thread_name_prefix="eval-worker"
)
if self.enabled:
self._initialize_evaluators()
self._start_evaluation_worker()
logger.info("Evaluation service initialized")
else:
logger.warning("Evaluation service disabled")
def _initialize_evaluators(self):
"""Initialize Phoenix evaluators with LLM"""
try:
# Use cost-effective model for evaluations
eval_model = "gpt-4o-mini" # Barato y bueno para evals
self.eval_model = OpenAIModel(
model=eval_model,
api_key=settings.openai_api_key
)
logger.info(f"Evaluators initialized with model: {eval_model}")
except Exception as e:
logger.exception("Failed to initialize evaluators")
self.enabled = False
def _start_evaluation_worker(self):
"""Start worker that processes evaluations in background"""
def worker():
logger.info("Evaluation worker started")
while True:
try:
# Get task from queue
eval_task = self.evaluation_queue.get(timeout=1.0)
if eval_task is None: # Shutdown signal
break
# Execute evaluation
self._run_evaluation(eval_task)
self.evaluation_queue.task_done()
except queue.Empty:
continue
except Exception as e:
logger.exception("Error in evaluation worker")
# Start worker in separate thread
self.executor.submit(worker)
def enqueue_evaluation(
self,
session_id: str,
question: str,
context: str,
answer: str,
documents: list[dict],
metadata: dict | None = None,
):
"""Enqueue evaluation for async processing (non-blocking)"""
if not self.enabled:
return
eval_task = {
"session_id": session_id,
"question": question,
"context": context,
"answer": answer,
"documents": documents,
"metadata": metadata or {},
"timestamp": datetime.now().isoformat(),
}
self.evaluation_queue.put(eval_task)
logger.debug(f"Evaluation enqueued for session {session_id}")
def _run_evaluation(self, eval_task: dict):
"""Execute evaluations (runs async loop)"""
try:
# Create new event loop for this thread
loop = asyncio.new_event_loop()
asyncio.set_event_loop(loop)
loop.run_until_complete(self._run_evaluation_async(eval_task))
finally:
loop.close()
async def _run_evaluation_async(self, eval_task: dict):
"""Execute Phoenix Evals in parallel (optimized)"""
session_id = eval_task["session_id"]
question = eval_task["question"]
context = eval_task["context"]
answer = eval_task["answer"]
try:
logger.info(f"Running parallel evaluations for session {session_id}")
start_time = time.time()
# ✅ Run all 3 evaluations in parallel
evaluation_results = await asyncio.gather(
asyncio.to_thread(
self._evaluate_relevance,
question, context, answer
),
asyncio.to_thread(
self._evaluate_hallucination,
question, context, answer
),
asyncio.to_thread(
self._evaluate_toxicity,
answer
),
return_exceptions=True, # Don't fail if one fails
)
eval_time = time.time() - start_time
# Extract results
relevance_score = (
evaluation_results[0]
if not isinstance(evaluation_results[0], Exception)
else None
)
hallucination_score = (
evaluation_results[1]
if not isinstance(evaluation_results[1], Exception)
else None
)
toxicity_score = (
evaluation_results[2]
if not isinstance(evaluation_results[2], Exception)
else None
)
# Track results in Phoenix
phoenix_service.track_evaluation(
session_id=session_id,
metrics={
"relevance": relevance_score,
"hallucination": hallucination_score,
"toxicity": toxicity_score,
"grounding": 1 - hallucination_score if hallucination_score else None,
},
metadata={
**eval_task["metadata"],
"evaluation_time": eval_time,
}
)
logger.info(
f"Evaluation completed for session {session_id} in {eval_time:.2f}s"
)
except Exception as e:
logger.exception(f"Evaluation failed for session {session_id}")
def _evaluate_relevance(self, question: str, context: str, answer: str) -> float:
"""Evaluate if answer is relevant to question"""
try:
# Use Phoenix RAG_RELEVANCY_PROMPT_TEMPLATE
from phoenix.evals import run_evals
df = pd.DataFrame([{
"question": question,
"context": context,
"answer": answer,
}])
results = run_evals(
dataframe=df,
template=RAG_RELEVANCY_PROMPT_TEMPLATE,
model=self.eval_model,
)
score = results["score"].iloc[0]
return float(score)
except Exception as e:
logger.exception("Failed to evaluate relevance")
return None
def _evaluate_hallucination(self, question: str, context: str, answer: str) -> float:
"""Evaluate if answer contains hallucinations"""
try:
# Use Phoenix HALLUCINATION_PROMPT_TEMPLATE
df = pd.DataFrame([{
"query": question,
"reference": context,
"response": answer,
}])
results = run_evals(
dataframe=df,
template=HALLUCINATION_PROMPT_TEMPLATE,
model=self.eval_model,
)
score = results["score"].iloc[0]
return float(score)
except Exception as e:
logger.exception("Failed to evaluate hallucination")
return None
def _evaluate_toxicity(self, answer: str) -> float:
"""Evaluate if answer contains toxic content"""
try:
# Simple heuristic or use Phoenix toxicity template
# For legal domain, toxicity is usually very low
toxic_keywords = ["offensive", "discriminatory", "hate"]
score = sum(1 for word in toxic_keywords if word in answer.lower())
# Normalize to 0-1 (0 = no toxic, 1 = very toxic)
return min(score / len(toxic_keywords), 1.0)
except Exception as e:
logger.exception("Failed to evaluate toxicity")
return None
# Global service instance
evaluation_service = EvaluationService()
Características:
- ✅ Async/Non-blocking: No retrasa respuesta al usuario
- ✅ Queue-based: ThreadPoolExecutor procesa en background
- ✅ Parallel evaluations: 3 evals en paralelo con
asyncio.gather - ✅ Phoenix Evals: Templates pre-optimizados
- ✅ Cost-effective: usa
gpt-4o-mini(barato)
Evaluaciones Implementadas
1. Relevance (Relevancia):
Question: "¿Cuántos días de vacaciones?"
Answer: "Según el Artículo 218, son 12 días hábiles."
Eval Prompt:
"Is the answer relevant to the question?
Question: {question}
Answer: {answer}
Score 0-1 where 1 = highly relevant"
Score: 0.95 ✅ (muy relevante)
2. Hallucination (Invención):
Context: "Artículo 218: todo trabajador que cumpla un año... 12 días hábiles"
Answer: "Son 12 días hábiles según Artículo 218"
Eval Prompt:
"Does the response contain information NOT in the reference?
Reference: {context}
Response: {answer}
Score 0-1 where 1 = full hallucination"
Score: 0.05 ✅ (casi sin hallucination)
3. Toxicity (Toxicidad):
Answer: "Según el Artículo 218..."
Eval Prompt:
"Does the text contain toxic, offensive, or harmful content?
Text: {answer}
Score 0-1 where 1 = very toxic"
Score: 0.0 ✅ (sin toxicidad)
📊 Métricas de Retrieval
Precision@k
Definición: ¿Cuántos de los top-k resultados son relevantes?
def precision_at_k(retrieved_docs: list[int], relevant_docs: list[int], k: int) -> float:
"""
Calculate Precision@k
Args:
retrieved_docs: IDs de documentos recuperados (ordenados por score)
relevant_docs: IDs de documentos realmente relevantes (ground truth)
k: Número de documentos a considerar
Returns:
Precision@k score (0-1)
"""
top_k = retrieved_docs[:k]
relevant_in_top_k = len(set(top_k) & set(relevant_docs))
return relevant_in_top_k / k
Ejemplo:
retrieved = [218, 42, 219, 81, 220] # Top-5 de Qdrant
relevant = [218, 219, 220] # Ground truth
precision_at_1 = 1/1 = 1.0 # Perfecto! #1 es relevante
precision_at_3 = 2/3 = 0.67 # De los top-3, 2 son relevantes
precision_at_5 = 3/5 = 0.60 # De los top-5, 3 son relevantes
Nuestra métrica en producción:
- Precision@1: 89% (con reranking)
- Precision@5: 96%
Recall@k
Definición: ¿Qué % de documentos relevantes recuperamos?
def recall_at_k(retrieved_docs: list[int], relevant_docs: list[int], k: int) -> float:
"""Calculate Recall@k"""
top_k = retrieved_docs[:k]
relevant_in_top_k = len(set(top_k) & set(relevant_docs))
return relevant_in_top_k / len(relevant_docs)
Ejemplo:
retrieved = [218, 42, 219, 81, 220]
relevant = [218, 219, 220] # 3 relevantes totales
recall_at_5 = 3/3 = 1.0 # Recuperamos TODOS los relevantes
MRR (Mean Reciprocal Rank)
Definición: Posición promedio del primer resultado relevante
def mrr(retrieved_docs: list[int], relevant_docs: list[int]) -> float:
"""Calculate Mean Reciprocal Rank"""
for i, doc_id in enumerate(retrieved_docs, 1):
if doc_id in relevant_docs:
return 1.0 / i # Reciprocal del rank
return 0.0 # No hay relevantes
Ejemplo:
# Query 1: Relevante en posición #1
retrieved_1 = [218, 42, 219]
relevant_1 = [218]
mrr_1 = 1/1 = 1.0
# Query 2: Relevante en posición #3
retrieved_2 = [42, 81, 219]
relevant_2 = [219]
mrr_2 = 1/3 = 0.33
# MRR promedio: (1.0 + 0.33) / 2 = 0.67
Nuestra métrica: MRR = 0.92 (documento relevante usualmente en posición 1-2)
🎯 Integración en el RAG Pipeline
En el RAGService
async def answer_question(self, question: str, session_id: str):
"""Complete RAG pipeline with evaluation"""
# 1. Retrieve documents
documents, _ = self._retrieve_documents(question, session_id)
# 2. Generate answer
answer = await self._generate_response(question, documents, session_id)
# 3. Build context
context = self._build_context(documents)
# 4. Enqueue evaluation (NON-BLOCKING)
evaluation_service.enqueue_evaluation(
session_id=session_id,
question=question,
context=context,
answer=answer,
documents=documents,
metadata={
"llm_provider": self.llm_provider,
"llm_model": self.llm_model,
"processing_time": processing_time,
}
)
# 5. Return response immediately (evaluation runs in background)
return {
"success": True,
"answer": answer,
"session_id": session_id,
...
}
Timeline:
t=0ms: User request arrives
t=100ms: Documents retrieved
t=2000ms: LLM generates answer
t=2010ms: Response returned to user ← USER HAPPY
t=2050ms: Evaluation enqueued (async)
t=4500ms: Evaluations complete (background) ← METRICS TRACKED
User experience: 2 seconds (no ve los 2.5s de evaluación)
📊 Visualización en Phoenix UI
Dashboard de Métricas
Acceso: http://localhost:6006 (local) o Phoenix Cloud
Vista de Evaluaciones:
┌─────────────────────────────────────────────────────┐
│ Evaluation Results (Last 7 days) │
│ │
│ Total Queries Evaluated: 1,234 │
│ │
│ Relevance (avg): 0.89 ████████████████░░ 89% │
│ Hallucination (avg): 0.05 ░░░░░░░░░░░░░░░░░░ 5% │
│ Toxicity (avg): 0.01 ░░░░░░░░░░░░░░░░░░ 1% │
│ Grounding (avg): 0.95 █████████████████░ 95% │
│ │
│ ┌─────────────────────────────────────────────┐ │
│ │ Relevance Distribution │ │
│ │ ┌────────────────────────────────────┐ │ │
│ │ │ [0.0-0.2]: ▓ 2% │ │ │
│ │ │ [0.2-0.4]: ▓ 3% │ │ │
│ │ │ [0.4-0.6]: ▓▓ 5% │ │ │
│ │ │ [0.6-0.8]: ▓▓▓▓▓ 12% │ │ │
│ │ │ [0.8-1.0]: ▓▓▓▓▓▓▓▓▓▓▓▓▓▓▓ 78% │ │ │
│ │ └────────────────────────────────────┘ │ │
│ └─────────────────────────────────────────────┘ │
│ │
│ Top Issues: │
│ - 12 queries with low relevance (<0.6) │
│ - 6 potential hallucinations detected (>0.3) │
│ - 0 toxic responses │
└─────────────────────────────────────────────────────┘
Drill-Down Individual
┌─────────────────────────────────────────────────────┐
│ Session: abc-123-def │
│ Timestamp: 2024-10-16 14:30:22 │
│ │
│ Question: │
│ "¿Cuántos días de vacaciones corresponden?" │
│ │
│ Answer: │
│ "Según el Artículo 218 del Código del Trabajo..." │
│ │
│ Evaluations: │
│ ├─ Relevance: 0.92 ✅ (Highly relevant) │
│ ├─ Hallucination: 0.03 ✅ (Minimal) │
│ ├─ Toxicity: 0.00 ✅ (None) │
│ └─ Grounding: 0.97 ✅ (Well grounded) │
│ │
│ Retrieved Documents: │
│ 1. Art. 218 (score: 0.912, rerank: 0.987) │
│ 2. Art. 219 (score: 0.876, rerank: 0.921) │
│ 3. Art. 220 (score: 0.845, rerank: 0.889) │
│ │
│ Metadata: │
│ - LLM: OpenAI GPT-3.5-turbo │
│ - Processing time: 1.234s │
│ - Reranking: Enabled │
└─────────────────────────────────────────────────────┘
🎯 Casos de Uso Reales
Para Identificar Problemas de Calidad:
"Algunas respuestas parecen incorrectas, pero no sé cuáles"
Solución:
- Abrir Phoenix UI
- Filtrar por
relevance < 0.6 - Ver queries problemáticas:
- "¿Qué es un contrato eventual?" → Relevance: 0.4
- Contexto recuperado: artículos sobre "contratos colectivos"
- Problema: Embeddings confunden "eventual" con "colectivo"
Fix: Mejorar pre-processing de queries o usar modelo de embeddings más grande
Para Detectar Hallucinations:
"¿El LLM está inventando información?"
Solución:
- Phoenix UI → Filter
hallucination > 0.3 - Ver casos:
- Query: "¿Cuál es el salario mínimo?"
- Context: (vacío - no hay artículos sobre salario)
- Answer: "El salario mínimo es $2,500,000..." ⚠️
- Hallucination score: 0.9
Fix: Agregar guardrail: Si no hay contexto relevante → "No tengo información suficiente"
Para Optimizar Performance:
"¿Qué configuración da mejores resultados?"
Solución: A/B Testing
# Configuración A: Sin reranking
results_a = run_eval_suite(use_reranking=False)
# Relevance: 0.85
# Configuración B: Con reranking
results_b = run_eval_suite(use_reranking=True)
# Relevance: 0.92 (+7%)
# Conclusión: Reranking vale la pena!
Para Monitoreo Continuo:
"Quiero alertas cuando calidad baja"
Solución:
# En Phoenix o monitoring system
if avg_relevance_last_24h < 0.80:
send_alert("⚠️ RAG quality degraded! Avg relevance: 0.78")
if hallucination_rate_last_24h > 0.10:
send_alert("🚨 High hallucination rate: 12%")
🚀 El Impacto Transformador
Antes de Evaluación Automática:
- 🤷 Quality unknown: "Parece que funciona bien..."
- 🐛 Reactive debugging: Usuarios reportan problemas
- 📊 No metrics: Sin datos para optimizar
- 🎲 Blind optimization: Cambios sin medir impacto
- ⏱️ Manual testing: Probar 10-20 queries manualmente
Después de Evaluación Automática:
- 📊 Quality measured: "89% relevance, 5% hallucination"
- 🔍 Proactive debugging: Identificar problemas antes que usuarios
- 📈 Data-driven: Optimizar basado en métricas reales
- 🎯 A/B testing: Comparar configuraciones objetivamente
- ⚡ Continuous evaluation: 100% de queries evaluadas automáticamente
Métricas de Mejora:
| Aspecto | Sin Evaluación | Con Evaluación | Mejora |
|---|---|---|---|
| Quality visibility | 0% | 100% | +∞ |
| Problem detection | Días (user reports) | Minutos (automated) | -99% |
| Evaluation coverage | 1-5% (manual) | 100% (automated) | +2000% |
| Optimization confidence | Baja (guessing) | Alta (data-driven) | N/A |
| Time to fix issues | Días | Horas | -90% |
💡 Lecciones Aprendidas
1. Async Evaluation es Crítico
Evaluación no debe agregar latencia a la respuesta del usuario. Queue + ThreadPoolExecutor = perfecto.
2. gpt-4o-mini es Ideal para Evals
- GPT-4: Muy caro para evaluar cada query ($30/1M tokens)
- GPT-3.5-turbo: Calidad inconsistente
- gpt-4o-mini: Balance perfecto ($0.15/1M tokens, buena calidad)
3. Phoenix Templates > Custom
Los templates de Phoenix (HALLUCINATION, RAG_RELEVANCY) están battle-tested. Úsalos.
4. Parallel Evaluations = 3x Faster
asyncio.gather ejecuta 3 evals en paralelo → 2.5s en lugar de 7.5s
5. Ground Truth es Gold
Tener un dataset de 50-100 query-answer pairs validados permite comparaciones objetivas.
6. Monitorea Tendencias, No Puntos
Un query malo no importa. 10% de queries malos sí importa. Phoenix dashboards muestran trends.
🎯 El Propósito Más Grande
Evaluación y métricas no son "nice to have" - son el sistema de calidad que garantiza que el RAG entrega valor real. Sin evaluación:
- No sabes si está funcionando
- No puedes optimizar
- No detectas regresiones
- No justificas inversión
Con evaluación automática continua:
- ✅ Visibility: Calidad medida en cada query
- ✅ Confidence: Cambios basados en datos
- ✅ Quality assurance: Detección temprana de problemas
- ✅ Continuous improvement: Optimización basada en métricas
- ✅ Stakeholder reporting: Dashboards para el negocio
- ✅ Cost optimization: Identificar oportunidades de ahorro
Estamos convirtiendo el RAG de una "caja negra" a un sistema medido, monitoreado y optimizable, donde cada respuesta contribuye a mejorar el siguiente millón de respuestas.
🔗 Recursos y Enlaces
Repositorio del Proyecto
- GitHub: lus-laboris-py
Documentación Técnica
-
Evaluation Service:
src/lus_laboris_api/api/services/evaluation_service.py -
Phoenix Service:
src/lus_laboris_api/api/services/phoenix_service.py -
RAG Service:
src/lus_laboris_api/api/services/rag_service.py
Recursos Externos
- Phoenix Evals: docs.arize.com/phoenix/evaluation
- RAG Evaluation Paper: arxiv.org/abs/2312.10997
- RAGAS Framework: github.com/explodinggradients/ragas
- Evaluating RAG Systems: arize.com/blog/rag-evaluation
🎊 Fin de la Serie LLPY
Hemos completado un viaje de 14 posts construyendo un sistema RAG de clase mundial:
Serie Completa:
- LLPY-01: Introducción al Sistema RAG
- LLPY-02: Configuración con UV
- LLPY-03: Extracción de Datos Legales
- LLPY-04: Vectorización y Embeddings
- LLPY-05: Qdrant Base de Datos Vectorial
- LLPY-06: FastAPI API REST Robusta
- LLPY-07: Integración LLMs (OpenAI + Gemini)
- LLPY-08: Reranking para Precisión
- LLPY-09: Phoenix y OpenTelemetry
- LLPY-10: Autenticación JWT con RSA
- LLPY-11: Terraform - Infraestructura como Código
- LLPY-12: Docker y Containerización
- LLPY-13: CI/CD con GitHub Actions
- LLPY-14: Evaluación y Métricas de Calidad ← ESTE POST
Lo que Construimos:
✅ Sistema RAG completo end-to-end
✅ 413 artículos legales vectorizados
✅ API REST producción-ready
✅ Multi-provider LLM (OpenAI + Gemini)
✅ Observabilidad completa (Phoenix + OpenTelemetry)
✅ Infraestructura como código (Terraform)
✅ CI/CD automatizado (GitHub Actions)
✅ Evaluación continua (LLM-as-a-judge)
Métricas del Sistema Final:
| Métrica | Valor | Benchmark |
|---|---|---|
| Latency (p50) | 1.2s | <2s ✅ |
| Latency (p95) | 2.8s | <5s ✅ |
| Precision@1 | 89% | >80% ✅ |
| Hallucination rate | 5% | <10% ✅ |
| Uptime | 99.9% | >99% ✅ |
| Cost per query | $0.003 | <$0.01 ✅ |
Impacto Transformador Final:
De 0 a sistema RAG de producción en 14 posts:
- 🚀 Performance: 30-50ms retrieval, 1-3s total
- 🎯 Quality: 89% relevance, 5% hallucination
- 💰 Cost: $0.003/query, escalable
- 🔒 Security: JWT auth, secrets management
- 📊 Observability: Phoenix tracking completo
- 🔄 Deployment: CI/CD en 15 minutos
- 📈 Evaluation: 100% de queries evaluadas
Gracias por acompañarnos en este viaje! 🎉
Si tienes preguntas o quieres contribuir al proyecto, visita el repositorio en GitHub.
Top comments (0)