6. Monitoring y Observability

Descripción

Trackear el estado de salud de tu sistema LLM en producción: latencia (p50/p95/p99), tasa de error, costo por request, y calidad del output. Alertas automáticas, dashboards, y herramientas de observabilidad específicas para LLMs.

Por Qué el Monitoring es Imprescindible

Sin monitoring, eres el último en saber cuando algo falla:

Sistema LLM SIN monitoring:
──────────────────────────
Día 1:  Nuevo modelo desplegado, prompt cambió ligeramente
Día 3:  Accuracy bajó del 94% al 71% — nadie lo sabe
Día 7:  Un usuario lo reporta como "respuestas raras"
Día 8:  Investigas — el problema lleva 7 días, miles de usuarios afectados

Sistema LLM CON monitoring:
──────────────────────────
Día 1:  Nuevo modelo desplegado
Día 1, 15 minutos después:  Alerta: accuracy bajó de 94% a 71%
Día 1, 20 minutos después:  Rollback ejecutado
Día 1: Impacto: 15 minutos, algunos cientos de requests

Las 4 Señales de Gold (Google SRE)

Aplicadas a sistemas LLM:

Para LLMs, añadimos una quinta señal:

Implementación: Métricas Core

import time
import threading
from collections import defaultdict, deque
from datetime import datetime
from typing import Optional
from openai import OpenAI

client = OpenAI()


class MetricsCollector:
    """
    Colector de métricas en memoria para sistemas LLM.
    Thread-safe, con ventanas de tiempo deslizantes.
    """
    
    def __init__(self, ventana_segundos: int = 300):  # 5 minutos default
        self.ventana_segundos = ventana_segundos
        self._lock = threading.Lock()
        
        # Almacenamiento de métricas con timestamp
        self._latencias: dict[str, deque] = defaultdict(lambda: deque(maxlen=10000))
        self._errores: dict[str, deque] = defaultdict(lambda: deque(maxlen=10000))
        self._requests: dict[str, deque] = defaultdict(lambda: deque(maxlen=10000))
        self._tokens: dict[str, deque] = defaultdict(lambda: deque(maxlen=10000))
        self._costos: dict[str, deque] = defaultdict(lambda: deque(maxlen=10000))
    
    def _limpiar_ventana(self, cola: deque) -> list:
        """Retorna solo los elementos dentro de la ventana de tiempo."""
        ahora = time.time()
        limite = ahora - self.ventana_segundos
        return [(ts, val) for ts, val in cola if ts > limite]
    
    def registrar_request(
        self,
        prompt_name: str,
        latencia_ms: float,
        exito: bool,
        tokens: int,
        costo: float
    ) -> None:
        """Registra las métricas de un request."""
        ahora = time.time()
        
        with self._lock:
            self._latencias[prompt_name].append((ahora, latencia_ms))
            self._requests[prompt_name].append((ahora, 1))
            self._tokens[prompt_name].append((ahora, tokens))
            self._costos[prompt_name].append((ahora, costo))
            
            if not exito:
                self._errores[prompt_name].append((ahora, 1))
    
    def calcular_percentil(self, valores: list[float], percentil: float) -> float:
        """Calcula el percentil P de una lista de valores."""
        if not valores:
            return 0.0
        
        ordenados = sorted(valores)
        idx = int(len(ordenados) * percentil / 100)
        idx = min(idx, len(ordenados) - 1)
        return ordenados[idx]
    
    def metricas_actuales(self, prompt_name: str) -> dict:
        """Retorna métricas calculadas para la ventana de tiempo actual."""
        with self._lock:
            latencias_recientes = self._limpiar_ventana(self._latencias[prompt_name])
            requests_recientes = self._limpiar_ventana(self._requests[prompt_name])
            errores_recientes = self._limpiar_ventana(self._errores[prompt_name])
            tokens_recientes = self._limpiar_ventana(self._tokens[prompt_name])
            costos_recientes = self._limpiar_ventana(self._costos[prompt_name])
        
        vals_latencia = [v for _, v in latencias_recientes]
        n_requests = len(requests_recientes)
        n_errores = len(errores_recientes)
        
        return {
            "prompt_name": prompt_name,
            "ventana_segundos": self.ventana_segundos,
            "n_requests": n_requests,
            "error_rate": n_errores / n_requests if n_requests > 0 else 0.0,
            "latencia_p50": self.calcular_percentil(vals_latencia, 50),
            "latencia_p95": self.calcular_percentil(vals_latencia, 95),
            "latencia_p99": self.calcular_percentil(vals_latencia, 99),
            "latencia_media": sum(vals_latencia) / len(vals_latencia) if vals_latencia else 0,
            "tokens_total": sum(v for _, v in tokens_recientes),
            "costo_total": sum(v for _, v in costos_recientes),
            "costo_por_request": (
                sum(v for _, v in costos_recientes) / n_requests
                if n_requests > 0 else 0.0
            ),
            "rps": n_requests / self.ventana_segundos  # Requests per second
        }
    
    def resumen_global(self) -> dict:
        """Métricas de todos los prompts."""
        return {
            nombre: self.metricas_actuales(nombre)
            for nombre in self._requests.keys()
        }


# Singleton global
_metrics = MetricsCollector(ventana_segundos=300)


def tracked_call(
    prompt_name: str,
    prompt_template: str,
    input_text: str,
    model: str = "gpt-4o-mini"
) -> dict:
    """
    Wrapper que ejecuta un prompt y registra las métricas automáticamente.
    """
    inicio = time.time()
    exito = True
    tokens = 0
    costo = 0.0
    output = ""
    
    try:
        response = client.chat.completions.create(
            model=model,
            messages=[{
                "role": "user",
                "content": prompt_template.format(input=input_text)
            }],
            temperature=0
        )
        
        output = response.choices[0].message.content
        tokens = response.usage.total_tokens
        
        # Costo (GPT-4o-mini)
        precio_input = 0.15 / 1_000_000
        precio_output = 0.60 / 1_000_000
        costo = (
            response.usage.prompt_tokens * precio_input +
            response.usage.completion_tokens * precio_output
        )
    
    except Exception as e:
        exito = False
        output = f"ERROR: {str(e)}"
    
    finally:
        latencia_ms = (time.time() - inicio) * 1000
        _metrics.registrar_request(
            prompt_name=prompt_name,
            latencia_ms=latencia_ms,
            exito=exito,
            tokens=tokens,
            costo=costo
        )
    
    return {
        "output": output,
        "latencia_ms": latencia_ms,
        "exito": exito,
        "tokens": tokens,
        "costo": costo
    }

Sistema de Alertas

from dataclasses import dataclass


@dataclass
class AlertaConfig:
    """Configuración de umbrales para alertas."""
    # Latencia
    latencia_p95_max_ms: float = 3000.0    # 3 segundos
    latencia_p99_max_ms: float = 5000.0    # 5 segundos
    
    # Errores
    error_rate_max: float = 0.01            # 1%
    
    # Costo
    costo_por_request_max: float = 0.005   # $0.005 por request
    
    # Calidad (requiere evaluation separada)
    accuracy_min: float = 0.85
    accuracy_degradacion_max: float = 0.05  # 5% de degradación vs baseline


class AlertaManager:
    """Gestiona alertas de monitoring con deduplicación."""
    
    def __init__(
        self,
        config: AlertaConfig,
        cooldown_segundos: int = 300  # No repetir misma alerta por 5 minutos
    ):
        self.config = config
        self.cooldown = cooldown_segundos
        self._ultima_alerta: dict[str, float] = {}
        self._handlers: list[callable] = []
    
    def add_handler(self, handler: callable) -> None:
        """Añade un handler para procesar alertas (Slack, email, log, etc.)."""
        self._handlers.append(handler)
    
    def _puede_alertar(self, clave: str) -> bool:
        """Verifica si podemos enviar una alerta (deduplicación)."""
        ahora = time.time()
        ultima = self._ultima_alerta.get(clave, 0)
        
        if ahora - ultima > self.cooldown:
            self._ultima_alerta[clave] = ahora
            return True
        return False
    
    def verificar(self, metrics: dict) -> list[dict]:
        """Verifica las métricas y genera alertas si es necesario."""
        prompt_name = metrics.get("prompt_name", "unknown")
        alertas = []
        
        # Latencia p95
        p95 = metrics.get("latencia_p95", 0)
        if p95 > self.config.latencia_p95_max_ms:
            clave = f"{prompt_name}:latencia_p95"
            if self._puede_alertar(clave):
                alerta = {
                    "tipo": "LATENCIA_ALTA",
                    "severidad": "WARNING",
                    "prompt": prompt_name,
                    "mensaje": f"Latencia p95={p95:.0f}ms supera el umbral {self.config.latencia_p95_max_ms:.0f}ms",
                    "valor": p95,
                    "umbral": self.config.latencia_p95_max_ms
                }
                alertas.append(alerta)
        
        # Error rate
        error_rate = metrics.get("error_rate", 0)
        if error_rate > self.config.error_rate_max:
            clave = f"{prompt_name}:error_rate"
            if self._puede_alertar(clave):
                alertas.append({
                    "tipo": "ERROR_RATE_ALTA",
                    "severidad": "CRITICO" if error_rate > 0.05 else "WARNING",
                    "prompt": prompt_name,
                    "mensaje": f"Error rate={error_rate:.1%} supera umbral {self.config.error_rate_max:.1%}",
                    "valor": error_rate,
                    "umbral": self.config.error_rate_max
                })
        
        # Costo por request
        costo_req = metrics.get("costo_por_request", 0)
        if costo_req > self.config.costo_por_request_max:
            clave = f"{prompt_name}:costo"
            if self._puede_alertar(clave):
                alertas.append({
                    "tipo": "COSTO_ALTO",
                    "severidad": "WARNING",
                    "prompt": prompt_name,
                    "mensaje": f"Costo=${costo_req:.6f}/request supera umbral ${self.config.costo_por_request_max:.6f}",
                    "valor": costo_req,
                    "umbral": self.config.costo_por_request_max
                })
        
        # Disparar handlers
        for alerta in alertas:
            for handler in self._handlers:
                try:
                    handler(alerta)
                except Exception as e:
                    print(f"Error en handler de alerta: {e}")
        
        return alertas
    
    def verificar_periodicamente(self, metrics_collector: MetricsCollector, intervalo: int = 60) -> None:
        """Verifica métricas cada N segundos en background."""
        def loop():
            while True:
                for prompt_name in metrics_collector._requests.keys():
                    metrics = metrics_collector.metricas_actuales(prompt_name)
                    self.verificar(metrics)
                time.sleep(intervalo)
        
        thread = threading.Thread(target=loop, daemon=True)
        thread.start()


# Handlers de ejemplo:
def log_handler(alerta: dict) -> None:
    """Loguea la alerta en consola."""
    severidad = alerta["severidad"]
    emoji = "🚨" if severidad == "CRITICO" else "⚠️"
    print(f"{emoji} [{severidad}] {alerta['mensaje']}")

def slack_handler(alerta: dict, webhook_url: str) -> None:
    """Envía la alerta a Slack."""
    import urllib.request
    import json
    
    texto = f"{'🚨' if alerta['severidad'] == 'CRITICO' else '⚠️'} *{alerta['tipo']}*\n{alerta['mensaje']}"
    payload = {"text": texto}
    
    req = urllib.request.Request(
        webhook_url,
        data=json.dumps(payload).encode(),
        headers={"Content-Type": "application/json"}
    )
    urllib.request.urlopen(req, timeout=5)


# Uso:
config = AlertaConfig(latencia_p95_max_ms=2000, error_rate_max=0.02)
alertas_mgr = AlertaManager(config)
alertas_mgr.add_handler(log_handler)
# alertas_mgr.add_handler(lambda a: slack_handler(a, SLACK_WEBHOOK))
alertas_mgr.verificar_periodicamente(_metrics, intervalo=60)

Prometheus + Grafana Integration

Para equipos con infraestructura de monitoring:

from prometheus_client import (
    Counter, Histogram, Gauge,
    start_http_server, REGISTRY
)

# Definir métricas Prometheus
latencia_histogram = Histogram(
    "llm_request_duration_milliseconds",
    "Latencia de requests a LLM",
    ["prompt_name", "model", "status"],
    buckets=[100, 250, 500, 1000, 2000, 3000, 5000, 10000]
)

requests_counter = Counter(
    "llm_requests_total",
    "Total de requests a LLM",
    ["prompt_name", "model", "status"]
)

tokens_counter = Counter(
    "llm_tokens_total",
    "Total de tokens procesados",
    ["prompt_name", "model", "tipo"]  # tipo: prompt | completion
)

costo_counter = Counter(
    "llm_cost_usd_total",
    "Costo total en USD",
    ["prompt_name", "model"]
)

quality_gauge = Gauge(
    "llm_quality_score",
    "Score de calidad (accuracy) del último evaluation run",
    ["prompt_name"]
)


def tracked_call_prometheus(
    prompt_name: str,
    prompt_template: str,
    input_text: str,
    model: str = "gpt-4o-mini"
) -> dict:
    """Llamada LLM con métricas Prometheus."""
    inicio = time.time()
    status = "success"
    
    try:
        response = client.chat.completions.create(
            model=model,
            messages=[{"role": "user", "content": prompt_template.format(input=input_text)}],
            temperature=0
        )
        output = response.choices[0].message.content
        
        # Registrar tokens
        tokens_counter.labels(
            prompt_name=prompt_name, model=model, tipo="prompt"
        ).inc(response.usage.prompt_tokens)
        
        tokens_counter.labels(
            prompt_name=prompt_name, model=model, tipo="completion"
        ).inc(response.usage.completion_tokens)
        
        # Costo
        costo = (
            response.usage.prompt_tokens * 0.15 / 1e6 +
            response.usage.completion_tokens * 0.60 / 1e6
        )
        costo_counter.labels(prompt_name=prompt_name, model=model).inc(costo)
        
    except Exception as e:
        status = "error"
        output = f"ERROR: {e}"
    
    finally:
        latencia_ms = (time.time() - inicio) * 1000
        
        latencia_histogram.labels(
            prompt_name=prompt_name, model=model, status=status
        ).observe(latencia_ms)
        
        requests_counter.labels(
            prompt_name=prompt_name, model=model, status=status
        ).inc()
    
    return {"output": output, "status": status, "latencia_ms": latencia_ms}


# Exponer métricas en /metrics para Prometheus
def iniciar_metrics_server(puerto: int = 8080) -> None:
    """Inicia servidor HTTP para métricas Prometheus."""
    start_http_server(puerto)
    print(f"Métricas disponibles en http://localhost:{puerto}/metrics")

LangSmith para Observabilidad

LangSmith (de LangChain) ofrece tracing nativo para LLMs:

import os
from langsmith import Client
from langsmith.wrappers import wrap_openai

# Configurar LangSmith
os.environ["LANGCHAIN_TRACING_V2"] = "true"
os.environ["LANGCHAIN_API_KEY"] = "ls__xxx"
os.environ["LANGCHAIN_PROJECT"] = "mi-proyecto-llm"

# Wrap del cliente OpenAI para tracing automático
from openai import OpenAI
wrapped_client = wrap_openai(OpenAI())

def call_con_tracing(prompt: str, input_text: str) -> str:
    """
    Llamada con tracing automático en LangSmith.
    Cada llamada aparece en el dashboard con: latencia, tokens, costo, input/output.
    """
    response = wrapped_client.chat.completions.create(
        model="gpt-4o-mini",
        messages=[{"role": "user", "content": prompt.format(input=input_text)}],
        temperature=0
    )
    return response.choices[0].message.content


# Evaluation integrada en LangSmith:
def evaluar_con_langsmith(
    prompt_name: str,
    golden_set: list[dict],
    prompt_template: str
) -> None:
    """
    Crea un dataset en LangSmith y evalúa el prompt.
    Los resultados aparecen en el dashboard con comparación histórica.
    """
    langsmith_client = Client()
    
    # Crear dataset si no existe
    dataset_name = f"golden_set_{prompt_name}"
    
    try:
        dataset = langsmith_client.create_dataset(dataset_name)
        for ej in golden_set:
            langsmith_client.create_example(
                inputs={"input": ej["input"]},
                outputs={"expected": ej["expected_output"]},
                dataset_id=dataset.id
            )
    except Exception:
        pass  # Dataset ya existe
    
    # La evaluación automática aparece en LangSmith con métricas y trazas

Quality Monitoring: Evaluación Continua en Producción

import random
import asyncio
from openai import AsyncOpenAI

async_client = AsyncOpenAI()

class QualityMonitor:
    """
    Monitorea la calidad del LLM en producción mediante sampling.
    
    En lugar de evaluar cada request (costoso), evalúa un sample aleatorio.
    """
    
    def __init__(
        self,
        baseline_accuracy: float,
        sample_rate: float = 0.05,  # Evaluar 5% del tráfico
        min_sample_size: int = 50   # Mínimo de requests para evaluar
    ):
        self.baseline = baseline_accuracy
        self.sample_rate = sample_rate
        self.min_sample_size = min_sample_size
        self._buffer: list[dict] = []
        self._quality_scores: list[float] = []
    
    def should_sample(self) -> bool:
        """Decide si este request debe ser evaluado."""
        return random.random() < self.sample_rate
    
    def add_to_buffer(self, input_text: str, output: str, expected: str | None = None) -> None:
        """Agrega un request al buffer de evaluación."""
        if self.should_sample():
            self._buffer.append({
                "input": input_text,
                "output": output,
                "expected": expected,
                "timestamp": time.time()
            })
    
    async def evaluar_buffer(self, judge_prompt: str) -> float:
        """
        Evalúa el buffer acumulado con LLM-as-judge.
        Retorna el quality score promedio.
        """
        if len(self._buffer) < self.min_sample_size:
            return None  # No hay suficiente muestra
        
        semaphore = asyncio.Semaphore(5)
        
        async def evaluar_uno(item):
            async with semaphore:
                if item.get("expected"):
                    # Exact match si hay expected
                    return 1.0 if item["output"].strip().lower() == item["expected"].strip().lower() else 0.0
                else:
                    # LLM-as-judge si no hay expected
                    prompt = judge_prompt.format(
                        input=item["input"],
                        output=item["output"]
                    )
                    r = await async_client.chat.completions.create(
                        model="gpt-4o-mini",
                        messages=[{"role": "user", "content": prompt}],
                        temperature=0,
                        max_tokens=5
                    )
                    raw = r.choices[0].message.content.strip()
                    try:
                        return float(raw.split()[0]) / 10.0
                    except:
                        return 0.5
        
        scores = await asyncio.gather(*[evaluar_uno(item) for item in self._buffer])
        quality = sum(scores) / len(scores)
        
        self._quality_scores.append(quality)
        self._buffer.clear()
        
        # Verificar degradación vs baseline
        degradacion = self.baseline - quality
        if degradacion > 0.05:  # > 5% degradación
            return quality, f"⚠️ Degradación de calidad: {quality:.2%} vs baseline {self.baseline:.2%}"
        
        return quality, "✅ Calidad dentro del rango normal"

Dashboard de Texto para Terminal

def generar_dashboard_terminal(
    metrics_collector: MetricsCollector,
    prompt_names: list[str]
) -> str:
    """
    Genera un dashboard de texto para visualizar en terminal.
    Útil para debugging rápido.
    """
    lineas = [
        "╔══════════════════════════════════════════════════════════╗",
        "║          LLM MONITORING DASHBOARD                        ║",
        f"║  Actualizado: {datetime.now().strftime('%Y-%m-%d %H:%M:%S')}                         ║",
        "╠══════════════════════════════════════════════════════════╣",
    ]
    
    for prompt_name in prompt_names:
        m = metrics_collector.metricas_actuales(prompt_name)
        
        error_rate = m['error_rate']
        p95 = m['latencia_p95']
        
        # Indicadores de estado
        latencia_ok = "✅" if p95 < 2000 else "⚠️" if p95 < 4000 else "❌"
        error_ok = "✅" if error_rate < 0.01 else "⚠️" if error_rate < 0.05 else "❌"
        
        lineas.extend([
            f"║  📊 {prompt_name[:30]:30s}                ║",
            f"║     Requests: {m['n_requests']:6d}  |  RPS: {m['rps']:.2f}              ║",
            f"║     {latencia_ok} Latencia p50/p95/p99: {m['latencia_p50']:.0f}/{p95:.0f}/{m['latencia_p99']:.0f}ms    ║",
            f"║     {error_ok} Error rate: {error_rate:.2%}                        ║",
            f"║     💰 Costo/req: ${m['costo_por_request']:.6f}                    ║",
            "║  ─────────────────────────────────────────────────────║",
        ])
    
    lineas.append("╚══════════════════════════════════════════════════════════╝")
    return "\n".join(lineas)


# Auto-refresh cada 30 segundos:
def dashboard_loop(metrics_collector: MetricsCollector, prompt_names: list[str]) -> None:
    """Loop que refresca el dashboard en terminal."""
    import os
    
    while True:
        os.system("clear")  # Limpiar terminal
        print(generar_dashboard_terminal(metrics_collector, prompt_names))
        time.sleep(30)

Logging Estructurado

import logging
import json

class LLMLogger:
    """Logger estructurado para LLMs. Compatible con ELK, Datadog, CloudWatch."""
    
    def __init__(self, name: str = "llm_system"):
        self.logger = logging.getLogger(name)
        handler = logging.StreamHandler()
        handler.setFormatter(logging.Formatter('%(message)s'))
        self.logger.addHandler(handler)
        self.logger.setLevel(logging.INFO)
    
    def log_request(
        self,
        prompt_name: str,
        prompt_version: str,
        input_text: str,
        output: str,
        latencia_ms: float,
        tokens: int,
        costo: float,
        exito: bool,
        request_id: str | None = None
    ) -> None:
        """Log de un request en formato JSON estructurado."""
        log_entry = {
            "timestamp": datetime.now().isoformat(),
            "request_id": request_id,
            "prompt_name": prompt_name,
            "prompt_version": prompt_version,
            "input_preview": input_text[:100],
            "output_preview": output[:200],
            "latencia_ms": latencia_ms,
            "tokens": tokens,
            "costo_usd": costo,
            "exito": exito,
            "nivel": "INFO" if exito else "ERROR"
        }
        
        self.logger.info(json.dumps(log_entry))
    
    def log_alerta(self, tipo: str, prompt_name: str, mensaje: str, detalles: dict) -> None:
        """Log de alerta en formato estructurado."""
        entry = {
            "timestamp": datetime.now().isoformat(),
            "nivel": "ALERTA",
            "tipo": tipo,
            "prompt_name": prompt_name,
            "mensaje": mensaje,
            **detalles
        }
        self.logger.warning(json.dumps(entry))


llm_logger = LLMLogger()

Troubleshooting

Problema 1: Overhead excesivo del monitoring

Síntoma: El monitoring añade 50ms de latencia a cada request.

Causa: Métricas procesadas en el request path.

Solución:

# Usar async/queue para procesar métricas fuera del request path
import asyncio
from queue import Queue

metrics_queue = Queue(maxsize=10000)

def registrar_async(metrics_data: dict) -> None:
    """Agrega métricas a la queue sin bloquear."""
    try:
        metrics_queue.put_nowait(metrics_data)
    except:
        pass  # Si la queue está llena, dropear (prefer losing metrics over latency)

def processor_loop():
    """Procesa métricas en background."""
    while True:
        try:
            data = metrics_queue.get(timeout=1)
            _metrics.registrar_request(**data)
        except:
            pass

import threading
threading.Thread(target=processor_loop, daemon=True).start()

Problema 2: Alertas ruidosas

Síntoma: 50+ alertas al día por picos momentáneos.

Causa: Umbrales evaluados en ventana muy corta.

Solución:

# Usar ventana deslizante más larga y porcentaje de tiempo excedido
def evaluar_umbral_robusto(
    valores: list[float],
    umbral: float,
    pct_tiempo_max: float = 0.05  # Alertar si excede umbral más del 5% del tiempo
) -> bool:
    """Solo alerta si el umbral se excede más del pct_tiempo_max."""
    if not valores:
        return False
    
    excedidos = sum(1 for v in valores if v > umbral)
    return excedidos / len(valores) > pct_tiempo_max

Problema 3: Quality monitoring costoso

Síntoma: El sample de quality evaluation duplica el costo.

Solución:

# Reducir sample rate o usar exact match cuando hay expected_output
def calidad_eficiente(buffer: list[dict]) -> float:
    """Prioriza exact match (gratis) sobre LLM-as-judge (costoso)."""
    con_expected = [b for b in buffer if b.get("expected")]
    sin_expected = [b for b in buffer if not b.get("expected")]
    
    scores = []
    
    # Exact match para los que tienen expected (gratis)
    for item in con_expected:
        correct = item["output"].strip().lower() == item["expected"].strip().lower()
        scores.append(1.0 if correct else 0.0)
    
    # LLM-as-judge solo para los sin expected, con sample del 10%
    sample_sin_expected = random.sample(sin_expected, max(1, int(len(sin_expected) * 0.1)))
    # ... evaluar con LLM
    
    return sum(scores) / len(scores) if scores else 0.5

Ejercicios

Ejercicio 1: Implementar un tracker básico de latencia

Crea una clase que trackee la latencia de llamadas LLM y calcule p50 y p95:

from collections import deque
from openai import OpenAI
import time

client = OpenAI()

class LatencyTracker:
    def __init__(self, max_samples: int = 1000):
        self._latencias = deque(maxlen=max_samples)
    
    def registrar(self, latencia_ms: float):
        self._latencias.append(latencia_ms)
    
    def percentil(self, p: float) -> float:
        if not self._latencias:
            return 0.0
        ordenadas = sorted(self._latencias)
        idx = min(int(len(ordenadas) * p / 100), len(ordenadas) - 1)
        return ordenadas[idx]
    
    def resumen(self) -> dict:
        return {
            "n": len(self._latencias),
            "p50": f"{self.percentil(50):.0f}ms",
            "p95": f"{self.percentil(95):.0f}ms",
            "p99": f"{self.percentil(99):.0f}ms",
            "promedio": f"{sum(self._latencias)/len(self._latencias):.0f}ms" if self._latencias else "0ms"
        }

tracker = LatencyTracker()

def call_tracked(prompt: str) -> str:
    inicio = time.time()
    r = client.chat.completions.create(
        model="gpt-4o-mini",
        messages=[{"role": "user", "content": prompt}],
        temperature=0
    )
    tracker.registrar((time.time() - inicio) * 1000)
    return r.choices[0].message.content

# Hacer 5 calls y ver estadísticas
for i in range(5):
    call_tracked(f"Di el número {i}")

print(tracker.resumen())

Ejercicio 2: Alerta de error rate

Crea un sistema que alerte cuando la tasa de errores supera el 5% en las últimas 10 llamadas:

from collections import deque
import time

class ErrorRateAlert:
    def __init__(self, ventana: int = 10, umbral: float = 0.05):
        self._resultados = deque(maxlen=ventana)
        self.umbral = umbral
        self._alertas_enviadas = 0
    
    def registrar(self, exito: bool):
        self._resultados.append(1 if exito else 0)
        
        if len(self._resultados) >= 5:  # Mínimo 5 para calcular
            error_rate = 1 - sum(self._resultados) / len(self._resultados)
            
            if error_rate > self.umbral:
                self._alertas_enviadas += 1
                print(f"🚨 ALERTA #{self._alertas_enviadas}: Error rate={error_rate:.1%} > umbral {self.umbral:.1%}")
                print(f"   Últimas {len(self._resultados)} llamadas: {list(self._resultados)}")

alerta = ErrorRateAlert(ventana=10, umbral=0.05)

# Simular llamadas con algunos errores
import random
for i in range(20):
    exito = random.random() > 0.15  # 15% de error rate (supera umbral de 5%)
    alerta.registrar(exito)
    print(f"Request {i+1}: {'OK' if exito else 'ERROR'}")

Resumen

• 4+1 señales: Latencia, tráfico, errores, saturación + calidad (específica de LLM)
• MetricsCollector: Thread-safe, con ventanas deslizantes para métricas en tiempo real
• Alertas: Umbrales con cooldown para evitar ruido — solo alertar cuando importa
• Prometheus: Para equipos con Grafana — histogramas, counters, gauges
• LangSmith: Tracing automático con wrapping del cliente OpenAI
• Quality monitoring: Sampling del 5% del tráfico para evaluar calidad sin costo excesivo
• Logging estructurado: JSON compatible con ELK, Datadog, CloudWatch

Recursos adicionales

1. LangSmith — Observabilidad y tracing para LLMs
2. Prometheus — Sistema de monitoring de código abierto
3. OpenTelemetry — Estándar de instrumentación
4. Datadog LLM Observability — Monitoring específico de LLMs
5. Grafana — Dashboards para Prometheus
6. Langfuse — Open source LLM observability con evaluation