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ápsula | Feature central | Impacto |
|---|---|---|---|
| 01 | Introducción | La transformación de print() a structured logging | — |
| 02 | Estrategias de logging LLM | Log levels, qué loguear, privacidad | Qué loguear |
| 03 | Request tracing | Correlation IDs, contexto por request | Debuggeabilidad |
| 04 | Token y cost tracking | Calcular y loguear costos | Visibilidad financiera |
| 05 | JSON logs | structlog, procesadores, producción vs dev | Infraestructura |
| 06 | Debugging no-determinístico | Reproducibility context, seed | Reproducibilidad |
| 07 | Proyecto AI Logging System | Integración completa | Proyecto |
| 08 | Resumen y troubleshooting | Cierre | — |
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:
- "¿Cuánto cuesta procesar 1000 requests con mi configuración actual?"
- "¿Cuántos usuarios han tenido errores 500 en la última semana?"
- "¿Qué inputs producen las respuestas más largas (y por tanto más caras)?"
- "¿Cuántas veces se activó el guardrail de PII ayer?"
- "¿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
request_received— inicio del request con metadata básicallm_request_completed— al completar la llamada al LLM (tokens, cost, duration)guardrail_activated— si algún guardrail se activa (injection, PII, content)validation_failed— si Pydantic no puede validar el outputrequest_completed— fin del request con costo total y status code
Eventos adicionales útiles:
input_truncated— si el input fue truncado por el sanitizadorfallback_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
- structlog Documentation — La biblioteca principal de este módulo
- The Twelve-Factor App: Logs — La filosofía de "logs como event streams"
- JSON Lines format — El formato estándar para logs JSON
- jq Tutorial — Para hacer queries a los logs JSON
- OpenTelemetry — Para cuando necesites distributed tracing avanzado
- Observability Engineering (O'Reilly) — El libro de referencia sobre observabilidad