Modulo 07: Observabilidad y calidad

Observabilidad: medir lo que importa en tus sistemas con IA

Que un sistema "funcione" no es suficiente. Necesitas saber como funciona, donde falla, cuanto cuesta cada operacion y si la calidad de las respuestas se mantiene. Esto es especialmente critico en sistemas con IA, donde el comportamiento no es determinista.

La observabilidad se sostiene sobre tres pilares:

• Logs: registro de eventos discretos. Que paso, cuando, con que datos de entrada y salida.
• Metricas: valores numericos agregados en el tiempo. Latencia, tokens consumidos, tasa de error, coste por request.
• Trazas: el recorrido completo de una peticion a traves de multiples servicios o agentes. Quien llamo a quien, cuanto tardo cada paso.

En un workflow con agentes IA, cada tool call, cada invocacion al LLM y cada decision del coordinador debe generar datos observables. Sin esto, depurar un error en produccion es como buscar una aguja en un pajar a oscuras.

OpenTelemetry para agentes IA

OpenTelemetry (OTel) es el estandar abierto para instrumentar aplicaciones. Funciona con cualquier lenguaje y se integra con backends como Grafana, Jaeger o Datadog. Para agentes IA, lo usamos para trazar cada tool call con su latencia, modelo utilizado y tokens consumidos.

graph TD
    A[Agent Tool Call] --> B[OpenTelemetry SDK]
    B --> C{Exporters}
    C --> D[Grafana / Prometheus]
    C --> E[Logs estructurados]
    C --> F[Trazas distribuidas]
    D --> G[Dashboard]
    G --> H{Alertas}
    H -->|Error mayor 5%| I[Circuit Breaker]
    H -->|Latencia P95 mayor 3s| J[Escalar]
    H -->|VRAM mayor 90%| K[Notificar]

# Decorador para trazar tool calls de agentes
from opentelemetry import trace
from opentelemetry.trace import StatusCode
import time

tracer = trace.get_tracer("agent-tools")

def traced_tool(tool_name: str):
    def decorator(func):
        def wrapper(*args, **kwargs):
            with tracer.start_as_current_span(tool_name) as span:
                span.set_attribute("tool.name", tool_name)
                span.set_attribute("tool.model", kwargs.get("model", "unknown"))
                start = time.time()
                try:
                    result = func(*args, **kwargs)
                    span.set_attribute("tool.tokens_in", result.get("usage", {}).get("input", 0))
                    span.set_attribute("tool.tokens_out", result.get("usage", {}).get("output", 0))
                    span.set_status(StatusCode.OK)
                    return result
                except Exception as e:
                    span.set_status(StatusCode.ERROR, str(e))
                    span.record_exception(e)
                    raise
                finally:
                    span.set_attribute("tool.latency_ms", (time.time() - start) * 1000)
        return wrapper
    return decorator

@traced_tool("analyze_alert")
def analyze_alert_tool(alert_data: dict, model: str = "qwen3.5-27b"):
    # Logica del tool...
    pass

Cuando un agente coordinador invoca a un worker, y ese worker invoca tools, las trazas se propagan automaticamente. Esto genera un arbol de spans que muestra el flujo completo:

# Traza distribuida (ejemplo visual):
Coordinator (250ms)
  +-- Worker-CTI (180ms)
  |     +-- fetch_ioc_tool (45ms)
  |     +-- llm_analyze (120ms) [tokens: 1200 in, 450 out]
  +-- Worker-SOC (200ms)
        +-- get_alert_context (30ms)
        +-- llm_classify (150ms) [tokens: 800 in, 200 out]

El exportador envia las trazas a Grafana Tempo, donde puedes buscar por trace_id, filtrar por latencia alta o por errores, y correlacionar con logs y metricas.

Evals: testing para IA

Los tests unitarios verifican que el codigo hace lo que debe. Pero con IA, el mismo input puede producir outputs distintos. Por eso necesitas evals: datasets de pares input/output esperado que miden la calidad de las respuestas del modelo.

graph LR
    A[Eval Dataset JSONL] --> B[CI Pipeline]
    B --> C{Accuracy mayor-igual 85%?}
    C -->|Si| D[Deploy]
    C -->|No| E[Bloquear + Notificar]
    D --> F[Monitorizar metricas]
    F --> G{Degradacion?}
    G -->|Si| H[Rollback prompt]
    G -->|No| I[Continuar]

# Eval dataset en formato JSONL (tests/evals/cti/ioc_analysis.jsonl)
{"input": "Analiza el hash: 44d88612fea8a8f36de82e1278abb02f", "expected": "EICAR test file", "tags": ["malware", "hash"]}
{"input": "Clasifica IP 192.168.1.1", "expected": "Private/RFC1918, no threat", "tags": ["ip", "benign"]}
{"input": "Reputacion dominio evil-corp.ru", "expected": "Malicious, phishing campaign", "tags": ["domain", "malicious"]}

Las metricas de evaluacion clave son:

• Accuracy: porcentaje de respuestas correctas frente al expected output.
• Relevance: la respuesta aborda la pregunta (no divaga).
• Faithfulness: la respuesta se basa en datos reales, no en alucinaciones.

# Pipeline de evaluacion en 3 pasos
def run_eval(dataset_path: str, agent_fn, threshold: float = 0.85):
    results = []
    for sample in load_jsonl(dataset_path):
        # Paso 1: Ejecutar el agente
        output = agent_fn(sample["input"])
        # Paso 2: Comparar con expected (LLM-as-judge o heuristicas)
        score = evaluate_response(output, sample["expected"])
        results.append({"input": sample["input"], "score": score})
    # Paso 3: Agregar y decidir
    avg_score = sum(r["score"] for r in results) / len(results)
    passed = avg_score >= threshold
    return {"avg_score": avg_score, "passed": passed, "details": results}

Prompts versionados

Un anti-pattern muy comun es tener prompts inline en el codigo, sin control de cambios. Si cambias un prompt y la calidad baja, no tienes forma de hacer rollback ni de saber que version estaba en produccion.

La solucion: tratar los prompts como artefactos versionados.

# Estructura de directorio para prompts versionados
docs/prompts/
  agent-cti_analyze-ioc_v1.0.txt
  agent-cti_analyze-ioc_v1.1.txt      # Mejora precision en hashes
  agent-cti_analyze-ioc_v2.0.txt      # Cambio mayor: nuevo formato output
  agent-soc_classify-alert_v1.0.txt
  agent-soc_classify-alert_v1.1.txt
  CHANGELOG.md

# CHANGELOG.md
## agent-cti_analyze-ioc
### v2.0 (2026-06-15)
- Output en formato JSON estructurado (breaking change)
- Eval accuracy: 0.92 (vs 0.87 en v1.1)

### v1.1 (2026-06-01)
- Mejor handling de hashes SHA-256
- Eval accuracy: 0.87 (vs 0.83 en v1.0)

### v1.0 (2026-05-15)
- Version inicial
- Eval accuracy: 0.83

Si no puedes responder "que version del prompt esta en produccion ahora mismo", tienes un problema de observabilidad.

Metricas de calidad en produccion

Un dashboard de observabilidad para sistemas con IA debe incluir estas metricas:

• Latencia P50/P95/P99: el 50%, 95% y 99% de las peticiones tardan menos de X milisegundos. P99 es critico para detectar outliers.
• Tokens por request: media y distribucion. Un pico indica prompts que han crecido sin control.
• Coste por request: tokens * precio/token. Permite proyectar gasto mensual.
• Error rate: porcentaje de invocaciones al LLM que fallan (timeout, rate limit, error de formato).
• HITL intervention rate: con que frecuencia un humano tiene que corregir o aprobar la decision del agente.

# Ejemplo de configuracion de alertas
alertas:
  - nombre: "Circuit breaker: error rate alto"
    condicion: error_rate > 5%
    duracion: 5 minutos
    accion: pausar agente, notificar Telegram

  - nombre: "VRAM critica"
    condicion: gpu_vram_usage > 90%
    duracion: 2 minutos
    accion: rechazar nuevas peticiones, alerta urgente

  - nombre: "Latencia P99 degradada"
    condicion: latency_p99 > 10000ms
    duracion: 10 minutos
    accion: escalar a N2, revisar modelo/batch size

  - nombre: "Coste diario excedido"
    condicion: daily_cost > budget_limit
    duracion: inmediato
    accion: activar modelo mas barato, alerta a finance

CI/CD con calidad IA integrada

El pipeline de CI/CD para proyectos con IA tiene un paso extra respecto al desarrollo tradicional: las evals. El flujo completo es:

# Pipeline CI/CD con gate de calidad IA
# .github/workflows/deploy.yml

name: Deploy con eval gate
on:
  push:
    branches: [main]

jobs:
  quality:
    runs-on: ubuntu-latest
    steps:
      - uses: actions/checkout@v4

      # Paso 1: Lint y formato
      - name: Lint
        run: ruff check src/

      # Paso 2: Tests unitarios clasicos
      - name: Tests
        run: pytest tests/unit/ -v

      # Paso 3: Evals de agentes (EL GATE CRITICO)
      - name: Eval agentes
        run: |
          python scripts/run_evals.py \
            --dataset tests/evals/ \
            --threshold 0.85 \
            --output eval-results.json
        env:
          VLLM_ENDPOINT: ${{ secrets.VLLM_ENDPOINT }}

      # Paso 4: Deploy solo si pasa
      - name: Deploy
        if: success()
        run: ./scripts/deploy.sh

Despues del deploy, la monitorizacion continua. Si las metricas en produccion se degradan, un canary deployment permite hacer rollback automatico:

• Canary para prompts: enviar el 10% del trafico al nuevo prompt. Comparar metricas durante 1 hora. Si accuracy baja, rollback automatico.
• Monitorizacion post-deploy: las primeras 2 horas tras deploy son criticas. Alertas mas sensibles (threshold mas bajo).
• Rollback rapido: volver al prompt anterior es un cambio de fichero, no un deploy completo. Por eso los prompts versionados son tan importantes.

El mejor deploy es el que puedes deshacer en 30 segundos. Prompts versionados + feature flags + canary = control total.