Módulo 5: Structured Logging para AI Systems
7. Proyecto: AI Logging System
Descripción
Este es el proyecto integrador del Módulo 5. Construyes un sistema de logging completo para la app de análisis de sentimiento del módulo 4, integrando todas las piezas: structlog con JSON output, correlation IDs con contextvars, token y cost tracking, reproducibility context, y conexión con los guardrails. Al terminar, tienes un sistema que permite responder cualquier pregunta sobre lo que ocurrió en tu app usando solo los logs.
Arquitectura del AI Logging System
┌─────────────────────────────────────────────────────────────┐
│ Request /analyze │
└────────────────────────────┬────────────────────────────────┘
│
┌──────────────▼──────────────┐
│ RequestTracingMiddleware │
│ → genera/extrae request_id │
│ → bind_contextvars() │
│ → log request_started │
└──────────────┬──────────────┘
│
┌──────────────▼──────────────┐
│ GuardrailsPipeline │
│ → sanitize input │ → WARNING si inyección
│ → check injection │ → WARNING si PII redactado
│ → redact PII │ → INFO si input modificado
└──────────────┬──────────────┘
│
┌──────────────▼──────────────┐
│ call_llm_with_logging() │
│ → INFO: tokens, cost │
│ → WARNING: alta latencia │
│ → ERROR: fallo del API │
│ → DEBUG: prompt (optional) │
└──────────────┬──────────────┘
│
┌──────────────▼──────────────┐
│ Output validation │
│ → WARNING si invalid JSON │
│ → WARNING si fallback used │
└──────────────┬──────────────┘
│
┌──────────────▼──────────────┐
│ RequestTracingMiddleware │
│ → log request_completed │
│ → total cost, duration │
│ → clear_contextvars() │
└─────────────────────────────┘
Todos los logs tienen request_id automáticamente.
Output: JSON Lines en stdout y/o archivo.
Estructura del proyecto
src/
├── logging_config.py ← Configuración central de structlog
├── tracing.py ← request_id con contextvars
├── middleware.py ← FastAPI middleware para tracing
├── cost_calculator.py ← Cálculo de costos por modelo
├── llm_wrapper.py ← Wrapper del cliente LLM con logging
├── reproducibility.py ← LLMCallContext para reproducibilidad
├── guardrails/
│ └── pipeline.py ← Ya implementado en Módulo 4 (con logging)
└── app/
└── main.py ← App FastAPI con middleware y endpoints
scripts/
├── analyze_logs.py ← Análisis de costos y métricas desde logs
└── reproduce_from_logs.py ← Reproducir request desde logs
tests/
├── unit/
│ ├── test_logging_config.py
│ ├── test_tracing.py
│ ├── test_cost_calculator.py
│ └── test_llm_wrapper.py
└── integration/
└── test_logging_integration.py
Paso 1: Configuración central
# src/logging_config.py
import logging
import os
import sys
import hashlib
from typing import Any, MutableMapping
import structlog
# ─── Precios por 1M tokens ────────────────────────────────────────────────
MODEL_PRICING = {
"gpt-4o-mini": {"input": 0.150, "output": 0.600},
"gpt-4o": {"input": 2.500, "output": 10.000},
"gpt-4-turbo": {"input": 10.000, "output": 30.000},
"gpt-4": {"input": 30.000, "output": 60.000},
"gpt-3.5-turbo": {"input": 0.500, "output": 1.500},
}
DEFAULT_PRICING = {"input": 0.150, "output": 0.600}
def calculate_cost(model: str, input_tokens: int, output_tokens: int) -> float:
"""Calcula el costo en USD de una llamada al LLM."""
pricing = MODEL_PRICING.get(model)
if pricing is None:
for key in MODEL_PRICING:
if model.startswith(key):
pricing = MODEL_PRICING[key]
break
pricing = pricing or DEFAULT_PRICING
return round(
(input_tokens / 1_000_000) * pricing["input"] +
(output_tokens / 1_000_000) * pricing["output"],
8
)
# ─── Procesadores custom ──────────────────────────────────────────────────
def add_app_metadata(logger, method_name, event_dict):
"""Añade metadatos de la aplicación a todos los logs."""
event_dict["app"] = os.getenv("APP_NAME", "sentiment-analyzer")
event_dict["version"] = os.getenv("APP_VERSION", "0.1.0")
event_dict["env"] = os.getenv("ENVIRONMENT", "development")
return event_dict
def sanitize_sensitive_fields(logger, method_name, event_dict):
"""Redacta campos sensibles antes de loguear."""
REDACT = {"password", "api_key", "secret", "authorization", "token"}
HASH = {"user_id", "email"}
for field in list(event_dict.keys()):
field_lower = field.lower()
if any(s in field_lower for s in REDACT):
event_dict[field] = "[REDACTED]"
elif any(p in field_lower for p in HASH):
event_dict[field] = hashlib.sha256(
str(event_dict[field]).encode()
).hexdigest()[:12]
return event_dict
def truncate_long_values(logger, method_name, event_dict, max_len=500):
"""Trunca strings largos para evitar logs gigantes."""
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)}chars]"
return event_dict
# ─── Configuración principal ──────────────────────────────────────────────
def configure_logging(
env: str = None,
log_level: int = None,
log_file: str = None
) -> None:
"""
Configura structlog para el entorno dado.
Args:
env: "production" | "development" | "testing"
log_level: Override del nivel de log
log_file: Si se especifica, además de stdout, escribe a este archivo
"""
env = env or os.getenv("ENVIRONMENT", "development")
log_level = log_level or getattr(
logging,
os.getenv("LOG_LEVEL", "INFO").upper(),
logging.INFO
)
use_json = env in ("production", "staging")
# Configurar el logging estándar de Python
handlers = [logging.StreamHandler(sys.stdout)]
if log_file:
from logging.handlers import TimedRotatingFileHandler
os.makedirs(os.path.dirname(log_file) or ".", exist_ok=True)
handlers.append(TimedRotatingFileHandler(
log_file, when="midnight", backupCount=30, encoding="utf-8"
))
logging.basicConfig(handlers=handlers, format="%(message)s", level=log_level)
shared_processors = [
structlog.contextvars.merge_contextvars,
add_app_metadata,
sanitize_sensitive_fields,
structlog.processors.add_log_level,
structlog.processors.TimeStamper(fmt="iso"),
structlog.stdlib.add_logger_name,
structlog.processors.StackInfoRenderer(),
]
if use_json:
processors = shared_processors + [
truncate_long_values,
structlog.processors.format_exc_info,
structlog.processors.JSONRenderer()
]
else:
processors = shared_processors + [
structlog.dev.ConsoleRenderer(colors=True)
]
structlog.configure(
processors=processors,
wrapper_class=structlog.make_filtering_bound_logger(log_level),
context_class=dict,
logger_factory=structlog.PrintLoggerFactory(),
cache_logger_on_first_use=True,
)Paso 2: Tracing con contextvars
# src/tracing.py
import uuid
import contextvars
from typing import Optional
_request_id_var: contextvars.ContextVar[Optional[str]] = contextvars.ContextVar(
"request_id", default=None
)
def generate_request_id() -> str:
return uuid.uuid4().hex[:8]
def set_request_id(request_id: str) -> None:
_request_id_var.set(request_id)
structlog.contextvars.bind_contextvars(request_id=request_id)
def get_request_id() -> Optional[str]:
return _request_id_var.get()
def get_or_create_request_id() -> str:
current = _request_id_var.get()
if current is None:
current = generate_request_id()
set_request_id(current)
return current
def clear_request_id() -> None:
_request_id_var.set(None)
structlog.contextvars.clear_contextvars()Paso 3: Middleware de FastAPI
# src/middleware.py
import time
import structlog
from starlette.middleware.base import BaseHTTPMiddleware
from src.tracing import generate_request_id, set_request_id, clear_request_id
log = structlog.get_logger()
class RequestTracingMiddleware(BaseHTTPMiddleware):
async def dispatch(self, request, call_next):
request_id = (
request.headers.get("X-Request-ID") or
generate_request_id()
)
set_request_id(request_id)
start = time.time()
log.info(
"request_started",
method=request.method,
path=request.url.path
)
try:
response = await call_next(request)
duration_ms = (time.time() - start) * 1000
log.info(
"request_completed",
status_code=response.status_code,
duration_ms=round(duration_ms, 1),
path=request.url.path
)
response.headers["X-Request-ID"] = request_id
return response
except Exception as e:
log.error(
"request_failed",
error_type=type(e).__name__,
duration_ms=round((time.time() - start) * 1000, 1)
)
raise
finally:
clear_request_id()Paso 4: LLM Wrapper con logging completo
# src/llm_wrapper.py
import time
import hashlib
import json
import structlog
from typing import Optional, Tuple, Any
from src.logging_config import calculate_cost
log = structlog.get_logger()
HIGH_COST_USD = 0.05
HIGH_LATENCY_MS = 10_000
HIGH_TOKENS = 10_000
def call_llm(
client,
model: str,
messages: list,
temperature: float = 0.0,
max_tokens: int = 500,
seed: Optional[int] = None,
debug_mode: bool = False,
**kwargs
) -> Tuple[Any, dict]:
"""
Realiza una llamada al LLM con logging completo.
Retorna (response, metrics_dict).
"""
start = time.time()
messages_hash = _hash_messages(messages)
# Pre-flight: advertir si el input parece muy largo
total_chars = sum(len(m.get("content", "")) for m in messages)
if total_chars > 40_000:
log.warning(
"large_input_pre_call",
estimated_tokens=total_chars // 4,
messages_hash=messages_hash
)
# DEBUG: mostrar prompt (solo si está explícitamente habilitado)
if debug_mode:
last_content = messages[-1].get("content", "") if messages else ""
log.debug(
"llm_prompt",
model=model,
temperature=temperature,
seed=seed,
prompt_preview=last_content[:500]
)
try:
response = client.chat.completions.create(
model=model,
messages=messages,
temperature=temperature,
max_tokens=max_tokens,
seed=seed,
**kwargs
)
duration_ms = (time.time() - start) * 1000
cost_usd = calculate_cost(
model,
response.usage.prompt_tokens,
response.usage.completion_tokens
)
metrics = {
"model": model,
"input_tokens": response.usage.prompt_tokens,
"output_tokens": response.usage.completion_tokens,
"total_tokens": response.usage.total_tokens,
"cost_usd": cost_usd,
"duration_ms": round(duration_ms, 1),
"finish_reason": response.choices[0].finish_reason,
"messages_hash": messages_hash,
"seed": seed,
"temperature": temperature,
}
# System fingerprint para detectar cambios de modelo
if hasattr(response, "system_fingerprint") and response.system_fingerprint:
metrics["system_fingerprint"] = response.system_fingerprint
# INFO: métricas de la llamada (siempre)
log.info("llm_call_completed", **metrics)
# Alertas
if response.choices[0].finish_reason == "length":
log.warning("response_truncated", max_tokens=max_tokens, **metrics)
if duration_ms > HIGH_LATENCY_MS:
log.warning("high_latency", **metrics)
if cost_usd > HIGH_COST_USD:
log.warning("high_cost", **metrics)
if response.usage.prompt_tokens > HIGH_TOKENS:
log.warning("high_input_tokens", **metrics)
# DEBUG: respuesta completa
if debug_mode:
raw = response.choices[0].message.content
log.debug("llm_response",
response_preview=(raw or "")[:300],
response_hash=hashlib.sha256(
(raw or "").encode()
).hexdigest()[:12]
)
return response, metrics
except Exception as e:
duration_ms = (time.time() - start) * 1000
last_content = messages[-1].get("content", "") if messages else ""
log.error(
"llm_call_failed",
error_type=type(e).__name__,
error_msg=str(e)[:200],
model=model,
duration_ms=round(duration_ms, 1),
messages_hash=messages_hash,
prompt_preview=last_content[:300] # Más contexto en error
)
raise
def _hash_messages(messages: list) -> str:
content = json.dumps(messages, sort_keys=True, ensure_ascii=False)
return hashlib.sha256(content.encode()).hexdigest()[:12]Paso 5: App FastAPI integrada
# src/app/main.py
import os
import structlog
from fastapi import FastAPI, Request
from pydantic import BaseModel
from src.logging_config import configure_logging
from src.middleware import RequestTracingMiddleware
from src.llm_wrapper import call_llm
from src.guardrails.pipeline import GuardrailsPipeline, PUBLIC_API_CONFIG
# Configurar logging al inicio
configure_logging(
env=os.getenv("ENVIRONMENT", "development"),
log_file="logs/app.json" if os.getenv("LOG_TO_FILE") else None
)
log = structlog.get_logger()
# Crear app y registrar middleware
app = FastAPI(title="AI Sentiment Analyzer with Logging")
app.add_middleware(RequestTracingMiddleware)
# Instancia del pipeline de guardrails
_pipeline: GuardrailsPipeline = None
def get_pipeline() -> GuardrailsPipeline:
global _pipeline
if _pipeline is None:
_pipeline = GuardrailsPipeline(
config=PUBLIC_API_CONFIG,
openai_client=_get_openai_client()
)
return _pipeline
def _get_openai_client():
from openai import OpenAI
return OpenAI(api_key=os.getenv("OPENAI_API_KEY"))
class AnalyzeRequest(BaseModel):
text: str
debug: bool = False # Habilita debug logging para este request
class AnalyzeResponse(BaseModel):
sentiment: str
score: float
request_id: str
cost_usd: float
duration_ms: float
@app.post("/analyze", response_model=AnalyzeResponse)
async def analyze_endpoint(body: AnalyzeRequest, request: Request):
"""Analiza el sentimiento de un texto."""
from src.tracing import get_request_id
import time
request_id = get_request_id()
start = time.time()
log.info("analyze_requested", text_length=len(body.text))
# Ejecutar guardrails (del Módulo 4)
pipeline = get_pipeline()
pipeline_result = pipeline.process(
body.text,
request_id=request_id,
debug_mode=body.debug
)
if pipeline_result.blocked:
log.warning(
"request_blocked_by_guardrails",
reason=pipeline_result.block_reason,
guardrail=pipeline_result.blocked_by
)
from fastapi import HTTPException
raise HTTPException(400, detail=pipeline_result.block_reason)
# Loguear guardrails activados
if pipeline_result.activated_guardrails:
log.info(
"guardrails_applied",
activated=pipeline_result.activated_guardrails
)
# Llamar al LLM con logging completo
client = _get_openai_client()
SENTIMENT_PROMPT = """Analyze the sentiment of the following text.
Respond with a JSON object: {"sentiment": "positive|negative|neutral|mixed", "score": 0.0}
Score: 1.0 = very positive, -1.0 = very negative, 0.0 = neutral.
Text: {text}"""
messages = [
{"role": "system", "content": "You are a sentiment analysis expert."},
{"role": "user", "content": SENTIMENT_PROMPT.format(
text=pipeline_result.processed_input
)}
]
response, metrics = call_llm(
client=client,
model="gpt-4o-mini",
messages=messages,
temperature=0.0,
max_tokens=200,
seed=42,
debug_mode=body.debug
)
# Parsear resultado
import json
raw = response.choices[0].message.content.strip()
try:
result = json.loads(raw)
sentiment = result["sentiment"]
score = float(result["score"])
except (json.JSONDecodeError, KeyError, ValueError):
log.warning(
"output_parse_failed",
raw_preview=raw[:200],
messages_hash=metrics["messages_hash"]
)
sentiment = "unknown"
score = 0.0
total_duration_ms = (time.time() - start) * 1000
log.info(
"analyze_completed",
sentiment=sentiment,
score=score,
total_cost_usd=metrics["cost_usd"],
total_duration_ms=round(total_duration_ms, 1)
)
return AnalyzeResponse(
sentiment=sentiment,
score=score,
request_id=request_id,
cost_usd=metrics["cost_usd"],
duration_ms=total_duration_ms
)
@app.get("/health")
async def health():
log.info("health_check")
return {"status": "ok"}Paso 6: Tests del sistema de logging
# tests/unit/test_llm_wrapper.py
import json
import io
import pytest
import structlog
from unittest.mock import MagicMock
def make_mock_response(model="gpt-4o-mini", input_tokens=245, output_tokens=87,
content='{"sentiment": "positive", "score": 0.8}',
finish_reason="stop"):
mock = MagicMock()
mock.usage.prompt_tokens = input_tokens
mock.usage.completion_tokens = output_tokens
mock.usage.total_tokens = input_tokens + output_tokens
mock.choices[0].message.content = content
mock.choices[0].finish_reason = finish_reason
mock.system_fingerprint = "fp_test123"
return mock
@pytest.fixture(autouse=True)
def json_log_output():
"""Captura los logs JSON para cada test."""
output = io.StringIO()
structlog.configure(
processors=[
structlog.contextvars.merge_contextvars,
structlog.processors.add_log_level,
structlog.processors.JSONRenderer()
],
logger_factory=structlog.PrintLoggerFactory(file=output),
cache_logger_on_first_use=False
)
yield output
structlog.contextvars.clear_contextvars()
def read_logs(output: io.StringIO) -> list:
return [json.loads(l) for l in output.getvalue().strip().split("\n") if l.strip()]
class TestCallLLM:
def test_logs_tokens_and_cost(self, json_log_output):
"""Verifica que se loguean tokens y costo."""
from src.llm_wrapper import call_llm
mock_client = MagicMock()
mock_client.chat.completions.create.return_value = make_mock_response()
structlog.contextvars.bind_contextvars(request_id="test-001")
_, metrics = call_llm(mock_client, "gpt-4o-mini",
[{"role": "user", "content": "test"}])
logs = read_logs(json_log_output)
completed = next(l for l in logs if l["event"] == "llm_call_completed")
assert completed["input_tokens"] == 245
assert completed["output_tokens"] == 87
assert "cost_usd" in completed
assert completed["cost_usd"] > 0
assert completed["request_id"] == "test-001"
def test_warns_on_truncated_response(self, json_log_output):
"""Si finish_reason es 'length', debe loguear un WARNING."""
from src.llm_wrapper import call_llm
mock_client = MagicMock()
mock_client.chat.completions.create.return_value = make_mock_response(
finish_reason="length"
)
call_llm(mock_client, "gpt-4o-mini", [{"role": "user", "content": "test"}])
logs = read_logs(json_log_output)
warnings = [l for l in logs if l.get("level") == "warning"]
assert any(l["event"] == "response_truncated" for l in warnings)
def test_no_prompt_in_info_logs(self, json_log_output):
"""El prompt no debe aparecer en logs de nivel INFO."""
from src.llm_wrapper import call_llm
mock_client = MagicMock()
mock_client.chat.completions.create.return_value = make_mock_response()
secret_text = "My credit card is 4111-1111-1111-1111"
call_llm(mock_client, "gpt-4o-mini",
[{"role": "user", "content": secret_text}],
debug_mode=False)
output = json_log_output.getvalue()
assert "4111-1111-1111-1111" not in output
def test_includes_system_fingerprint(self, json_log_output):
"""El system_fingerprint debe aparecer en el log."""
from src.llm_wrapper import call_llm
mock_client = MagicMock()
mock_client.chat.completions.create.return_value = make_mock_response()
call_llm(mock_client, "gpt-4o-mini", [{"role": "user", "content": "test"}])
logs = read_logs(json_log_output)
completed = next(l for l in logs if l["event"] == "llm_call_completed")
assert completed.get("system_fingerprint") == "fp_test123"
class TestCostCalculator:
def test_gpt4o_mini_pricing(self):
"""Verifica el cálculo de costo para gpt-4o-mini."""
from src.logging_config import calculate_cost
# 1M input = $0.15, 1M output = $0.60
cost = calculate_cost("gpt-4o-mini", 1_000_000, 0)
assert abs(cost - 0.15) < 0.001
def test_cost_returned_in_metrics(self):
"""call_llm debe retornar costo en metrics dict."""
from src.llm_wrapper import call_llm
mock_client = MagicMock()
mock_client.chat.completions.create.return_value = make_mock_response(
input_tokens=500, output_tokens=200
)
_, metrics = call_llm(mock_client, "gpt-4o-mini",
[{"role": "user", "content": "test"}])
assert "cost_usd" in metrics
assert metrics["cost_usd"] > 0Paso 7: Script de análisis
# scripts/analyze_logs.py
"""
Analiza los logs JSON Lines de la app y genera un reporte.
Uso: python scripts/analyze_logs.py [log_file] [date_prefix]
"""
import json
import sys
from collections import defaultdict
def load_jsonl(path: str) -> list:
logs = []
with open(path) as f:
for line in f:
line = line.strip()
if line:
try:
logs.append(json.loads(line))
except json.JSONDecodeError:
pass
return logs
def report(logs: list, date_prefix: str = None):
if date_prefix:
logs = [l for l in logs if l.get("timestamp", "").startswith(date_prefix)]
llm_logs = [l for l in logs if l.get("event") == "llm_call_completed"]
error_logs = [l for l in logs if l.get("level") == "error"]
guard_logs = [l for l in logs if l.get("event") == "guardrail_activated"]
warn_logs = [l for l in logs if l.get("level") == "warning"]
print("=" * 60)
print("AI LOGGING SYSTEM — REPORT")
if date_prefix:
print(f"Date filter: {date_prefix}")
print("=" * 60)
# Summary
print(f"\n📊 TOTALS")
print(f" LLM calls: {len(llm_logs)}")
print(f" Errors: {len(error_logs)}")
print(f" Warnings: {len(warn_logs)}")
print(f" Guardrail events: {len(guard_logs)}")
# Cost
if llm_logs:
total = sum(l.get("cost_usd", 0) for l in llm_logs)
tokens = sum(l.get("total_tokens", 0) for l in llm_logs)
avg = total / len(llm_logs)
top = max(llm_logs, key=lambda l: l.get("cost_usd", 0))
print(f"\n💰 COST")
print(f" Total: ${total:.6f}")
print(f" Avg per call: ${avg:.8f}")
print(f" Total tokens: {tokens:,}")
print(f" Most expensive: ${top.get('cost_usd', 0):.6f}")
print(f" → request_id: {top.get('request_id', 'N/A')}")
print(f" → tokens: {top.get('total_tokens', 0):,}")
# Performance
if llm_logs:
durations = [l.get("duration_ms", 0) for l in llm_logs]
avg_d = sum(durations) / len(durations)
slow = [l for l in llm_logs if l.get("duration_ms", 0) > 5000]
print(f"\n⚡ PERFORMANCE")
print(f" Avg LLM latency: {avg_d:.0f}ms")
print(f" Slow (>5s): {len(slow)}")
# Errors
if error_logs:
etypes = defaultdict(int)
for l in error_logs:
etypes[l.get("error_type", "unknown")] += 1
print(f"\n❌ ERRORS ({len(error_logs)})")
for et, cnt in sorted(etypes.items(), key=lambda x: -x[1]):
print(f" {et}: {cnt}")
# Guardrails
if guard_logs:
gtypes = defaultdict(int)
for l in guard_logs:
gtypes[l.get("guardrail_type", "unknown")] += 1
print(f"\n🛡️ GUARDRAILS ({len(guard_logs)})")
for gt, cnt in sorted(gtypes.items(), key=lambda x: -x[1]):
print(f" {gt}: {cnt}")
print("\n" + "=" * 60)
if __name__ == "__main__":
log_file = sys.argv[1] if len(sys.argv) > 1 else "logs/app.json"
date_prefix = sys.argv[2] if len(sys.argv) > 2 else None
logs = load_jsonl(log_file)
report(logs, date_prefix)Verificación: qué debes ver al terminar
# Arrancar la app
ENVIRONMENT=development python -m uvicorn src.app.main:app --reload
# En otro terminal, hacer un request
curl -X POST http://localhost:8000/analyze \
-H "Content-Type: application/json" \
-d '{"text": "The product is great but shipping was slow"}'
# Response incluye request_id:
# {
# "sentiment": "mixed",
# "score": 0.2,
# "request_id": "a1b2c3d4", ← Para debugging
# "cost_usd": 0.0000888,
# "duration_ms": 1240.5
# }
# Logs en consola (modo desarrollo):
# [info] request_started path=/analyze
# [info] llm_call_completed model=gpt-4o-mini input_tokens=156 cost_usd=0.0000888
# [info] analyze_completed sentiment=mixed score=0.2
# [info] request_completed status_code=200 duration_ms=1240.5
# Con ENVIRONMENT=production, los logs serían JSON Lines:
# {"event": "request_started", "request_id": "a1b2c3d4", "path": "/analyze", ...}
# {"event": "llm_call_completed", "request_id": "a1b2c3d4", "model": "gpt-4o-mini", ...}
# ...
# Para buscar un request específico:
# jq 'select(.request_id == "a1b2c3d4")' logs/app.json | jq -s 'sort_by(.timestamp)'
# Para ver el costo total del día:
# python scripts/analyze_logs.py logs/app.json 2024-01-15Checklist del proyecto
[ ] src/logging_config.py implementado
[ ] configure_logging() funciona con env dev y production
[ ] Procesadores custom: sanitize_sensitive_fields, add_app_metadata
[ ] JSON output en production, ConsoleRenderer en development
[ ] src/tracing.py implementado
[ ] generate_request_id() genera IDs únicos
[ ] set_request_id() bind en contextvars Y structlog context
[ ] get_request_id() retorna el ID del contexto actual
[ ] src/middleware.py implementado
[ ] Extrae/genera request_id
[ ] Logguea request_started y request_completed
[ ] Añade X-Request-ID al response
[ ] src/llm_wrapper.py implementado
[ ] Logguea tokens, costo, duración en INFO
[ ] Alertas WARNING para alta latencia y alto costo
[ ] ERROR con más contexto que INFO
[ ] No expone prompt en INFO
[ ] App integrada
[ ] Middleware registrado
[ ] call_llm() usado en el endpoint
[ ] Guardrails logueados correctamente
[ ] Tests pasan
[ ] test_logs_tokens_and_cost
[ ] test_warns_on_truncated_response
[ ] test_no_prompt_in_info_logs
[ ] Scripts funcionan
[ ] analyze_logs.py genera reporte
[ ] jq queries retornan resultados esperados