Módulo 5: Structured Logging para AI Systems

1. Introducción: Structured Logging para AI

Descripción

El estado actual de muchas apps AI en producción: print(response) como "logging". Cuando algo falla a las 3am, el debugging es añadir más prints y redeployar. Structured logging reemplaza eso con datos queryables que te permiten responder cualquier pregunta sobre lo que pasó en tu sistema — sin tocar el código. Esta cápsula introduce el concepto, la diferencia radical con logging tradicional, y por qué es especialmente crítico para sistemas AI.


La situación sin structured logging

# App AI típica sin logging profesional:
def analyze_sentiment(text: str) -> dict:
    print(f"Processing: {text[:50]}...")  # "Logging"
    
    response = client.chat.completions.create(
        model="gpt-4o-mini",
        messages=[{"role": "user", "content": text}]
    )
    
    result = parse_response(response)
    print(f"Done: {result['sentiment']}")  # "Logging"
    return result

# Cuando algo falla en producción:
# - ¿Cuánto tiempo tardó? No sé
# - ¿Cuántos tokens usó? No sé
# - ¿Cuánto costó? No sé
# - ¿Qué input causó el error? No sé
# - ¿Cuántas veces pasó? No sé
# - ¿Se activó algún guardrail? No sé

# El "debugging" es:
# 1. "Déjame añadir más prints"
# 2. Deploy
# 3. Esperar que vuelva a pasar
# 4. Ver los prints
# Latencia del debugging: horas o días

La situación con structured logging

# La misma app con structured logging:
def analyze_sentiment(text: str, request_id: str) -> dict:
    log = structlog.get_logger().bind(request_id=request_id)
    start = time.time()
    
    log.info("llm_request_started", text_length=len(text))
    
    response = client.chat.completions.create(
        model="gpt-4o-mini",
        messages=[{"role": "user", "content": text}]
    )
    
    duration_ms = (time.time() - start) * 1000
    cost = calculate_cost(response.usage)
    
    log.info("llm_request_completed",
        model="gpt-4o-mini",
        input_tokens=response.usage.prompt_tokens,
        output_tokens=response.usage.completion_tokens,
        duration_ms=duration_ms,
        cost_usd=cost
    )
    
    result = parse_response(response)
    return result

# Cuando algo falla en producción:
# → "request_id=abc123 tiene error 500 las 3am"
# → grep "abc123" logs.json
# {"event": "llm_request_started", "request_id": "abc123", "text_length": 4500}
# {"event": "llm_request_completed", "request_id": "abc123", "duration_ms": 15234}  ← 15 segundos!
# {"event": "pydantic_validation_failed", "request_id": "abc123", "error": "truncated JSON"}
# Historia completa en 30 segundos.

Logging tradicional vs structured: la diferencia fundamental

# ─── Logging TRADICIONAL ──────────────────────────────────────────
import logging
logger = logging.getLogger(__name__)

logger.info("Request processed in 2.3 seconds by gpt-4o-mini, 450 tokens")
# Output: "2024-01-15 10:30:00 INFO Request processed in 2.3 seconds by gpt-4o-mini, 450 tokens"

# Para buscar todos los requests que tardaron más de 10s:
# → grep con regex complicado
# → No puedes hacer sum de duraciones
# → No puedes ordenar por tokens
# → No puedes filtrar por modelo

# ─── Structured LOGGING ───────────────────────────────────────────
import structlog
log = structlog.get_logger()

log.info("request_completed",
    duration_ms=2300,
    model="gpt-4o-mini",
    tokens=450,
    cost_usd=0.000068
)
# Output: {"event": "request_completed", "duration_ms": 2300, "model": "gpt-4o-mini",
#           "tokens": 450, "cost_usd": 0.000068, "timestamp": "2024-01-15T10:30:00Z"}

# Para buscar requests lentos:
# jq 'select(.duration_ms > 10000)' logs.json ← trivial

# Para calcular el costo total del día:
# jq '[.cost_usd] | add' logs.json ← una línea

# Para encontrar el modelo más caro:
# jq -r 'select(.cost_usd > 0.1) | .model' logs.json ← trivial

Por qué structured logging es especialmente crítico para AI

En apps web tradicionales, los fallos son determinísticos: mismo input → mismo output → mismo error. En apps AI, los fallos son no-determinísticos:

Problema 1: Un usuario reporta "la respuesta fue muy rara"
Sin logging: "¿Cuál era exactamente su pregunta? ¿Qué modelo usamos? ¿Qué temperatura?"
Con logging: buscar request_id, ver prompt exacto, model, temperature, output, duration

Problema 2: Factura de OpenAI inesperadamente alta este mes
Sin logging: "No sé qué está generando tanto costo"
Con logging: query por cost_usd > 0.05, encontrar que un edge case genera prompts enormes

Problema 3: El guardrail de injection detectó muchos ataques ayer
Sin logging: No tenías forma de saberlo
Con logging: alert automático cuando guardrail_activations > 10/hora, investigar con request_ids

Problema 4: Migración de gpt-4o a gpt-4o-mini
Sin logging: "No sé si el nuevo modelo es igual de bueno"
Con logging: comparar métricas antes y después: avg_duration_ms, avg_cost_usd, error_rate

Las 4 dimensiones del logging AI

Dimensión 1: REQUEST LIFECYCLE
  ┌──────────────────────────────────────────────────────┐
  │ request_id = "abc123" (conecta todos los eventos)    │
  │                                                      │
  │ input_received → sanitized → injection_check →      │
  │ llm_called → llm_responded → validated →            │
  │ pii_redacted → output_sent                          │
  └──────────────────────────────────────────────────────┘

Dimensión 2: PERFORMANCE
  - duration_ms (cuánto tarda cada paso)
  - queue_time_ms (tiempo esperando)
  - llm_latency_ms (cuánto tarda el LLM)

Dimensión 3: COST
  - input_tokens, output_tokens
  - cost_usd (calculado)
  - model usado

Dimensión 4: GUARDRAILS
  - guardrail activado o no
  - tipo de guardrail
  - acción tomada (blocked/modified)

Preguntas que structured logging permite responder

Con un sistema de logging bien configurado, estas son las consultas que puedes hacer sobre tu app en producción:

# ¿Cuál es el costo total de hoy?
jq -s '[.[].cost_usd // 0] | add' logs.json

# ¿Cuáles son los 10 requests más caros?
jq -s 'sort_by(-.cost_usd) | .[0:10] | .[] | {request_id, cost_usd, duration_ms}' logs.json

# ¿Cuántos requests fallaron en la última hora?
jq 'select(.level == "error" and .timestamp > "2024-01-15T09:00:00Z")' logs.json | wc -l

# ¿Cuántos guardrails se activaron hoy?
jq 'select(.event == "guardrail_activated")' logs.json | wc -l

# ¿Cuál es la latencia promedio por modelo?
jq -s 'group_by(.model) | map({model: .[0].model, avg_ms: ([.[].duration_ms] | add / length)})' logs.json

# ¿Qué request causó el error de las 3am?
jq 'select(.level == "error" and (.timestamp | startswith("2024-01-15T03:")))' logs.json

La transformación mental del módulo

Antes:
  → "Tengo un error en producción, voy a añadir prints y redeployar"
  → "No sé cuánto cuesta cada request"
  → "Un usuario reportó algo raro, no puedo reproducirlo"

Después:
  → "Tengo el request_id del error, busco en logs y veo todo el stack"
  → "Sé exactamente cuánto cuesta cada modelo y cada endpoint"
  → "Con el request_id reproduzco el problema en 2 minutos"

Prerequisitos del módulo

# Dependencias principales
pip install structlog python-json-logger

# Opcional pero recomendado
pip install colorama  # Para colores en ConsoleRenderer (desarrollo)

Roadmap del módulo

#CápsulaFeature centralImpacto
01IntroducciónLa transformación de print() a structured logging
02Estrategias de logging LLMLog levels, qué loguear, privacidadQué loguear
03Request tracingCorrelation IDs, contexto por requestDebuggeabilidad
04Token y cost trackingCalcular y loguear costosVisibilidad financiera
05JSON logsstructlog, procesadores, producción vs devInfraestructura
06Debugging no-determinísticoReproducibility context, seedReproducibilidad
07Proyecto AI Logging SystemIntegración completaProyecto
08Resumen y troubleshootingCierre

Ejercicios

Ejercicio 1: El costo de NO tener logging

Para tu app actual, lista 3 preguntas que NO puedes responder sin structured logging:

Ver guía

Ejemplos:

  1. "¿Cuánto cuesta procesar 1000 requests con mi configuración actual?"
  2. "¿Cuántos usuarios han tenido errores 500 en la última semana?"
  3. "¿Qué inputs producen las respuestas más largas (y por tanto más caras)?"
  4. "¿Cuántas veces se activó el guardrail de PII ayer?"
  5. "¿Cuál es la distribución de latencias de las llamadas al LLM?"

Ejercicio 2: Diseñar el schema de un log de request

Define los campos que debe tener el log de un request completado en tu app:

Ver solución
# Schema del log de request completado:
{
    # Identificación
    "event": "request_completed",
    "request_id": "abc12345",          # UUID corto, único por request
    "timestamp": "2024-01-15T10:30:00Z",
    
    # Request info
    "endpoint": "/analyze",
    "method": "POST",
    
    # LLM info
    "model": "gpt-4o-mini",
    "input_tokens": 245,
    "output_tokens": 87,
    "cost_usd": 0.0000888,
    
    # Performance
    "duration_ms": 1240,
    "llm_latency_ms": 1150,
    
    # Resultado
    "level": "info",
    "status_code": 200,
    
    # Guardrails
    "guardrails_activated": [],     # Lista de guardrails que se activaron
    
    # Opcional (solo en debug o error)
    # "prompt_hash": "abc123ef",    # Hash del prompt para reproducibilidad
}

Ejercicio 3: Convertir un print() a structured log

Convierte este código a structured logging:

print(f"User {user_id} request took {duration}s, used {tokens} tokens")
Ver solución
import structlog
log = structlog.get_logger()

log.info(
    "request_completed",
    user_id=hash(user_id),      # Hash para privacidad
    duration_ms=int(duration * 1000),
    tokens_used=tokens,
    request_id=request_id
)
# Output JSON: {"event": "request_completed", "user_id": 1234567,
#               "duration_ms": 2300, "tokens_used": 450, "request_id": "abc123"}

Ejercicio 4: Identificar qué loguear en tu app

Para el endpoint /analyze de la app de sentimiento, lista los 5 eventos más importantes que deberías loguear:

Ver guía
  1. request_received — inicio del request con metadata básica
  2. llm_request_completed — al completar la llamada al LLM (tokens, cost, duration)
  3. guardrail_activated — si algún guardrail se activa (injection, PII, content)
  4. validation_failed — si Pydantic no puede validar el output
  5. request_completed — fin del request con costo total y status code

Eventos adicionales útiles:

  • input_truncated — si el input fue truncado por el sanitizador
  • fallback_used — si se usó el output por defecto en lugar del LLM

Resumen

  • Structured logging = datos queryables, no texto libre
  • La transformación: de print() a logs que responden preguntas en 30 segundos
  • 4 dimensiones críticas para AI: lifecycle del request, performance, cost, guardrails
  • Correlation IDs son el hilo conductor: todo conectado por request_id
  • No es overhead — es infraestructura: como tests o guardrails, no es opcional para producción

Recursos adicionales

  1. structlog Documentation — La biblioteca principal de este módulo
  2. The Twelve-Factor App: Logs — La filosofía de "logs como event streams"
  3. JSON Lines format — El formato estándar para logs JSON
  4. jq Tutorial — Para hacer queries a los logs JSON
  5. OpenTelemetry — Para cuando necesites distributed tracing avanzado
  6. Observability Engineering (O'Reilly) — El libro de referencia sobre observabilidad