Cápsula 03: Monitoring y Observabilidad para Sistemas RAG

Descripción de la cápsula

Sin métricas confiables, operar RAG en producción es reaccionar tarde. Esta cápsula te define qué medir y cómo usar esas señales para actuar antes de que el usuario note degradación: Prometheus, Grafana, alerting con umbrales concretos, logging estructurado en JSON, runbooks, decoradores en Python para métricas, middleware FastAPI para request tracking, y troubleshooting práctico.

Tiempo estimado: 25–35 minutos
Prerequisitos: Haber leído cápsulas 01–02 (scaling, estrategias producción).

1. Métricas mínimas para RAG

Antes de instrumentar, define qué te importa. Para un sistema RAG típico necesitas estas familias de métricas:

Sin ellas, estarás debuggeando a ciegas cuando algo falle.

2. Prometheus: métricas esenciales

2.1 Query latency

Latencia del camino completo (embedding + retrieval + opcional LLM) y del retrieval aislado:

from prometheus_client import Histogram, Counter, Gauge

# Latencia de retrieval (ms)
retrieval_latency = Histogram(
    "rag_retrieval_latency_ms",
    "Latency of vector search retrieval in milliseconds",
    ["collection", "top_k"],
    buckets=[10, 25, 50, 100, 200, 500, 1000, 2500, 5000],
)

# Latencia total del request (incluye LLM si aplica)
query_latency_total = Histogram(
    "rag_query_latency_ms",
    "Full RAG query latency in milliseconds",
    ["endpoint"],
    buckets=[50, 100, 200, 500, 1000, 2000, 5000, 10000],
)

2.2 Ingestion throughput

Documentos y vectores ingeridos por unidad de tiempo:

ingestion_documents_total = Counter(
    "rag_ingestion_documents_total",
    "Total documents ingested",
    ["collection", "status"],
)
ingestion_duration_seconds = Histogram(
    "rag_ingestion_duration_seconds",
    "Time to ingest a batch of documents",
    ["collection"],
    buckets=[1, 5, 10, 30, 60, 120],
)

2.3 Index size

Tamaño del índice y número de vectores:

index_vectors_total = Gauge(
    "rag_index_vectors_total",
    "Number of vectors in the index",
    ["collection"],
)
index_size_bytes = Gauge(
    "rag_index_size_bytes",
    "Approximate index size in bytes",
    ["collection"],
)

2.4 Error rate

Errores por endpoint y por tipo:

query_errors_total = Counter(
    "rag_query_errors_total",
    "Total query errors",
    ["endpoint", "error_type"],
)
query_total = Counter(
    "rag_queries_total",
    "Total queries",
    ["endpoint"],
)

2.5 Cache hit rate (si usas caché)

cache_hits_total = Counter("rag_cache_hits_total", "Cache hits", ["collection"])
cache_misses_total = Counter("rag_cache_misses_total", "Cache misses", ["collection"])

3. Python: decoradores para recolección de métricas

Encapsula la instrumentación en decoradores reutilizables:

import time
from functools import wraps
from prometheus_client import Histogram

retrieval_latency = Histogram(
    "rag_retrieval_latency_ms",
    "Retrieval latency in ms",
    ["collection"],
    buckets=[10, 25, 50, 100, 200, 500, 1000],
)


def track_retrieval_latency(collection: str = "default"):
    def decorator(func):
        @wraps(func)
        async def async_wrapper(*args, **kwargs):
            start = time.perf_counter()
            try:
                result = await func(*args, **kwargs)
                return result
            finally:
                elapsed_ms = (time.perf_counter() - start) * 1000
                retrieval_latency.labels(collection=collection).observe(elapsed_ms)

        @wraps(func)
        def sync_wrapper(*args, **kwargs):
            start = time.perf_counter()
            try:
                return func(*args, **kwargs)
            finally:
                elapsed_ms = (time.perf_counter() - start) * 1000
                retrieval_latency.labels(collection=collection).observe(elapsed_ms)

        if asyncio.iscoroutinefunction(func):
            return async_wrapper
        return sync_wrapper

    return decorator

Uso:

@track_retrieval_latency(collection="docs")
async def search(query: str, top_k: int = 5):
    return await chroma_collection.query(query_texts=[query], n_results=top_k)

4. FastAPI: middleware para request tracking

Middleware que mide latencia y errores por ruta:

from fastapi import FastAPI, Request
from prometheus_client import Histogram, Counter
import time

request_latency = Histogram(
    "rag_http_request_latency_seconds",
    "HTTP request latency",
    ["method", "path", "status"],
    buckets=[0.01, 0.05, 0.1, 0.25, 0.5, 1.0, 2.5, 5.0],
)
request_total = Counter(
    "rag_http_requests_total",
    "Total HTTP requests",
    ["method", "path", "status"],
)


@app.middleware("http")
async def metrics_middleware(request: Request, call_next):
    start = time.perf_counter()
    response = await call_next(request)
    elapsed = time.perf_counter() - start
    path = request.scope.get("path", "unknown")
    method = request.method
    status = response.status_code
    request_latency.labels(method=method, path=path, status=str(status)).observe(elapsed)
    request_total.labels(method=method, path=path, status=str(status)).inc()
    return response

5. Grafana: configuración de dashboard

5.1 Queries PromQL esenciales

# p95 latencia retrieval (ms)
histogram_quantile(0.95, sum(rate(rag_retrieval_latency_ms_bucket[5m])) by (le, collection))

# p50, p95, p99
histogram_quantile(0.50, sum(rate(rag_retrieval_latency_ms_bucket[5m])) by (le))
histogram_quantile(0.95, sum(rate(rag_retrieval_latency_ms_bucket[5m])) by (le))
histogram_quantile(0.99, sum(rate(rag_retrieval_latency_ms_bucket[5m])) by (le))

# Throughput queries/seg
sum(rate(rag_queries_total[1m]))

# Error rate (%)
100 * sum(rate(rag_query_errors_total[5m])) / sum(rate(rag_queries_total[5m]))

# Cache hit rate (%)
100 * sum(rate(rag_cache_hits_total[5m])) / (sum(rate(rag_cache_hits_total[5m])) + sum(rate(rag_cache_misses_total[5m])))

# Ingestion throughput docs/seg
sum(rate(rag_ingestion_documents_total[5m]))

5.2 Panel JSON para Grafana (esqueleto)

Guarda esto como rag-dashboard.json e impórtalo en Grafana:

{
  "dashboard": {
    "title": "RAG Observability",
    "panels": [
      {
        "title": "Retrieval Latency p50/p95/p99",
        "type": "timeseries",
        "targets": [
          {"expr": "histogram_quantile(0.95, sum(rate(rag_retrieval_latency_ms_bucket[5m])) by (le))", "legendFormat": "p95"},
          {"expr": "histogram_quantile(0.50, sum(rate(rag_retrieval_latency_ms_bucket[5m])) by (le))", "legendFormat": "p50"},
          {"expr": "histogram_quantile(0.99, sum(rate(rag_retrieval_latency_ms_bucket[5m])) by (le))", "legendFormat": "p99"}
        ],
        "fieldConfig": {"defaults": {"unit": "ms"}}
      },
      {
        "title": "Query Throughput",
        "type": "timeseries",
        "targets": [{"expr": "sum(rate(rag_queries_total[1m]))", "legendFormat": "queries/s"}]
      },
      {
        "title": "Error Rate (%)",
        "type": "timeseries",
        "targets": [
          {"expr": "100 * sum(rate(rag_query_errors_total[5m])) / sum(rate(rag_queries_total[5m]))", "legendFormat": "error %"}
        ]
      },
      {
        "title": "Index Vectors",
        "type": "stat",
        "targets": [{"expr": "rag_index_vectors_total", "legendFormat": "vectors"}]
      }
    ]
  }
}

5.3 Layout recomendado

1. Fila 1: Latencia p50/p95/p99 (líneas).
2. Fila 2: Throughput queries/s y error rate (%).
3. Fila 3: Index size, cache hit rate, ingestion throughput.
4. Fila 4: Errores por tipo (tabla o breakdown).

6. Alerting: umbrales concretos

6.1 Reglas Prometheus (ejemplo)

groups:
  - name: rag_alerts
    rules:
      # Warning: p95 > 200ms durante 5 min
      - alert: RAGHighLatencyWarning
        expr: histogram_quantile(0.95, sum(rate(rag_retrieval_latency_ms_bucket[5m])) by (le)) > 200
        for: 5m
        labels:
          severity: warning
        annotations:
          summary: "RAG retrieval p95 > 200ms"

      # Critical: p95 > 500ms durante 5 min
      - alert: RAGHighLatencyCritical
        expr: histogram_quantile(0.95, sum(rate(rag_retrieval_latency_ms_bucket[5m])) by (le)) > 500
        for: 5m
        labels:
          severity: critical
        annotations:
          summary: "RAG retrieval p95 > 500ms - degradación severa"

      # Critical: error rate > 2% durante 5 min
      - alert: RAGHighErrorRate
        expr: 100 * sum(rate(rag_query_errors_total[5m])) / sum(rate(rag_queries_total[5m])) > 2
        for: 5m
        labels:
          severity: critical
        annotations:
          summary: "RAG error rate > 2%"

      # Warning: cache hit rate cae >15% vs baseline (requiere recording rule de baseline)
      - alert: RAGCacheHitRateDrop
        expr: (rag_cache_hit_rate - rag_cache_hit_rate_baseline) < -15
        for: 10m
        labels:
          severity: warning

6.2 Tabla rápida de umbrales

7. Logging estructurado en JSON (Python)

Logs estructurados facilitan búsqueda en Elasticsearch, Loki o CloudWatch.

7.1 Configuración con structlog

import structlog
import logging
import json

def configure_structured_logging():
    structlog.configure(
        processors=[
            structlog.stdlib.filter_by_level,
            structlog.stdlib.add_logger_name,
            structlog.stdlib.add_log_level,
            structlog.stdlib.PositionalArgumentsFormatter(),
            structlog.processors.TimeStamper(fmt="iso"),
            structlog.processors.StackInfoRenderer(),
            structlog.processors.format_exc_info,
            structlog.processors.UnicodeDecoder(),
            structlog.processors.JSONRenderer(),
        ],
        context_class=dict,
        logger_factory=structlog.stdlib.LoggerFactory(),
        wrapper_class=structlog.stdlib.BoundLogger,
        cache_logger_on_first_use=True,
    )

7.2 Uso en endpoints RAG

logger = structlog.get_logger()

async def rag_query_endpoint(query: str):
    log = logger.bind(
        endpoint="rag_query",
        query_id=str(uuid.uuid4()),
        query_length=len(query),
    )
    try:
        results = await retrieve_and_generate(query)
        log.info(
            "query_completed",
            latency_ms=results.get("latency_ms"),
            num_results=len(results.get("documents", [])),
        )
        return results
    except Exception as e:
        log.error("query_failed", error=str(e), error_type=type(e).__name__, exc_info=True)
        raise

7.3 Ejemplo de salida JSON

{
  "event": "query_completed",
  "endpoint": "rag_query",
  "query_id": "a1b2c3d4-...",
  "query_length": 42,
  "latency_ms": 156,
  "num_results": 5,
  "timestamp": "2026-03-13T14:32:01.123Z",
  "level": "info",
  "logger": "app.routes"
}

8. Runbook templates

8.1 Runbook: RAGHighLatencyCritical

Condición: p95 retrieval > 500 ms durante 5 min.

Pasos:

1. Confirmar alerta: Revisa Grafana. ¿Es regresión real o pico puntual?
2. Identificar componente:
   • ¿Embedding API lenta? Revisa latencia de llamadas a OpenAI/Cohere.
   • ¿Chroma/vector DB? Revisa CPU, memoria, disco del contenedor.
   • ¿Cantidad de vectores? Si index_vectors_total subió mucho, puede ser reindexing.
3. Mitigación temporal:
   • Rate limit si hay saturación.
   • Aumentar réplicas del servicio RAG.
   • Activar/verificar caché para queries frecuentes.
4. Post-incidente: Registrar en doc de incidentes, actualizar umbrales si hace falta.

8.2 Runbook: RAGHighErrorRate

Condición: Error rate > 2% durante 5 min.

Pasos:

1. Confirmar: Revisar rag_query_errors_total por error_type.
2. Clasificar errores:
   • timeout → Revisar timeouts de embedding/LLM, aumentar o escalar.
   • connection_refused → Revisar salud de Chroma/Postgres.
   • validation_error → Revisar logs, posible input malformado.
3. Mitigación:
   • Circuit breaker si hay dependencia externa caída.
   • Rollback de deploy reciente si correlaciona.
4. Comunicación: Si afecta usuarios, notificar según proceso de incidentes.

8.3 Runbook genérico (plantilla)

## [NOMBRE_ALERTA]

**Condición:** [expresión o descripción]
**Severidad:** warning | critical

### 1. Confirmación
- [ ] Revisar dashboards
- [ ] Verificar que no sea falso positivo

### 2. Diagnóstico
- [ ] Identificar componente afectado
- [ ] Revisar logs y métricas recientes

### 3. Mitigación
- [ ] Acción inmediata
- [ ] Acción de corto plazo

### 4. Resolución
- [ ] Causa raíz (si aplica)
- [ ] Acciones preventivas

### 5. Post-mortem
- [ ] Documentar en [enlace]

9. Troubleshooting de observabilidad

9.1 "Tenemos muchas métricas pero no son accionables"

Problema: Dashboards llenos de gráficos que nadie usa para decidir.

Solución: Reduce a métricas ligadas a objetivos de producto o SLA. Por ejemplo: p95 retrieval para "el usuario no espera más de 300 ms", error rate para "menos del 1% de errores". Oculta o archive el resto hasta que tengas runbooks para cada alerta.

9.2 "Alertas constantes que nadie atiende"

Problema: Fatiga de alertas; el equipo ignora notificaciones.

Solución: Sube umbrales o amplía ventanas (por ejemplo, for: 10m en vez de 5m). Agrupa alertas relacionadas. Solo crea alertas que tengan un runbook definido y un dueño.

9.3 "No sabemos si mejoramos"

Problema: Cambios en código o índices sin forma objetiva de comparar.

Solución: Define baselines (p.ej. p95 en semana X) y compáralos siempre. Usa dashboards con overlays de períodos anteriores. Documenta qué baseline usas para cada release.

9.4 "Los percentiles no cuadran con lo que veo en logs"

Problema: p95 en Prometheus distinto al "peor caso" que ves en logs.

Solución: Los percentiles son sobre una ventana temporal (p.ej. 5 min). Un spike corto puede no verse bien en percentiles. Revisa buckets del Histogram y añade más si hace falta (p.ej. 200, 500, 1000 ms). Usa también p99 para capturar outliers.

9.5 "No tenemos budget para Grafana Cloud / Datadog"

Problema: Stack comercial caro para equipo pequeño.

Solución: Prometheus + Grafana self-hosted + Alertmanager son gratuitos. Para logs, Loki (Grafana) o Elasticsearch básico. Empieza con lo mínimo (Prometheus + Grafana en Docker Compose) y escala cuando lo necesites.

10. Ejercicios

Ejercicio 1: Implementar decorador de métricas

Objetivo: Crear un decorador @track_latency(metric_name="my_metric") que registre latencia en un Histogram de Prometheus.

from functools import wraps
import time
from prometheus_client import Histogram

def track_latency(metric_name: str = "custom_latency", buckets=None):
    buckets = buckets or [0.01, 0.05, 0.1, 0.25, 0.5, 1.0]
    hist = Histogram(metric_name + "_seconds", "Latency", buckets=buckets)

    def decorator(func):
        @wraps(func)
        def wrapper(*args, **kwargs):
            start = time.perf_counter()
            try:
                return func(*args, **kwargs)
            finally:
                hist.observe(time.perf_counter() - start)
        return wrapper
    return decorator

# Uso:
@track_latency("my_retrieval")
def my_search():
    ...

Ejercicio 2: Middleware de latencia por ruta

Objetivo: Añadir middleware FastAPI que registre latencia por ruta en un Histogram con labels path y method.

from fastapi import Request
from prometheus_client import Histogram
import time

latency = Histogram("http_latency", "HTTP latency", ["method", "path"], buckets=[0.05, 0.1, 0.25, 0.5, 1.0])

@app.middleware("http")
async def add_latency(request: Request, call_next):
    start = time.perf_counter()
    response = await call_next(request)
    latency.labels(method=request.method, path=request.url.path).observe(time.perf_counter() - start)
    return response

Ejercicio 3: PromQL para error rate

Objetivo: Escribir una expresión PromQL que calcule el error rate (%) en los últimos 5 minutos, asumiendo rag_queries_total y rag_query_errors_total.

100 * sum(rate(rag_query_errors_total[5m])) / sum(rate(rag_queries_total[5m]))

100 * sum(rate(rag_query_errors_total[5m])) / (sum(rate(rag_queries_total[5m])) or vector(1))

Ejercicio 4: Log estructurado con structlog

Objetivo: Configurar structlog para que cada log incluya request_id, endpoint y salga en JSON.

import structlog

structlog.configure(
    processors=[
        structlog.processors.TimeStamper(fmt="iso"),
        structlog.processors.JSONRenderer(),
    ],
)
log = structlog.get_logger()

# En tu handler:
log = log.bind(request_id=request_id, endpoint="/rag/query")
log.info("request_started", query=query[:50])
# ... después ...
log.info("request_completed", latency_ms=123)

Ejercicio 5: Regla de alerta para ingestion lenta

Objetivo: Crear una regla que alerte si la cola de ingestion tiene más de 1000 documentos pendientes durante 10 minutos. Asume una métrica rag_ingestion_queue_size.

- alert: RAGIngestionBacklog
  expr: rag_ingestion_queue_size > 1000
  for: 10m
  labels:
    severity: warning
  annotations:
    summary: "RAG ingestion backlog > 1000 documents"
    description: "Queue has been above 1000 for 10 minutes. Check ingestion workers and embedding API."

Ejercicio 6: Runbook para "Index size crece muy rápido"

Objetivo: Escribir los pasos de un runbook cuando la métrica rag_index_size_bytes crece más del 20% en 24 horas.

11. Resumen

• Define métricas mínimas: latencia (p50/p95/p99), throughput, error rate, index size, cache hit rate.
• Instrumenta con Prometheus (Histogram, Counter, Gauge) y exponelas en /metrics.
• Usa decoradores Python para capturar latencia de funciones críticas (retrieval, embedding).
• Usa middleware FastAPI para request tracking y latencia por ruta.
• Configura Grafana con PromQL para percentiles, throughput y error rate.
• Define alertas con umbrales concretos: p95 > 200 ms warning, > 500 ms critical; error rate > 2% critical.
• Logging estructurado en JSON (structlog) para trazabilidad y búsqueda en sistemas de logs.
• Escribe runbooks para cada alerta: confirmación, diagnóstico, mitigación y post-mortem.
• Evita fatiga de alertas: solo alertas con runbook y dueño definido.

12. Recursos adicionales

• Prometheus - Overview
• Grafana Documentation
• prometheus_client (Python) - GitHub
• Structlog - Structured Logging for Python
• PromQL Tutorial
• Grafana Alerting
• The Four Golden Signals (Google SRE)
• OpenTelemetry for Python

Tiempo estimado: 25–35 minutos
Siguiente: 04-backup-disaster-recovery.md