Módulo 5: Structured Logging para AI Systems

8. Resumen y Troubleshooting del Módulo 5

Descripción

Este módulo transformó tu app de "print() como logging" a un sistema de observabilidad que permite responder cualquier pregunta sobre tu app en producción usando solo los logs. Esta cápsula consolida todo lo aprendido en un mapa de decisiones, los errores más comunes con sus soluciones, y un checklist de producción.

Lo que construiste en este módulo

Antes:
  app AI
  └── print(response)  ← "logging"
  
  Debugging: "añado prints y redeployar"
  Costo: "no sé, lo veo en la factura de fin de mes"
  Error en prod: "no sé qué pasó"

Después:
  app AI
  ├── RequestTracingMiddleware   → request_id en todos los logs
  ├── logging_config.py         → structlog con JSON output
  ├── tracing.py                → contextvars (async-safe)
  ├── llm_wrapper.py            → tokens, costo, latencia, alertas
  └── guardrails/               → activaciones logueadas
  
  Debugging: "busco request_id, veo historia completa en 30s"
  Costo: "jq 'cost' logs.json | sum"
  Error en prod: "logs muestran error_type, duration, model, hash"

Mapa de decisiones del módulo

¿Qué debo loguear en este evento?
│
├─ ¿Es el inicio/fin de un request?
│   └─ INFO: request_id, path, method, duration, status_code
│
├─ ¿Es una llamada al LLM?
│   └─ INFO: model, input/output_tokens, cost_usd, duration_ms, finish_reason
│       + WARNING si: latencia > 10s, costo > $0.05, finish_reason == "length"
│       + ERROR si: excepción — con prompt_preview y messages_hash
│
├─ ¿Es la activación de un guardrail?
│   └─ WARNING: guardrail_type, action (blocked/modified), layer
│       NO el input completo (puede ser un ataque o tener PII)
│
├─ ¿Es un error de parsing/validación?
│   └─ WARNING: raw_preview, messages_hash — para correlacionar con la llamada
│
└─ ¿Es un error fatal?
    └─ ERROR + exc_info=True — con todo el stack trace en JSON

¿Con qué nivel?
  └─ DEBUG: prompt completo, respuesta completa — SOLO si debug_mode=True
  └─ INFO: resumen del request, métricas (siempre en prod)
  └─ WARNING: anomalías que requieren atención pero no son fallo
  └─ ERROR: algo falló y requiere intervención

Referencia rápida: las herramientas del módulo

Herramienta	Para qué	Cuándo usar
`structlog.contextvars.bind_contextvars()`	Vincular request_id al contexto	Al inicio de cada request
`structlog.contextvars.clear_contextvars()`	Limpiar al terminar	Al final de cada request (en finally)
`structlog.contextvars.merge_contextvars`	Processor que inyecta el contexto	Siempre en la lista de processors
`structlog.processors.JSONRenderer()`	Output en JSON Lines	En producción
`structlog.dev.ConsoleRenderer()`	Output legible	En desarrollo
`log.bind(key=value)`	Vincular campos a un logger local	Para subtareas con campos adicionales
`log.exception("event")`	Loguear error + stack trace	En except blocks

Los 7 errores más comunes y sus soluciones

Error 1: request_id no aparece en algunos logs

# Síntoma: algunos logs tienen request_id, otros no

# Causa A: El processor merge_contextvars no está configurado
# Fix:
structlog.configure(
    processors=[
        structlog.contextvars.merge_contextvars,  # ← Primero en la lista
        ...
    ]
)

# Causa B: Hay un logger creado ANTES de configurar structlog
# Fix: siempre llamar configure_logging() ANTES de crear cualquier logger
# y usar cache_logger_on_first_use=True (el default)

# Causa C: El contexto se limpió antes de que se logueara
# Fix: asegurarse de que clear_contextvars() está en el finally del middleware,
# no en el try

Error 2: Logs mezclados en async (request_id de otro request)

# Síntoma: el request_id en un log pertenece a otro request concurrente

# Causa: No se usa contextvars — se usa una variable global en su lugar
# ❌ INCORRECTO:
_current_request_id = None  # Variable global (NO funciona en async)

def set_request_id(rid): 
    global _current_request_id
    _current_request_id = rid

# ✅ CORRECTO: usar contextvars
import contextvars
_request_id_var = contextvars.ContextVar("request_id", default=None)

def set_request_id(rid):
    _request_id_var.set(rid)
    structlog.contextvars.bind_contextvars(request_id=rid)

# contextvars es task-scoped en async, no request_id de un coroutine
# contamina a otro

Error 3: Los logs son texto plano en producción, no JSON

# Síntoma: en producción los logs son: "2024-01-15 INFO request_completed"

# Causa: ENVIRONMENT no está seteado, o configure_logging no se llama
# Fix:
# 1. Setear en docker/k8s: ENVIRONMENT=production
# 2. Asegurarse de llamar configure_logging() antes de cualquier log:
configure_logging()  # ← Al inicio de main.py o app factory

# Verificar:
print(os.getenv("ENVIRONMENT"))  # Debe ser "production"

Error 4: cost_usd es 0 o None en los logs

# Síntoma: logs de LLM request sin costo o con costo 0

# Causa A: response.usage es None (algunos endpoints no lo retornan)
# Fix:
if response.usage:
    cost = calculate_cost(model, response.usage.prompt_tokens,
                          response.usage.completion_tokens)
else:
    # Estimar con tiktoken
    input_tokens = count_tokens_in_messages(messages, model)
    cost = calculate_cost(model, input_tokens, 500)  # Estimación
    log.warning("usage_not_available", estimated=True)

# Causa B: El modelo no está en la tabla de precios
# Fix: añadir el modelo a MODEL_PRICING o verificar fallback

Error 5: Logs enormes (>1MB por entrada)

# Síntoma: archivos de logs crecen muy rápido, logs con MB de tamaño

# Causa: Logueando objetos grandes (prompt completo en INFO, 
# respuestas muy largas, objetos Python serializados completos)

# Fix 1: Usar el procesador truncate_long_values
def truncate_long_values(logger, method_name, event_dict, max_len=500):
    for key, value in event_dict.items():
        if isinstance(value, str) and len(value) > max_len:
            event_dict[key] = value[:max_len] + f"...[{len(value)}]"
    return event_dict

# Fix 2: Solo loguear lo que necesitas
# ❌ log.info("response", full_response=response.model_dump())  # Objeto grande
# ✅ log.info("response", tokens=response.usage.total_tokens, cost=cost)

Error 6: PII o secrets en los logs

# Síntoma: emails, API keys, o datos personales aparecen en los logs

# Causa: loguear el input del usuario sin sanitizar
# log.info("request", input=user_text)  # user_text puede tener PII

# Fix 1: Procesador sanitize_sensitive_fields (ya implementado)
# Fix 2: Nunca loguear el input completo en INFO
# ✅ log.info("request", input_length=len(user_text), input_hash=hash(user_text))

# Fix 3: Si necesitas loguear el input (para debugging):
# → Solo en ERROR o DEBUG mode
# → Solo los primeros 200-300 chars
# → Redactar PII primero con el guardrail de PII del Módulo 4

Error 7: structlog no captura logs de librerías externas

# Síntoma: logs de httpx, openai, etc. aparecen en un formato diferente
# o no aparecen en los logs JSON

# Causa: structlog y el logging estándar de Python son sistemas distintos
# Fix: configurar el bridge entre ambos

import logging
import structlog

# Configurar el logging estándar de Python para usar el mismo handler
logging.basicConfig(format="%(message)s", level=logging.INFO)

# Configurar structlog para procesar logs del stdlib también
structlog.configure(
    processors=[
        structlog.stdlib.filter_by_level,
        structlog.stdlib.add_log_level,
        structlog.contextvars.merge_contextvars,
        structlog.processors.TimeStamper(fmt="iso"),
        structlog.processors.JSONRenderer()
    ],
    wrapper_class=structlog.stdlib.BoundLogger,
    logger_factory=structlog.stdlib.LoggerFactory(),  # ← stdlib, no PrintLoggerFactory
)

Árbol de diagnóstico

PROBLEMA: "No puedo debuggear un error en producción"
│
├─ ¿Tienes el request_id?
│   ├─ NO → ¿El response incluye X-Request-ID header?
│   │        ├─ NO → Añadir request_id al response (middleware)
│   │        └─ SÍ → Pedir al usuario que reporte ese header
│   │
│   └─ SÍ → jq 'select(.request_id == "XXX")' logs.json | jq -s 'sort_by(.timestamp)'
│           ├─ ¿Ves todos los pasos del request?
│           │   └─ NO → ¿Están todos los eventos logueados? Revisar gaps en la secuencia
│           └─ ¿Ves el error?
│               └─ NO → ¿El nivel del log es suficientemente bajo? Verificar LOG_LEVEL

PROBLEMA: "Los logs no son JSON en producción"
│
└─ ¿ENVIRONMENT=production está seteado?
    ├─ NO → Setear en docker-compose/k8s env
    └─ SÍ → ¿configure_logging() se llama antes del primer log?
             ├─ NO → Mover configure_logging() a inicio de la app
             └─ SÍ → ¿La app usa cache_logger_on_first_use=True y re-usa loggers viejos?
                      └─ Fix: structlog.reset_defaults() al iniciar (solo en tests)

PROBLEMA: "No sé cuánto gasté ayer en OpenAI"
│
├─ ¿Los logs tienen cost_usd?
│   ├─ NO → ¿call_llm retorna metrics con cost_usd?
│   │        ├─ NO → Implementar calculate_cost() y loguearlo en call_llm
│   │        └─ SÍ → Pero no se usa el wrapper. Usar call_llm() en todos los lugares
│   └─ SÍ → jq -s '[.[].cost_usd // 0] | add' logs.json
│             o python scripts/analyze_logs.py logs/app.json 2024-01-15

PROBLEMA: "El cost_usd es incorrecto"
│
└─ ¿Los precios en MODEL_PRICING están actualizados?
    └─ Verificar en openai.com/pricing
       ├─ Actualizar MODEL_PRICING en logging_config.py
       └─ Añadir test que valida precios conocidos

Checklist de producción del módulo 5

CONFIGURACIÓN
[ ] configure_logging() se llama al inicio de la app
[ ] ENVIRONMENT=production está seteado en producción
[ ] LOG_LEVEL=INFO en producción (DEBUG solo si es necesario)
[ ] LOG_FILE está configurado si quieres logs en archivo
[ ] Log rotation configurada (backupCount=30)

TRACING
[ ] RequestTracingMiddleware registrado en la app FastAPI
[ ] Todos los endpoints usan contextvars para request_id
[ ] Response headers incluyen X-Request-ID
[ ] clear_contextvars() en el finally del middleware

LOGGING EN EL CÓDIGO
[ ] Todos los llm calls pasan por call_llm()
[ ] Todos los logs INFO tienen request_id (verificar con test)
[ ] No hay API keys ni PII en los logs
[ ] Procesador sanitize_sensitive_fields activo

GUARDRAILS
[ ] Activaciones de guardrails logueadas con WARNING
[ ] guardrail_type y action en cada log de guardrail
[ ] No se logua el input completo en WARNING de guardrail

COST TRACKING
[ ] MODEL_PRICING tiene todos los modelos en uso
[ ] cost_usd aparece en cada log de LLM call
[ ] Alertas de high_cost activas
[ ] Script analyze_logs.py disponible para equipo

TESTS
[ ] test: request_id aparece en todos los logs
[ ] test: prompt no aparece en INFO logs
[ ] test: cost_usd está en los logs de LLM call
[ ] test: warnings para finish_reason='length'

Vocabulario del módulo

Término	Definición
Structured logging	Logging que produce datos JSON en vez de texto libre, queryable por máquinas
JSON Lines (JSONL)	Formato donde cada línea es un objeto JSON independiente, ideal para append-only logs
Correlation ID / request_id	Identificador único que conecta todos los logs del lifecycle de un request
contextvars	Módulo de Python que almacena estado por-coroutine (async-safe, thread-safe)
bind_contextvars()	Función de structlog para vincular campos al contexto del coroutine actual
finish_reason	Razón por la que el LLM dejó de generar: "stop", "length", "content_filter"
system_fingerprint	Identificador del servidor/versión del modelo de OpenAI, útil para detectar cambios
reproducibility context	El conjunto de parámetros (model, temperature, seed, prompt) necesarios para re-ejecutar una llamada al LLM
seed	Parámetro que pide al modelo intentar ser reproducible para el mismo input
Log aggregation	Sistema que centraliza logs de múltiples instancias o servicios (ELK, Loki, Datadog)

Conexión con el Módulo 6

El Módulo 6 (Code Quality Patterns para AI) organiza todo el código que has construido hasta ahora. Tienes:

Tests (Módulo 2, 3)
Guardrails (Módulo 4)
Logging (Módulo 5)

El problema es que probablemente estén entrelazados: main.py importa directamente de guardrails, llm_wrapper depende de logging_config, y los tests son difíciles de aislar porque todo está acoplado.

El Módulo 6 aplica clean architecture para separar concerns:

Domain layer: la lógica de negocio (¿qué significa un sentimiento positivo?)
Application layer: los use cases (analize_sentiment)
Infrastructure layer: guardrails, logging, llamadas al API de OpenAI
Interface layer: FastAPI endpoints

Esta separación hace el código mantenible, testeable, y extensible — la transición de "código que funciona" a "código de producción."

Recursos adicionales del módulo

structlog Documentation — La referencia completa
structlog contextvars guide — Async logging
jq Manual — Para queries ad-hoc sobre los logs
OpenTelemetry for Python — Para distributed tracing avanzado
Twelve-Factor App — Logs — La filosofía detrás del módulo
OpenAI Pricing — Precios actualizados para MODEL_PRICING
Loki (Grafana) — Sistema de log aggregation optimizado para JSON logs
Datadog APM — Observabilidad enterprise con soporte para LLM apps

← Anterior7. Proyecto: AI Logging System Siguiente →1. Introducción: Code Quality para AI