2. Estrategias de Logging para LLM

Descripción

No todo debe loguearse — y no todo debe loguearse con el mismo nivel. Esta cápsula define la estrategia completa de logging para apps LLM: qué loguear en cada nivel (DEBUG/INFO/WARNING/ERROR), por qué el full prompt no debe loguearse en producción por defecto, cómo habilitar verbose logging on-demand para un request específico, y cómo conectar el logging con los guardrails del módulo anterior.

El problema del logging sin estrategia

# Dos errores opuestos:

# Error 1: Demasiado poco — "logging" con prints
print(f"Response: {result}")
# → No hay contexto, no hay correlación, no hay métricas

# Error 2: Demasiado — loguear todo en todos los niveles
log.debug("Full prompt", prompt=full_prompt)  # En PRODUCCIÓN, para cada request
log.debug("Full response", response=full_response)
# → Storage enorme, PII expuesto, logs ilegibles, performance degradada
# Con 1000 requests/día × 2KB de prompt = 2GB de logs/día

# La estrategia correcta: loguear lo suficiente para responder
# las preguntas más importantes, sin los costos de loguear todo

Log levels para AI: la definición

import logging
import structlog

# CRITICAL: El sistema no puede continuar
# - Budget de API excedido completamente
# - Fallo total del sistema de logging
# - Datos de configuración corruptos

# ERROR: Algo falló y requiere atención inmediata
# - API call del LLM falló y no hay fallback
# - Parsing del output del LLM imposible después de retries
# - Guardrail de seguridad falló por error interno (no por detectar ataque)
# - Excepción no manejada en el pipeline

# WARNING: Comportamiento anómalo que NO es fallo pero requiere monitoreo
# - Se usó el modelo de fallback (el primario no respondió)
# - Guardrail se activó y bloqueó un request (comportamiento esperado, pero monitorear)
# - Latencia > 10s (alto, pero no error)
# - Costo de un request > $0.10 (anómalo)
# - JSON del LLM necesitó múltiples intentos de parseo

# INFO: Estado normal del sistema, métricas de negocio
# - Request completado (con tokens, costo, duración)
# - Inicio y fin del servidor
# - Configuración cargada

# DEBUG: Información detallada para desarrollo y debugging
# - Full prompt (solo si está habilitado)
# - Full response
# - Stack completo del guardrail pipeline
# - Parámetros exactos de la llamada al LLM

Schema estándar para cada nivel

# src/logging_config.py

import structlog
import logging
import time
from typing import Any, Optional

def build_llm_request_log(
    request_id: str,
    model: str,
    input_tokens: int,
    output_tokens: int,
    duration_ms: float,
    cost_usd: float,
    endpoint: str = None,
    guardrails_activated: list = None
) -> dict:
    """Schema estándar para log INFO de request completado."""
    return {
        "event": "llm_request_completed",
        "request_id": request_id,
        "model": model,
        "input_tokens": input_tokens,
        "output_tokens": output_tokens,
        "total_tokens": input_tokens + output_tokens,
        "cost_usd": round(cost_usd, 8),
        "duration_ms": round(duration_ms, 1),
        "endpoint": endpoint,
        "guardrails_activated": guardrails_activated or [],
    }

def build_error_log(
    request_id: str,
    error_type: str,
    error_message: str,
    model: str = None,
    prompt_hash: str = None  # Hash del prompt (no el prompt completo)
) -> dict:
    """Schema estándar para log ERROR de request fallido."""
    return {
        "event": "llm_request_failed",
        "request_id": request_id,
        "error_type": error_type,
        "error_message": error_message[:200],  # Truncar para evitar logs enormes
        "model": model,
        "prompt_hash": prompt_hash,  # Para correlacionar con reproducibility log
    }

def build_guardrail_log(
    request_id: str,
    guardrail_type: str,
    action: str,  # "blocked", "modified", "allowed"
    layer: str = None,  # "pattern", "llm_judge", "regex"
    reason: str = None
) -> dict:
    """Schema estándar para log WARNING de guardrail activado."""
    return {
        "event": "guardrail_activated",
        "request_id": request_id,
        "guardrail_type": guardrail_type,
        "action": action,
        "layer": layer,
        "reason": reason,
    }

La función de logging completa para una llamada al LLM

# src/llm_wrapper.py
import time
import hashlib
import structlog
from typing import Optional, Any

log = structlog.get_logger()

def call_llm_with_logging(
    client,
    model: str,
    messages: list,
    request_id: str,
    temperature: float = 0.0,
    max_tokens: int = 500,
    endpoint: str = None,
    debug_mode: bool = False,
    **kwargs
) -> Any:
    """
    Wrapper para llamadas al LLM que incluye logging completo.
    
    - INFO: siempre (tokens, costo, duración)
    - DEBUG: solo si debug_mode=True (prompt completo, respuesta completa)
    - WARNING: si hay fallback o alta latencia
    - ERROR: si la llamada falla
    """
    bound_log = log.bind(request_id=request_id, endpoint=endpoint)
    start_time = time.time()
    
    # DEBUG: loguear el prompt completo (solo en modo debug)
    if debug_mode:
        prompt_content = messages[-1].get("content", "") if messages else ""
        bound_log.debug(
            "llm_prompt_sent",
            model=model,
            temperature=temperature,
            max_tokens=max_tokens,
            prompt_length=len(prompt_content),
            # Solo loguear primeros 500 chars del prompt incluso en debug
            prompt_preview=prompt_content[:500]
        )
    
    # Calcular hash del prompt para reproducibilidad (siempre)
    prompt_hash = _hash_messages(messages)
    
    try:
        response = client.chat.completions.create(
            model=model,
            messages=messages,
            temperature=temperature,
            max_tokens=max_tokens,
            **kwargs
        )
        
        duration_ms = (time.time() - start_time) * 1000
        
        # Calcular costo
        from src.logging_config import PRICING
        cost_usd = _calculate_cost(model, response.usage.prompt_tokens, response.usage.completion_tokens)
        
        # INFO: log de request completado (siempre)
        bound_log.info(
            "llm_request_completed",
            model=model,
            input_tokens=response.usage.prompt_tokens,
            output_tokens=response.usage.completion_tokens,
            total_tokens=response.usage.total_tokens,
            duration_ms=round(duration_ms, 1),
            cost_usd=round(cost_usd, 8),
            prompt_hash=prompt_hash  # Para reproducibilidad
        )
        
        # WARNING: si latencia es alta
        if duration_ms > 10_000:
            bound_log.warning(
                "high_latency_request",
                duration_ms=round(duration_ms, 1),
                model=model
            )
        
        # WARNING: si costo es alto
        if cost_usd > 0.10:
            bound_log.warning(
                "high_cost_request",
                cost_usd=round(cost_usd, 6),
                model=model,
                total_tokens=response.usage.total_tokens
            )
        
        # DEBUG: loguear la respuesta completa (solo en modo debug)
        if debug_mode:
            raw_response = response.choices[0].message.content
            bound_log.debug(
                "llm_response_received",
                response_length=len(raw_response or ""),
                response_preview=(raw_response or "")[:200],
                finish_reason=response.choices[0].finish_reason
            )
        
        return response
    
    except Exception as e:
        duration_ms = (time.time() - start_time) * 1000
        
        # ERROR: la llamada al LLM falló
        bound_log.error(
            "llm_request_failed",
            error_type=type(e).__name__,
            error_message=str(e)[:200],
            model=model,
            duration_ms=round(duration_ms, 1),
            prompt_hash=prompt_hash
        )
        raise

def _hash_messages(messages: list) -> str:
    """Calcula un hash corto del prompt para reproducibilidad."""
    content = "".join(m.get("content", "") for m in messages)
    return hashlib.sha256(content.encode()).hexdigest()[:12]

def _calculate_cost(model: str, input_tokens: int, output_tokens: int) -> float:
    PRICES = {
        "gpt-4o-mini": {"input": 0.15, "output": 0.60},
        "gpt-4o": {"input": 2.50, "output": 10.00},
        "gpt-4": {"input": 30.00, "output": 60.00},
    }
    prices = PRICES.get(model, PRICES["gpt-4o-mini"])
    return (
        input_tokens / 1_000_000 * prices["input"] +
        output_tokens / 1_000_000 * prices["output"]
    )

Debug logging on-demand: no redeployar para debuggear

# Habilitar debug logging para un request específico sin afectar a los demás:

# Opción 1: Header especial (para APIs)
from fastapi import Request

async def get_debug_mode(request: Request) -> bool:
    """Solo habilitar debug mode si el header está presente Y hay una clave válida."""
    debug_key = request.headers.get("X-Debug-Logging")
    valid_debug_key = os.getenv("DEBUG_LOGGING_KEY", "")
    return bool(valid_debug_key and debug_key == valid_debug_key)

@app.post("/analyze")
async def analyze(request: Request, body: AnalyzeRequest):
    debug_mode = await get_debug_mode(request)
    request_id = get_or_create_request_id(request)
    
    result = analyze_sentiment_with_guardrails(
        body.text,
        request_id=request_id,
        debug_mode=debug_mode
    )
    return result

# Opción 2: Allowlist de request_ids para debug
DEBUG_REQUESTS = set()  # IDs que deben ser logueados en debug

def should_debug(request_id: str) -> bool:
    return request_id in DEBUG_REQUESTS

# Uso: antes de procesar un request sospechoso, añadir su ID a la lista

Lo que NUNCA debes loguear

# ❌ API keys
log.info("openai_configured", api_key=os.getenv("OPENAI_API_KEY"))

# ❌ PII sin redactar
log.info("request", input_text=user_text)  # Si contiene nombres, emails, etc.

# ❌ Full prompt en INFO (producción)
log.info("llm_call", prompt=full_prompt_string)  # 2KB × 1000 req/día = 2MB/día sin valor

# ❌ Contraseñas u otros secrets
log.debug("db_connection", password=db_password)

# ✅ Lo que SÍ loguear:
log.info("request_completed",
    request_id=request_id,
    input_length=len(user_text),     # Longitud, no el contenido
    input_hash=hash(user_text),      # Hash para correlacionar, no el texto
    model=model,
    tokens=response.usage.total_tokens,
    cost_usd=cost
)

Conectar guardrails con logging

# src/guardrails/pipeline.py (con logging integrado)
def _log(self, event: str, data: dict = None):
    """Logguear activación de guardrail con nivel apropiado."""
    if not self.config.log_activations:
        return
    
    bound_log = log.bind(request_id=data.get("request_id", "unknown"))
    
    if event in ("injection_blocked", "content_filtered"):
        # WARNING: el guardrail bloqueó algo — comportamiento esperado pero notable
        bound_log.warning(
            "guardrail_activated",
            guardrail_event=event,
            **{k: v for k, v in (data or {}).items() if k != "request_id"}
        )
    elif event in ("pii_redacted", "input_sanitized"):
        # INFO: modificación normal del pipeline
        bound_log.info(
            "guardrail_applied",
            guardrail_event=event,
            **{k: v for k, v in (data or {}).items() if k != "request_id"}
        )

Estimación de storage

# Ayuda a tomar decisiones de qué loguear:

LOG_SIZE_ESTIMATES = {
    "INFO summary (no content)": "200-400 bytes",
    "WARNING guardrail": "300-500 bytes",
    "ERROR with hash": "400-600 bytes",
    "DEBUG with prompt preview (500 chars)": "800-1200 bytes",
    "DEBUG with full prompt (5K chars)": "5000-6000 bytes",
}

# Con 1,000 requests/día:
DAILY_STORAGE = {
    "Solo INFO": "0.4 MB/día",
    "INFO + WARNING + ERROR": "0.6 MB/día",
    "Full DEBUG para todos": "5-6 MB/día",
    "Full prompt en DEBUG para todos": "50-60 MB/día",
}
# Conclusión: INFO para todos es trivial. Full DEBUG solo para requests específicos.

Tests del sistema de logging

# tests/unit/test_logging.py
import pytest
import json
import io
import structlog

@pytest.fixture
def capture_logs():
    """Captura logs de structlog para verificar en tests."""
    output = io.StringIO()
    structlog.configure(
        processors=[structlog.processors.JSONRenderer()],
        logger_factory=structlog.PrintLoggerFactory(file=output)
    )
    yield output
    output.close()

def test_llm_request_logs_tokens_and_cost(capture_logs):
    """El log de request completado incluye tokens y costo."""
    from src.llm_wrapper import call_llm_with_logging
    mock_client = create_mock_client()
    
    call_llm_with_logging(
        client=mock_client,
        model="gpt-4o-mini",
        messages=[{"role": "user", "content": "test"}],
        request_id="test-123"
    )
    
    logs = [json.loads(line) for line in capture_logs.getvalue().strip().split("\n") if line]
    completed_log = next((l for l in logs if l.get("event") == "llm_request_completed"), None)
    
    assert completed_log is not None
    assert "input_tokens" in completed_log
    assert "cost_usd" in completed_log
    assert "duration_ms" in completed_log
    assert completed_log["request_id"] == "test-123"

def test_no_full_prompt_in_info_logs(capture_logs):
    """El prompt completo NO debe aparecer en logs de nivel INFO."""
    from src.llm_wrapper import call_llm_with_logging
    mock_client = create_mock_client()
    
    secret_prompt = "This is a very secret prompt with PII: juan@example.com"
    call_llm_with_logging(
        client=mock_client,
        model="gpt-4o-mini",
        messages=[{"role": "user", "content": secret_prompt}],
        request_id="test-456",
        debug_mode=False  # Explícitamente sin debug
    )
    
    log_output = capture_logs.getvalue()
    assert secret_prompt not in log_output  # El prompt completo no debe aparecer
    assert "juan@example.com" not in log_output  # PII tampoco

Ejercicios

Ejercicio 1: Definir schema de log

Para cada escenario, escribe el log con el nivel correcto y todos los campos necesarios:

1. El LLM respondió en 12 segundos
2. El guardrail de injection bloqueó un request
3. El JSON del LLM no pudo parsearse

# Escenario 1: Alta latencia
log.warning("high_latency_request",
    request_id=request_id,
    duration_ms=12000,
    model="gpt-4o-mini",
    threshold_ms=10000
)

# Escenario 2: Guardrail activado
log.warning("guardrail_activated",
    request_id=request_id,
    guardrail_type="prompt_injection",
    action="blocked",
    layer="pattern",
    endpoint="/analyze"
)

# Escenario 3: Error de parseo
log.error("json_parse_failed",
    request_id=request_id,
    error_type="JSONDecodeError",
    response_preview=raw_response[:100],  # Preview, no completo
    model="gpt-4o-mini"
)

Ejercicio 2: Identificar logs incorrectos

¿Qué está mal en estos logs?

log.debug("processing", api_key=os.getenv("OPENAI_API_KEY"))
log.info("user_input", text=request.text)
log.info("llm_done")  # Sin campos

Ejercicio 3: Debug mode implementation

Implementa un sistema de debug mode que permita activar el logging verbose para un request específico usando un header X-Debug-Request-ID:

# Set global de IDs en modo debug (en memoria, se pierde al redeployar)
_debug_request_ids: set = set()

def enable_debug_for_request(request_id: str):
    """Habilita debug mode para un request específico."""
    _debug_request_ids.add(request_id)

def is_debug_mode(request_id: str) -> bool:
    return request_id in _debug_request_ids

# En FastAPI:
@app.post("/debug/enable/{request_id}")
async def enable_debug(request_id: str, admin_key: str = Header()):
    if admin_key != os.getenv("ADMIN_KEY"):
        raise HTTPException(403)
    enable_debug_for_request(request_id)
    return {"message": f"Debug enabled for {request_id}"}

Resumen

• INFO por defecto en producción: tokens, costo, duración, request_id — sin contenido
• DEBUG solo on-demand: prompt completo, respuesta completa — para requests específicos
• WARNING: guardrail activado, fallback usado, alta latencia, costo alto
• ERROR: falla del LLM sin fallback, excepción no manejada
• Nunca loguear: API keys, PII sin redactar, prompts completos en INFO
• Storage estimado: INFO summary = ~0.4MB/día/1000 requests — completamente manejable

Recursos adicionales

1. structlog Log Levels — Configuración de niveles
2. Python Logging HOWTO — Conceptos base de logging en Python
3. GDPR y logs (ICO) — Consideraciones legales de PII en logs
4. The Art of Logging — Guía práctica de qué loguear
5. 12 Factor — Logs — Filosofía de logging en apps cloud-native