4. Los Tres Pilares en Contexto AI

Descripción de la cápsula

Ya sabes qué es observabilidad, por qué es diferente de monitoring, y por qué sistemas AI necesitan un enfoque distinto. Ahora es momento de abrir los tres pilares — logs, métricas, traces — y ver qué hay dentro cuando el sistema que observas hace llamadas a LLMs, ejecuta pipelines RAG, y genera outputs no determinísticos.

Los tres pilares no son un invento nuevo. Llevan años en el mundo de infraestructura y backend. Pero cuando los aplicas a sistemas AI, el contenido de cada pilar cambia radicalmente. Un log ya no es solo "request recibido, status 200". Ahora necesita capturar el prompt, los tokens consumidos, el modelo utilizado, la temperatura, y el costo. Una métrica ya no es solo latencia promedio. Necesitas TTFT (time to first token), tokens por request, costo por endpoint. Un trace ya no es "HTTP → base de datos → respuesta". Es "HTTP → validación → embedding → vector search → prompt construction → LLM call → output validation → respuesta".

Pero el insight más importante de esta cápsula no son los pilares individuales — es cómo se correlacionan. Una alerta de métrica te dice que la latencia subió. Un trace te muestra qué paso del pipeline está lento. Un log te revela que el prompt incluía un contexto de 15,000 tokens porque el retrieval no filtró bien. Los tres pilares, usados juntos, te dan una narrativa completa. Usados por separado, te dan fragmentos sin contexto.

Logs en AI: Más que Eventos de Aplicación

El log tradicional vs el log AI

En una API REST clásica, un log típico captura eventos de infraestructura:

2026-03-08 14:23:01 INFO  Request received: GET /api/users/42
2026-03-08 14:23:01 INFO  Database query executed in 12ms
2026-03-08 14:23:01 INFO  Response sent: 200 OK (15ms)

Útil para debugging básico. Pero si tu endpoint llama a un LLM, ese log no te dice nada sobre lo que realmente pasó. ¿Qué prompt se envió? ¿Cuántos tokens consumió? ¿Qué modelo se usó? ¿Cuánto costó? ¿El output fue correcto?

Un log AI-aware necesita capturar datos que no existen en software tradicional:

Dato                    ¿Por qué?
──────────────────────────────────────────────────────────────────
prompt_content          Para reproducir y debuggear outputs
completion_content      Para verificar calidad post-facto
model                   Modelos diferentes = comportamiento diferente
temperature             Afecta determinismo del output
prompt_tokens           Costo y tamaño del input
completion_tokens       Costo del output generado
total_tokens            Costo total del request
latency_ms              Performance de la llamada LLM
cost_usd                Impacto financiero directo
system_prompt_version   Para rastrear cambios en el comportamiento
rag_context_chunks      Cuántos documentos se incluyeron en el contexto

Structured logging para AI con structlog

La diferencia entre un print() y un log estructurado es la diferencia entre poder buscar información y tener que leer líneas manualmente. Los logs estructurados son JSON: cada campo es buscable, filtrable, y agregable.

import structlog
import time
from openai import OpenAI
from dotenv import load_dotenv

load_dotenv()

structlog.configure(
    processors=[
        structlog.processors.TimeStamper(fmt="iso"),
        structlog.processors.add_log_level,
        structlog.processors.StackInfoRenderer(),
        structlog.processors.JSONRenderer(),
    ],
    wrapper_class=structlog.BoundLogger,
    context_class=dict,
    logger_factory=structlog.PrintLoggerFactory(),
)

logger = structlog.get_logger()
client = OpenAI()

COST_PER_1K = {
    "gpt-4o": {"prompt": 0.0025, "completion": 0.01},
    "gpt-4o-mini": {"prompt": 0.00015, "completion": 0.0006},
}


def calculate_cost(model: str, prompt_tokens: int, completion_tokens: int) -> float:
    rates = COST_PER_1K.get(model, {"prompt": 0.0, "completion": 0.0})
    return (prompt_tokens / 1000 * rates["prompt"]) + (
        completion_tokens / 1000 * rates["completion"]
    )


def call_llm_with_logging(
    prompt: str,
    model: str = "gpt-4o-mini",
    temperature: float = 0.7,
    system_prompt: str = "Eres un asistente útil.",
    system_prompt_version: str = "v1.0",
    endpoint: str = "/unknown",
    user_id: str = "anonymous",
) -> str:
    log = logger.bind(
        endpoint=endpoint,
        user_id=user_id,
        model=model,
        temperature=temperature,
        system_prompt_version=system_prompt_version,
    )

    log.info("llm_request_started", prompt_length=len(prompt))

    start = time.time()
    try:
        response = client.chat.completions.create(
            model=model,
            temperature=temperature,
            messages=[
                {"role": "system", "content": system_prompt},
                {"role": "user", "content": prompt},
            ],
            max_tokens=500,
        )
        latency_ms = (time.time() - start) * 1000
        usage = response.usage
        output = response.choices[0].message.content
        cost = calculate_cost(model, usage.prompt_tokens, usage.completion_tokens)

        log.info(
            "llm_request_completed",
            prompt_tokens=usage.prompt_tokens,
            completion_tokens=usage.completion_tokens,
            total_tokens=usage.total_tokens,
            latency_ms=round(latency_ms, 2),
            cost_usd=round(cost, 6),
            output_length=len(output),
            finish_reason=response.choices[0].finish_reason,
        )

        return output

    except Exception as e:
        latency_ms = (time.time() - start) * 1000
        log.error(
            "llm_request_failed",
            error_type=type(e).__name__,
            error_message=str(e),
            latency_ms=round(latency_ms, 2),
        )
        raise


result = call_llm_with_logging(
    prompt="¿Qué es observabilidad en sistemas AI?",
    model="gpt-4o-mini",
    endpoint="/api/chat",
    user_id="user_12345",
    system_prompt_version="v2.1",
)
print(f"\nOutput: {result[:100]}...")

Output esperado (JSON formateado para legibilidad):

{
  "event": "llm_request_started",
  "timestamp": "2026-03-08T14:23:01.234Z",
  "level": "info",
  "endpoint": "/api/chat",
  "user_id": "user_12345",
  "model": "gpt-4o-mini",
  "temperature": 0.7,
  "system_prompt_version": "v2.1",
  "prompt_length": 42
}
{
  "event": "llm_request_completed",
  "timestamp": "2026-03-08T14:23:02.567Z",
  "level": "info",
  "endpoint": "/api/chat",
  "user_id": "user_12345",
  "model": "gpt-4o-mini",
  "prompt_tokens": 28,
  "completion_tokens": 87,
  "total_tokens": 115,
  "latency_ms": 1332.45,
  "cost_usd": 0.000056,
  "output_length": 312,
  "finish_reason": "stop"
}

Fíjate en lo que puedes hacer con logs así:

• 🔍 Buscar todos los requests de un usuario: user_id = "user_12345"
• 💰 Sumar costo por endpoint: SELECT SUM(cost_usd) WHERE endpoint = "/api/chat"
• ⚡ Filtrar requests lentos: latency_ms > 3000
• 📊 Agrupar por modelo: GROUP BY model
• 🚨 Alertar en errores: event = "llm_request_failed"

Con un print("Request completed"), nada de esto es posible.

Qué NO loguear

Cuidado con la privacidad y el volumen:

• ❌ No loguees prompts completos en producción por defecto. Si tus prompts contienen datos personales de usuarios, cumplir GDPR/CCPA requiere anonimización o consentimiento.
• ❌ No loguees completions completas siempre. Un output de 4,000 tokens por request genera logs masivos. Loguea un hash o los primeros N caracteres.
• ✅ Sí loguea metadata siempre: tokens, costo, latencia, modelo, endpoint, user_id (anonimizado si es necesario).
• ✅ Loguea prompts/completions en ambientes de desarrollo y debug sessions específicos.

import hashlib


def safe_log_content(content: str, max_preview: int = 100) -> dict:
    """Loguea metadata de contenido sin exponer el texto completo."""
    return {
        "length": len(content),
        "preview": content[:max_preview] + "..." if len(content) > max_preview else content,
        "hash": hashlib.sha256(content.encode()).hexdigest()[:16],
    }

Métricas en AI: Números que Cambian Decisiones

Métricas tradicionales vs métricas AI

Las métricas de infraestructura que ya conoces siguen siendo necesarias:

Métrica tradicional          Qué te dice
──────────────────────────────────────────────────
request_count                Volumen de tráfico
error_rate                   Porcentaje de fallos
latency_p50 / p99            Performance general
uptime                       Disponibilidad

Pero en AI necesitas una capa adicional de métricas que no existen en software convencional:

Métrica AI-específica         Qué te dice
──────────────────────────────────────────────────────────────────────
ttft_ms                       Time to first token — percepción de velocidad
tti_ms                        Time to last token (TTI) — duración total
tokens_per_request            Costo en tokens (prompt + completion)
cost_per_request_usd          Impacto financiero de cada request
cost_per_user_day_usd         Cuánto cuesta servir a cada usuario
tokens_per_endpoint           Qué endpoints consumen más tokens
hallucination_rate            Porcentaje de outputs con información fabricada
quality_score                 Evaluación de relevancia/utilidad del output
model_switch_count            Cuántas veces se cambió de modelo (si hay fallback)
rag_chunks_per_request        Cantidad de contexto incluido en cada prompt
cache_hit_rate                Porcentaje de respuestas servidas desde cache

TTFT vs TTI vs End-to-End

Estos tres tipos de latencia miden cosas diferentes:

                   ┌─── TTFT ───┐
                   │             │
Request ──────────► Primer token ────────────► Último token ─────► Response
                                  │                          │
                                  └──── Streaming time ──────┘
                   │                                               │
                   └──────────────── End-to-end ───────────────────┘

• TTFT (Time To First Token): Cuánto tarda el modelo en empezar a generar. Afecta la percepción del usuario — si TTFT es alto, el usuario siente que "no pasa nada".
• TTI (Time To last token / Inference): Cuánto tarda en generar todo el output. Depende del número de tokens generados.
• End-to-end: Desde que el request llega a tu servidor hasta que el response sale. Incluye validación, embedding, RAG, LLM, post-procesamiento.

Recolección de métricas AI con contadores simples

Antes de llegar a Prometheus (módulo 4), puedes empezar con contadores en memoria que te dan visibilidad inmediata:

import time
from dataclasses import dataclass, field
from collections import defaultdict


@dataclass
class AIMetricsCollector:
    """Recolector simple de métricas AI.

    En producción usarías Prometheus/OpenTelemetry.
    Este collector demuestra qué métricas capturar.
    """

    total_requests: int = 0
    total_errors: int = 0
    total_tokens: int = 0
    total_cost_usd: float = 0.0
    latencies_ms: list = field(default_factory=list)
    ttft_ms_list: list = field(default_factory=list)
    tokens_by_endpoint: dict = field(default_factory=lambda: defaultdict(int))
    cost_by_endpoint: dict = field(default_factory=lambda: defaultdict(float))
    cost_by_user: dict = field(default_factory=lambda: defaultdict(float))
    errors_by_type: dict = field(default_factory=lambda: defaultdict(int))

    def record_request(
        self,
        endpoint: str,
        user_id: str,
        model: str,
        prompt_tokens: int,
        completion_tokens: int,
        latency_ms: float,
        cost_usd: float,
        ttft_ms: float | None = None,
    ):
        self.total_requests += 1
        self.total_tokens += prompt_tokens + completion_tokens
        self.total_cost_usd += cost_usd
        self.latencies_ms.append(latency_ms)
        self.tokens_by_endpoint[endpoint] += prompt_tokens + completion_tokens
        self.cost_by_endpoint[endpoint] += cost_usd
        self.cost_by_user[user_id] += cost_usd

        if ttft_ms is not None:
            self.ttft_ms_list.append(ttft_ms)

    def record_error(self, error_type: str):
        self.total_errors += 1
        self.errors_by_type[error_type] += 1

    def get_summary(self) -> dict:
        sorted_latencies = sorted(self.latencies_ms)
        p50_idx = len(sorted_latencies) // 2
        p99_idx = int(len(sorted_latencies) * 0.99)

        return {
            "total_requests": self.total_requests,
            "total_errors": self.total_errors,
            "error_rate": (
                round(self.total_errors / self.total_requests * 100, 2)
                if self.total_requests > 0
                else 0
            ),
            "total_tokens": self.total_tokens,
            "avg_tokens_per_request": (
                round(self.total_tokens / self.total_requests)
                if self.total_requests > 0
                else 0
            ),
            "total_cost_usd": round(self.total_cost_usd, 4),
            "avg_cost_per_request_usd": (
                round(self.total_cost_usd / self.total_requests, 6)
                if self.total_requests > 0
                else 0
            ),
            "latency_p50_ms": (
                round(sorted_latencies[p50_idx], 2) if sorted_latencies else 0
            ),
            "latency_p99_ms": (
                round(sorted_latencies[p99_idx], 2) if sorted_latencies else 0
            ),
            "top_endpoints_by_cost": dict(
                sorted(
                    self.cost_by_endpoint.items(), key=lambda x: x[1], reverse=True
                )[:5]
            ),
            "top_users_by_cost": dict(
                sorted(self.cost_by_user.items(), key=lambda x: x[1], reverse=True)[
                    :5
                ]
            ),
        }


metrics = AIMetricsCollector()

metrics.record_request(
    endpoint="/api/chat",
    user_id="user_001",
    model="gpt-4o-mini",
    prompt_tokens=150,
    completion_tokens=200,
    latency_ms=1200.5,
    cost_usd=0.000142,
    ttft_ms=340.0,
)
metrics.record_request(
    endpoint="/api/summarize",
    user_id="user_001",
    model="gpt-4o",
    prompt_tokens=4000,
    completion_tokens=500,
    latency_ms=3400.2,
    cost_usd=0.015,
    ttft_ms=890.0,
)
metrics.record_request(
    endpoint="/api/chat",
    user_id="user_002",
    model="gpt-4o-mini",
    prompt_tokens=80,
    completion_tokens=120,
    latency_ms=900.1,
    cost_usd=0.000084,
    ttft_ms=280.0,
)
metrics.record_error("RateLimitError")

summary = metrics.get_summary()
print("=" * 60)
print("AI METRICS SUMMARY")
print("=" * 60)
for key, value in summary.items():
    print(f"  {key}: {value}")

Output esperado:

============================================================
AI METRICS SUMMARY
============================================================
  total_requests: 3
  total_errors: 1
  error_rate: 33.33
  total_tokens: 5050
  avg_tokens_per_request: 1683
  total_cost_usd: 0.0152
  avg_cost_per_request_usd: 0.005075
  latency_p50_ms: 1200.5
  latency_p99_ms: 3400.2
  top_endpoints_by_cost: {'/api/summarize': 0.015, '/api/chat': 0.000226}
  top_users_by_cost: {'user_001': 0.015142, 'user_002': 8.4e-05}

Con tres requests ya puedes ver que /api/summarize cuesta 66x más que /api/chat, y que user_001 genera el 99% del gasto. Sin métricas, esa información es invisible.

Traces en AI: La Historia Completa de un Request

El trace tradicional vs el trace AI

Un trace en un backend REST clásico es simple:

[Trace: abc-123]
  └── HTTP GET /api/users/42 .................. 15ms
       └── Database query (SELECT * FROM users) .. 12ms

Dos spans. Directo. Si el request está lento, el bottleneck está en la base de datos.

Un trace en un sistema AI es fundamentalmente más complejo:

[Trace: xyz-789]
  └── HTTP POST /api/chat ............................ 4,200ms
       ├── Input validation ........................... 5ms
       ├── Embedding call (text-embedding-3-small) .... 120ms
       │    └── tokens: 45, cost: $0.000002
       ├── Vector search (Pinecone) ................... 230ms
       │    └── results: 8 chunks, relevance: [0.92, 0.89, 0.87, ...]
       ├── Context construction ....................... 15ms
       │    └── selected: 4 chunks, total_tokens: 3,200
       ├── Prompt assembly ............................ 2ms
       │    └── system_prompt_v2.1 + context + user_query
       ├── LLM call (gpt-4o) ......................... 3,750ms
       │    ├── prompt_tokens: 3,400
       │    ├── completion_tokens: 280
       │    ├── cost: $0.0113
       │    ├── ttft: 650ms
       │    └── finish_reason: stop
       └── Output validation .......................... 78ms
            └── hallucination_check: pass, format_check: pass

Ahora el trace te cuenta una historia. La latencia total es 4,200ms, pero el 89% del tiempo está en la llamada LLM. El embedding fue rápido. El vector search devolvió 8 chunks pero solo usaste 4. El prompt final tenía 3,400 tokens. El output pasó validación. Cada span tiene atributos AI-específicos.

Implementación: trace manual con logging estructurado

Antes de usar OpenTelemetry (módulo 3), puedes construir traces simples con logging. Esto demuestra el concepto y ya te da valor:

import time
import uuid
import structlog
from dataclasses import dataclass

structlog.configure(
    processors=[
        structlog.processors.TimeStamper(fmt="iso"),
        structlog.processors.add_log_level,
        structlog.processors.JSONRenderer(),
    ],
    wrapper_class=structlog.BoundLogger,
    context_class=dict,
    logger_factory=structlog.PrintLoggerFactory(),
)


@dataclass
class SpanResult:
    name: str
    duration_ms: float
    attributes: dict


class SimpleTracer:
    """Tracer manual para demostrar el concepto antes de OTel."""

    def __init__(self):
        self.trace_id = str(uuid.uuid4())[:8]
        self.spans: list[SpanResult] = []
        self.logger = structlog.get_logger().bind(trace_id=self.trace_id)

    def span(self, name: str):
        return SpanContext(self, name)

    def summary(self) -> dict:
        total_ms = sum(s.duration_ms for s in self.spans)
        return {
            "trace_id": self.trace_id,
            "total_duration_ms": round(total_ms, 2),
            "span_count": len(self.spans),
            "spans": [
                {
                    "name": s.name,
                    "duration_ms": round(s.duration_ms, 2),
                    "pct_of_total": round(s.duration_ms / total_ms * 100, 1)
                    if total_ms > 0
                    else 0,
                    **s.attributes,
                }
                for s in self.spans
            ],
        }


class SpanContext:
    def __init__(self, tracer: SimpleTracer, name: str):
        self.tracer = tracer
        self.name = name
        self.attributes: dict = {}
        self.start_time = 0.0

    def __enter__(self):
        self.start_time = time.time()
        self.tracer.logger.info(f"span_start", span=self.name)
        return self

    def __exit__(self, exc_type, exc_val, exc_tb):
        duration_ms = (time.time() - self.start_time) * 1000
        self.tracer.spans.append(
            SpanResult(
                name=self.name,
                duration_ms=duration_ms,
                attributes=self.attributes,
            )
        )
        self.tracer.logger.info(
            "span_end",
            span=self.name,
            duration_ms=round(duration_ms, 2),
            **self.attributes,
        )
        return False

    def set(self, key: str, value) -> "SpanContext":
        self.attributes[key] = value
        return self


def simulate_rag_pipeline(query: str) -> dict:
    """Simula un pipeline RAG completo con tracing manual."""
    tracer = SimpleTracer()

    with tracer.span("input_validation") as span:
        time.sleep(0.005)
        is_valid = len(query) > 0 and len(query) < 10000
        span.set("query_length", len(query))
        span.set("is_valid", is_valid)

    with tracer.span("embedding") as span:
        time.sleep(0.12)
        span.set("model", "text-embedding-3-small")
        span.set("input_tokens", 45)
        span.set("dimensions", 1536)

    with tracer.span("vector_search") as span:
        time.sleep(0.23)
        span.set("index", "knowledge_base")
        span.set("top_k", 8)
        span.set("results_returned", 8)
        span.set("min_relevance_score", 0.82)

    with tracer.span("context_construction") as span:
        time.sleep(0.015)
        span.set("chunks_selected", 4)
        span.set("chunks_discarded", 4)
        span.set("context_tokens", 3200)
        span.set("selection_strategy", "relevance_threshold_0.85")

    with tracer.span("prompt_assembly") as span:
        time.sleep(0.002)
        span.set("system_prompt_version", "v2.1")
        span.set("total_prompt_tokens", 3400)
        span.set("includes_rag_context", True)

    with tracer.span("llm_call") as span:
        time.sleep(0.8)
        span.set("model", "gpt-4o")
        span.set("temperature", 0.3)
        span.set("prompt_tokens", 3400)
        span.set("completion_tokens", 280)
        span.set("cost_usd", 0.0113)
        span.set("ttft_ms", 650)
        span.set("finish_reason", "stop")

    with tracer.span("output_validation") as span:
        time.sleep(0.08)
        span.set("hallucination_check", "pass")
        span.set("format_check", "pass")
        span.set("confidence_score", 0.91)

    return tracer.summary()


result = simulate_rag_pipeline("¿Cuál es la política de devoluciones?")

print("\n" + "=" * 60)
print("TRACE SUMMARY")
print("=" * 60)
print(f"Trace ID: {result['trace_id']}")
print(f"Total: {result['total_duration_ms']}ms ({result['span_count']} spans)")
print("-" * 60)
for span in result["spans"]:
    name = span.pop("name")
    duration = span.pop("duration_ms")
    pct = span.pop("pct_of_total")
    bar = "█" * int(pct / 2)
    print(f"  {name:.<30} {duration:>8.1f}ms ({pct:>5.1f}%) {bar}")
    for k, v in span.items():
        print(f"    {k}: {v}")

Output esperado:

============================================================
TRACE SUMMARY
============================================================
Trace ID: a3f2c1d8
Total: 1252.34ms (7 spans)
------------------------------------------------------------
  input_validation............      5.1ms ( 0.4%)
    query_length: 40
    is_valid: True
  embedding...................    120.3ms ( 9.6%) ████
    model: text-embedding-3-small
    input_tokens: 45
    dimensions: 1536
  vector_search...............    230.5ms (18.4%) █████████
    index: knowledge_base
    top_k: 8
    results_returned: 8
    min_relevance_score: 0.82
  context_construction........     15.2ms ( 1.2%)
    chunks_selected: 4
    chunks_discarded: 4
    context_tokens: 3200
    selection_strategy: relevance_threshold_0.85
  prompt_assembly.............      2.1ms ( 0.2%)
    system_prompt_version: v2.1
    total_prompt_tokens: 3400
    includes_rag_context: True
  llm_call....................    800.8ms (64.0%) ████████████████████████████████
    model: gpt-4o
    temperature: 0.3
    prompt_tokens: 3400
    completion_tokens: 280
    cost_usd: 0.0113
    ttft_ms: 650
    finish_reason: stop
  output_validation...........     80.3ms ( 6.4%) ███
    hallucination_check: pass
    format_check: pass
    confidence_score: 0.91

Ahora puedes ver que el 64% del tiempo se gasta en la llamada LLM. Si quieres optimizar latencia, ahí está tu bottleneck. El vector search toma el 18% — también significativo. Y el output_validation es un 6.4% que vale cada milisegundo porque atrapa hallucinations antes de que lleguen al usuario.

Los Tres Pilares Correlacionados: El Insight Clave

No son tres herramientas — son tres perspectivas

El error más común al aprender observabilidad es tratar los tres pilares como herramientas independientes: "tengo mis logs por aquí, mis métricas por allá, y si un día necesito traces, ya veré." Eso es como tener un departamento de ventas, uno de marketing, y uno de producto que nunca se hablan.

Los tres pilares son tres perspectivas del mismo evento:

UN SOLO REQUEST genera:
├── LOG:    {"event": "llm_request_completed", "tokens": 3680, "cost": 0.0113, ...}
├── METRIC: request_latency_ms = 4200, tokens_total = 3680, cost_usd = 0.0113
└── TRACE:  [validation → embedding → search → context → llm → validation] = 4200ms

El log te da el detalle (qué pasó exactamente en este request). La métrica te da la tendencia (cómo se compara este request con otros). El trace te da el flujo (qué pasos se ejecutaron y cuánto duró cada uno).

El ciclo de investigación: métrica → trace → log

En producción, la investigación típica de un incidente sigue este flujo:

1. ALERTA DE MÉTRICA
   "latency_p99 subió de 2s a 8s en los últimos 15 minutos"
   → Sabes QUÉ pasa, pero no POR QUÉ.

2. INVESTIGACIÓN DE TRACES
   Filtras traces donde latency > 5000ms.
   Descubres que el span "vector_search" pasó de 200ms a 3,500ms.
   → Sabes DÓNDE está el problema, pero no la CAUSA.

3. DETALLE EN LOGS
   Filtras logs del span "vector_search" para esos traces lentos.
   Descubres que top_k cambió de 8 a 50 (alguien modificó la config).
   → Sabes POR QUÉ pasó. Puedes arreglar.

Sin correlación entre los tres, cada pilar te da una pieza incompleta:

• Solo métricas: "Algo está lento." → ¿Qué? ¿Dónde? ¿Por qué?
• Solo logs: "Este request hizo X." → ¿Es un caso aislado o un patrón?
• Solo traces: "Este request pasó por estos pasos." → ¿Cuántos requests están así?

Ejemplo práctico de correlación

import time
import uuid
import structlog
from collections import defaultdict

structlog.configure(
    processors=[
        structlog.processors.TimeStamper(fmt="iso"),
        structlog.processors.add_log_level,
        structlog.processors.JSONRenderer(),
    ],
    wrapper_class=structlog.BoundLogger,
    context_class=dict,
    logger_factory=structlog.PrintLoggerFactory(),
)


class CorrelatedObservability:
    """Demuestra cómo los tres pilares se conectan con IDs compartidos."""

    def __init__(self):
        self.logger = structlog.get_logger()
        self.metrics = defaultdict(list)

    def handle_request(self, query: str, user_id: str, endpoint: str):
        trace_id = str(uuid.uuid4())[:12]
        log = self.logger.bind(
            trace_id=trace_id, user_id=user_id, endpoint=endpoint
        )

        request_start = time.time()
        spans = []

        # Span 1: Embedding
        t0 = time.time()
        time.sleep(0.1)
        embed_ms = (time.time() - t0) * 1000
        spans.append(("embedding", embed_ms))
        log.info("span_completed", span="embedding", duration_ms=round(embed_ms, 1))

        # Span 2: Vector search
        t0 = time.time()
        time.sleep(0.2)
        search_ms = (time.time() - t0) * 1000
        chunks_found = 6
        spans.append(("vector_search", search_ms))
        log.info(
            "span_completed",
            span="vector_search",
            duration_ms=round(search_ms, 1),
            chunks_found=chunks_found,
        )

        # Span 3: LLM call
        t0 = time.time()
        time.sleep(0.5)
        llm_ms = (time.time() - t0) * 1000
        prompt_tokens = 2800
        completion_tokens = 200
        cost = 0.0085
        spans.append(("llm_call", llm_ms))
        log.info(
            "span_completed",
            span="llm_call",
            duration_ms=round(llm_ms, 1),
            prompt_tokens=prompt_tokens,
            completion_tokens=completion_tokens,
            cost_usd=cost,
        )

        total_ms = (time.time() - request_start) * 1000

        # PILAR 1: Log completo del request
        log.info(
            "request_completed",
            total_ms=round(total_ms, 1),
            total_tokens=prompt_tokens + completion_tokens,
            cost_usd=cost,
        )

        # PILAR 2: Métricas agregadas
        self.metrics["latency_ms"].append(total_ms)
        self.metrics["cost_usd"].append(cost)
        self.metrics["tokens"].append(prompt_tokens + completion_tokens)

        # PILAR 3: Trace (spans)
        # En OTel real, los spans se exportan automáticamente.
        # Aquí los imprimimos como resumen.
        log.info(
            "trace_summary",
            spans=[
                {"name": name, "ms": round(ms, 1)} for name, ms in spans
            ],
            bottleneck=max(spans, key=lambda x: x[1])[0],
        )

        return trace_id

    def get_metrics_snapshot(self) -> dict:
        latencies = sorted(self.metrics["latency_ms"])
        costs = self.metrics["cost_usd"]
        return {
            "request_count": len(latencies),
            "latency_p50_ms": round(latencies[len(latencies) // 2], 1)
            if latencies
            else 0,
            "latency_p99_ms": round(latencies[int(len(latencies) * 0.99)], 1)
            if latencies
            else 0,
            "total_cost_usd": round(sum(costs), 4),
            "avg_cost_per_request": round(sum(costs) / len(costs), 6)
            if costs
            else 0,
        }


obs = CorrelatedObservability()

print("--- Simulando 3 requests ---\n")
for i in range(3):
    trace_id = obs.handle_request(
        query=f"Pregunta #{i+1}",
        user_id=f"user_{i % 2 + 1:03d}",
        endpoint="/api/chat",
    )
    print()

print("=" * 60)
print("METRICS SNAPSHOT (lo que ves en un dashboard)")
print("=" * 60)
snapshot = obs.get_metrics_snapshot()
for k, v in snapshot.items():
    print(f"  {k}: {v}")

Observa cómo el trace_id aparece en cada log. Eso es la correlación: cuando una métrica te alerta de un problema, buscas los trace_ids de los requests afectados, y con esos IDs puedes encontrar cada log individual. Sin ese hilo conector, cada pilar vive aislado.

Comparación: Pilares en Software Tradicional vs AI

La columna de AI no reemplaza la tradicional — la extiende. Sigues necesitando logs de HTTP, métricas de infraestructura, y traces de red. Pero sin la capa AI-específica, tu observabilidad tiene un agujero exactamente donde más la necesitas.

Conexión con Proyecto

Cómo los pilares alimentan tu Observability Assessment

El proyecto del módulo (cápsula 08) es un Observability Assessment. Para cada pilar, evalúas tu sistema:

PILAR: LOGS
  ¿Capturo prompts enviados?                    □ Sí  □ No
  ¿Capturo tokens por request?                  □ Sí  □ No
  ¿Mis logs son estructurados (JSON)?            □ Sí  □ No
  ¿Puedo filtrar logs por modelo/endpoint/user?  □ Sí  □ No

PILAR: MÉTRICAS
  ¿Mido TTFT y TTI por separado?                □ Sí  □ No
  ¿Calculo costo por request en USD?             □ Sí  □ No
  ¿Tengo métricas de calidad de output?          □ Sí  □ No
  ¿Sé qué endpoint es más costoso?               □ Sí  □ No

PILAR: TRACES
  ¿Puedo ver el flujo completo de un request?    □ Sí  □ No
  ¿Cada paso tiene su propia latencia?           □ Sí  □ No
  ¿Puedo identificar el bottleneck de un request? □ Sí  □ No
  ¿Tengo trace_id para correlacionar pilares?     □ Sí  □ No

Si la mayoría de tus respuestas son "No", sabes exactamente dónde están tus gaps. Los módulos 2-3 te dan las herramientas para convertir esos "No" en "Sí".

Troubleshooting

Problema: "Mis logs son demasiado grandes por los prompts"

Síntoma: El volumen de logs se multiplicó después de loguear prompts y completions.

Solución: No loguees contenido completo en producción por defecto. Loguea metadata (tokens, costo, latencia) siempre, y contenido solo en modo debug o para un sampling controlado (por ejemplo, 1 de cada 100 requests). Usa el patrón safe_log_content de esta cápsula.

Problema: "No sé cómo calcular el costo en USD"

Síntoma: Tienes tokens pero no dinero. El dashboard muestra "3,400 prompt tokens" pero no qué significa en USD.

Solución: Mantén una tabla de costos por modelo actualizada (como COST_PER_1K en el código de esta cápsula). Calcula costo en el momento del log y registra cost_usd como campo. Los proveedores publican sus precios — la conversión es aritmética simple.

Problema: "Mis traces no muestran el LLM call como un span separado"

Síntoma: El trace muestra el request entero como un solo bloque. No puedes ver qué paso tomó más tiempo.

Solución: Cada operación significativa (embedding, vector search, LLM call, validation) debe ser un span separado. Si usas una librería que hace todo "dentro de una función", necesitas instrumentar esa función internamente. OpenTelemetry (módulo 3) facilita esto.

Problema: "No puedo correlacionar un log con un trace"

Síntoma: Tienes logs y tienes traces, pero cuando una métrica alerta un problema, no puedes ir del dashboard al log específico.

Solución: Asegúrate de que cada log incluya el trace_id. Structlog con un bound logger (como en los ejemplos) hace esto automático. El trace_id es el hilo que conecta métricas → traces → logs.

Problema: "Los tres pilares parecen redundantes"

Síntoma: Sientes que estás registrando la misma información tres veces.

Solución: No es redundancia — son perspectivas diferentes. El log tiene el detalle de un request individual. La métrica es un número agregado de miles de requests. El trace es el flujo temporal de un request. Sin alguno de los tres, te falta una dimensión de análisis. Piensa en ello como latitud, longitud y altitud: los tres números describen un punto, pero ninguno es redundante.

Ejercicios

Ejercicio 1: Identifica los campos AI-específicos

Dado este log de un sistema AI, identifica qué campos son específicos de AI (no existirían en un backend REST convencional):

{
  "timestamp": "2026-03-08T14:23:02.567Z",
  "level": "info",
  "event": "request_completed",
  "method": "POST",
  "path": "/api/chat",
  "status_code": 200,
  "model": "gpt-4o-mini",
  "prompt_tokens": 1250,
  "completion_tokens": 340,
  "temperature": 0.7,
  "latency_ms": 2100,
  "cost_usd": 0.000391,
  "user_id": "usr_abc",
  "system_prompt_version": "v3.2",
  "rag_chunks_included": 4,
  "finish_reason": "stop"
}

Ejercicio 2: Diseña métricas para un chatbot

Tienes un chatbot que responde preguntas sobre documentación de producto. Define al menos 8 métricas que deberías capturar, categorizadas por pilar. Para cada métrica, indica: nombre, tipo (counter, gauge, histogram), y por qué importa.

MÉTRICAS DE LATENCIA
────────────────────
1. chatbot_ttft_ms (histogram)
   Time to first token. Afecta percepción de velocidad.
   Labels: model, endpoint

2. chatbot_e2e_latency_ms (histogram)
   Latencia total del request incluyendo RAG.
   Labels: model, endpoint, has_rag_context

3. chatbot_embedding_latency_ms (histogram)
   Latencia del paso de embedding para búsqueda.
   Labels: embedding_model

MÉTRICAS DE COSTO
─────────────────
4. chatbot_tokens_total (counter)
   Total de tokens consumidos (prompt + completion).
   Labels: model, token_type (prompt/completion), endpoint

5. chatbot_cost_usd_total (counter)
   Costo acumulado en USD.
   Labels: model, endpoint

6. chatbot_rag_chunks_per_request (histogram)
   Chunks de contexto incluidos por request.
   Indica eficiencia del retrieval.

MÉTRICAS DE CALIDAD
───────────────────
7. chatbot_responses_total (counter)
   Total de respuestas generadas.
   Labels: finish_reason (stop/length/error), model

8. chatbot_hallucination_detected_total (counter)
   Respuestas donde se detectó hallucination.
   Labels: detection_method, severity

MÉTRICAS DE ERROR
─────────────────
9. chatbot_errors_total (counter)
   Errores totales por tipo.
   Labels: error_type (rate_limit/timeout/invalid_output/model_error)

10. chatbot_retries_total (counter)
    Reintentos necesarios.
    Labels: retry_reason, model

Ejercicio 3: Dibuja el trace de tu sistema

Toma el sistema AI con el que trabajas (o uno que conozcas) y dibuja el trace completo de un request. Para cada span, indica:

• Nombre del paso
• Duración estimada
• Atributos AI-relevantes que capturarías

[Trace: request a chatbot RAG con tool calls]

  └── POST /api/chat ................................. ~5,500ms total
       ├── auth_validation ........................... ~10ms
       │    └── user_id, auth_method
       │
       ├── input_moderation .......................... ~200ms
       │    └── model: gpt-4o-mini, flagged: false, categories: []
       │
       ├── query_embedding ........................... ~80ms
       │    └── model: text-embedding-3-small, tokens: 32, dimensions: 1536
       │
       ├── vector_search ............................. ~150ms
       │    └── index: product_docs, top_k: 10, results: 10, min_score: 0.78
       │
       ├── context_ranking ........................... ~300ms
       │    └── model: gpt-4o-mini, chunks_in: 10, chunks_out: 4
       │    └── tokens_used: 800, strategy: llm_reranking
       │
       ├── prompt_assembly ........................... ~5ms
       │    └── system_v: 3.1, context_tokens: 2400, total_tokens: 2600
       │
       ├── llm_call_main ............................. ~3,200ms
       │    └── model: gpt-4o, temp: 0.3, prompt_tk: 2600, completion_tk: 180
       │    └── cost: $0.0083, ttft: 580ms, finish: tool_calls
       │
       ├── tool_execution ............................ ~800ms
       │    └── tool: get_pricing, args: {product: "pro"}, result_tokens: 45
       │
       ├── llm_call_final ............................ ~650ms
       │    └── model: gpt-4o, prompt_tk: 2825, completion_tk: 120
       │    └── cost: $0.0083, finish: stop
       │
       └── output_validation ......................... ~50ms
            └── format_ok: true, hallucination_check: pass

Ejercicio 4: Implementa el cálculo de costo multi-modelo

Extiende la función calculate_cost para soportar al menos 5 modelos diferentes (incluyendo modelos de Anthropic y modelos de embedding). Agrégale soporte para cache discount (algunos providers cobran menos si el prompt está en cache).

from dataclasses import dataclass


@dataclass
class ModelPricing:
    prompt_per_1k: float
    completion_per_1k: float
    cached_prompt_per_1k: float | None = None


PRICING_TABLE: dict[str, ModelPricing] = {
    "gpt-4o": ModelPricing(
        prompt_per_1k=0.0025,
        completion_per_1k=0.01,
        cached_prompt_per_1k=0.00125,
    ),
    "gpt-4o-mini": ModelPricing(
        prompt_per_1k=0.00015,
        completion_per_1k=0.0006,
        cached_prompt_per_1k=0.000075,
    ),
    "claude-sonnet-4-20250514": ModelPricing(
        prompt_per_1k=0.003,
        completion_per_1k=0.015,
        cached_prompt_per_1k=0.0003,
    ),
    "claude-3-5-haiku-20241022": ModelPricing(
        prompt_per_1k=0.0008,
        completion_per_1k=0.004,
        cached_prompt_per_1k=0.00008,
    ),
    "text-embedding-3-small": ModelPricing(
        prompt_per_1k=0.00002,
        completion_per_1k=0.0,
    ),
}


def calculate_cost_v2(
    model: str,
    prompt_tokens: int,
    completion_tokens: int,
    cached_prompt_tokens: int = 0,
) -> dict:
    pricing = PRICING_TABLE.get(model)
    if pricing is None:
        return {
            "cost_usd": 0.0,
            "warning": f"Unknown model: {model}. Cost not calculated.",
        }

    non_cached_prompt = prompt_tokens - cached_prompt_tokens
    prompt_cost = non_cached_prompt / 1000 * pricing.prompt_per_1k
    completion_cost = completion_tokens / 1000 * pricing.completion_per_1k

    cache_cost = 0.0
    cache_savings = 0.0
    if cached_prompt_tokens > 0 and pricing.cached_prompt_per_1k is not None:
        cache_cost = cached_prompt_tokens / 1000 * pricing.cached_prompt_per_1k
        full_price = cached_prompt_tokens / 1000 * pricing.prompt_per_1k
        cache_savings = full_price - cache_cost

    total = prompt_cost + completion_cost + cache_cost

    return {
        "cost_usd": round(total, 8),
        "prompt_cost_usd": round(prompt_cost, 8),
        "completion_cost_usd": round(completion_cost, 8),
        "cache_cost_usd": round(cache_cost, 8),
        "cache_savings_usd": round(cache_savings, 8),
        "model": model,
    }


# Test
print(calculate_cost_v2("gpt-4o", 3400, 280))
print(calculate_cost_v2("claude-sonnet-4-20250514", 5000, 400, cached_prompt_tokens=3000))
print(calculate_cost_v2("text-embedding-3-small", 512, 0))

{'cost_usd': 0.0113, 'prompt_cost_usd': 0.0085, 'completion_cost_usd': 0.0028,
 'cache_cost_usd': 0.0, 'cache_savings_usd': 0.0, 'model': 'gpt-4o'}

{'cost_usd': 0.0129, 'prompt_cost_usd': 0.006, 'completion_cost_usd': 0.006,
 'cache_cost_usd': 0.0009, 'cache_savings_usd': 0.0081,
 'model': 'claude-sonnet-4-20250514'}

{'cost_usd': 1.024e-05, 'prompt_cost_usd': 1.024e-05, 'completion_cost_usd': 0.0,
 'cache_cost_usd': 0.0, 'cache_savings_usd': 0.0,
 'model': 'text-embedding-3-small'}

Ejercicio 5: De alerta a root cause

Recibes esta alerta de métricas a las 3pm:

⚠️ ALERT: latency_p99 > 8000ms (threshold: 5000ms)
   Endpoint: /api/summarize
   Window: últimos 15 minutos
   Current value: 8,432ms

Describe paso a paso cómo usarías los tres pilares para llegar al root cause. Indica qué buscarías en cada pilar, qué filtros aplicarías, y qué conclusiones podrías sacar.

Consulta: latency_p99 para /api/summarize, últimas 2 horas, desglosado por modelo

Resultado: latency subió de 2,100ms a 8,400ms hace ~18 minutos.
Solo afecta el modelo gpt-4o. gpt-4o-mini sigue normal.

Consulta adicional: tokens_per_request promedio para /api/summarize, misma ventana

Resultado: tokens promedio subió de 1,200 a 6,800 hace ~18 minutos.

Conclusión parcial: el problema es que los requests están enviando muchos más
tokens al modelo. No es un problema del modelo en sí (no hay cambio en TTFT
normalizado por tokens).

Filtro: traces donde endpoint=/api/summarize AND total_duration > 5000ms
         en los últimos 20 minutos

Resultado: 47 traces coinciden.

Inspección de trace representativo:
  - embedding: 80ms (normal)
  - vector_search: 180ms (normal)
  - context_construction: 12ms (normal)
    → PERO: context_tokens = 5,800 (normalmente es ~800)
    → chunks_selected = 12 (normalmente es 3-4)
  - llm_call: 7,800ms
    → prompt_tokens = 6,200 (normalmente ~1,000)

Conclusión parcial: context_construction está seleccionando demasiados chunks.
El problema no está en el LLM ni en el vector search — está en la lógica
de selección de contexto.

Filtro: logs donde trace_id IN (traces del paso 2) AND span="context_construction"

Resultado: logs muestran:
  {"event": "context_selection", "strategy": "include_all",
   "config_version": "v4.0", "chunks_in": 12, "chunks_out": 12}

¡La estrategia cambió de "relevance_threshold_0.85" a "include_all"!

Filtro adicional: logs donde event="config_change" en las últimas 2 horas

Resultado:
  {"event": "config_change", "key": "rag.context_strategy",
   "old_value": "relevance_threshold_0.85", "new_value": "include_all",
   "changed_by": "deploy_pipeline", "timestamp": "2026-03-08T14:42:00Z"}

ROOT CAUSE: Un deploy hace 18 minutos cambió la estrategia de selección
de contexto. Ahora incluye TODOS los chunks del vector search en lugar
de filtrar por relevancia. Esto multiplicó los tokens enviados al LLM,
causando latencia alta y probablemente costo elevado.

Ejercicio 6: Audita un pilar de tu propio sistema

Elige uno de los tres pilares (logs, métricas, o traces) y realiza una auditoría rápida de tu sistema AI. Responde:

1. ¿Qué datos capturas actualmente en este pilar?
2. ¿Qué datos AI-específicos te faltan?
3. ¿Qué preguntas no puedes responder por esa falta?
4. ¿Qué necesitarías implementar para cerrar los gaps?

AUDITORÍA DE PILAR: LOGS
═════════════════════════

1. QUÉ CAPTURO HOY:
   ✅ Timestamp de cada request
   ✅ HTTP method y path
   ✅ Status code de respuesta
   ✅ Errores con stack trace
   ❌ No capturo en formato estructurado (uso print/logging básico)

2. QUÉ ME FALTA (AI-ESPECÍFICO):
   ❌ Modelo utilizado por request
   ❌ Tokens (prompt + completion)
   ❌ Costo en USD
   ❌ Latencia de la llamada LLM (solo tengo end-to-end)
   ❌ Temperature y otros parámetros
   ❌ System prompt version
   ❌ Contenido del prompt (ni siquiera hash)
   ❌ Chunks de RAG incluidos

3. PREGUNTAS QUE NO PUEDO RESPONDER:
   - "¿Cuánto costó el request más caro de ayer?"
   - "¿Qué prompt generó la respuesta incorrecta que reportó el usuario?"
   - "¿Cambió el system prompt entre ayer y hoy?"
   - "¿Cuántos tokens promedio consume /api/chat vs /api/summarize?"
   - "¿Hay requests que fallan silenciosamente (200 OK pero output vacío)?"

4. QUÉ NECESITO IMPLEMENTAR:
   a. Migrar de print()/logging a structlog (1-2 horas)
   b. Agregar campos AI en cada log de LLM call (30 min)
   c. Implementar calculate_cost para registrar USD (30 min)
   d. Agregar safe_log_content para prompts (30 min)
   e. Configurar un log aggregator básico (para buscar/filtrar)

   PRIORIDAD: (a) y (b) primero — sin logs estructurados con campos AI,
   el resto de los pilares tampoco funcionará bien.

Resumen

• Los tres pilares de observabilidad (logs, métricas, traces) aplican a sistemas AI, pero su contenido cambia radicalmente respecto a software tradicional.
• Logs en AI necesitan capturar prompt metadata, tokens, costo, modelo, y parámetros — no solo eventos HTTP. Structured logging con structlog te da campos buscables y filtrables.
• Métricas en AI incluyen dimensiones que no existen en backend convencional: TTFT, TTI, tokens por request, costo en USD, hallucination rate, quality score.
• Traces en AI son más complejos: un request típico pasa por validación → embedding → vector search → context construction → LLM call → output validation. Cada paso es un span con atributos AI-relevantes.
• Los pilares no son independientes — son tres perspectivas del mismo sistema. Una métrica alerta, un trace localiza, un log explica. Sin correlación (via trace_id), cada pilar da fragmentos sin contexto.
• El ciclo de investigación fluye: alerta de métrica → filtro de traces → detalle en logs → root cause identificado. Practica ese flujo antes de que lo necesites a las 2am.
• No loguees todo: metadata siempre, contenido de prompts/completions con cuidado (privacidad, volumen). Usa sampling y hashing.
• El cálculo de costo es aritmética simple pero requiere una tabla de precios actualizada por modelo. Registra cost_usd en cada log — es la métrica que más impacta decisiones de negocio.

Recursos Adicionales

1. OpenTelemetry — Logs, Metrics, Traces — Definición oficial de los tres pilares como "signals" en OTel
2. OpenTelemetry Semantic Conventions for GenAI — Convenciones de atributos para spans de LLM (model, tokens, etc.)
3. structlog Documentation — Referencia completa de la librería usada en esta cápsula
4. Google SRE — Monitoring Distributed Systems — El capítulo clásico sobre los cuatro golden signals
5. Charity Majors — Observability Engineering, Chapter 3 — Logs, metrics, traces como perspectivas correlacionadas
6. OpenAI Pricing — Precios actualizados para calcular cost_usd
7. Anthropic API Pricing — Precios de Claude incluyendo cache discounts
8. Distributed Tracing in Practice (O'Reilly) — Libro de referencia sobre tracing distribuido