Módulo 1: Observability para AI Systems

5. Gaps en Sistemas AI sin Observabilidad

Descripción de la cápsula

Hasta ahora sabes qué es observabilidad, por qué AI necesita un enfoque diferente, y cuáles son los tres pilares reinterpretados para sistemas AI. Esta cápsula hace algo diferente: en lugar de mostrarte qué deberías tener, te muestra qué no puedes hacer sin observabilidad. El enfoque es revelador, no aspiracional.

La idea es simple pero poderosa. Toma las preguntas que cualquier equipo necesita responder en producción — sobre costos, calidad, latencia, debugging — y demuestra que sin instrumentación adecuada, cada una de esas preguntas queda sin respuesta. No porque sean preguntas difíciles, sino porque los datos para responderlas nunca se capturaron.

Vas a ver código real: primero una app AI que opera como la mayoría opera hoy (con print() y esperanza), y luego esa misma app con instrumentación básica que ya te permite responder preguntas que antes eran imposibles. La diferencia no es un stack de observabilidad sofisticado — es agregar 20 líneas de código en los lugares correctos. Al final, tendrás un checklist que puedes ejecutar contra tu propio sistema para descubrir exactamente dónde están tus puntos ciegos.


Las Cuatro Categorías de Ceguera Operativa

Cuando tu sistema AI no tiene observabilidad, los gaps se agrupan en cuatro categorías. Cada una representa un tipo de pregunta que necesitas responder en producción pero no puedes.

Categoría        Qué no puedes ver                     Impacto
───────────────────────────────────────────────────────────────────────
Costo            Cuánto gastas, en qué, por quién      Facturas sorpresa
Calidad          Si los outputs son correctos           Usuarios insatisfechos
Latencia         Qué está lento y por qué               UX degradada
Debugging        Qué pasó en un request específico      Incidentes sin resolver

No son categorías teóricas. Son las cuatro conversaciones que van a ocurrir en tu equipo cuando algo falle en producción. Y en cada una, la respuesta honesta sin observabilidad es: "no sé".


Gaps de Costo: La Factura Invisible

Preguntas que no puedes responder

Pregunta                                        Sin observabilidad
──────────────────────────────────────────────────────────────────────
¿Cuánto gastamos en tokens ayer?                 "Ni idea. Reviso la factura
                                                  de OpenAI a fin de mes."

¿Cuál endpoint es más caro?                      "Probablemente /api/chat...
                                                  pero no tengo datos."

¿Cuánto cuesta servir a un usuario por día?       "No tracking por usuario."

¿Un solo usuario está quemando el presupuesto?    "No sé. ¿Cómo sabría?"

¿Cuánto ahorraríamos cambiando de gpt-4o a       "Tendría que estimar
 gpt-4o-mini en este endpoint?                     manualmente."

¿Los prompts de RAG están creciendo en tokens?    "No mido tokens de contexto."

Por qué los gaps de costo son peligrosos

En software tradicional, el costo por request es relativamente fijo: CPU, memoria, ancho de banda. Si tienes 10,000 requests/día, el costo es predecible. En AI, un request puede costar $0.001 (pregunta simple con gpt-4o-mini) o $2.00+ (contexto RAG largo con gpt-4o y tool calls). Esa variabilidad de 2000x hace que los gaps de costo sean particularmente peligrosos.

Escenarios reales que ocurren sin tracking de costos:

  • 📈 Prompt creep: Alguien agrega "contexto adicional" al system prompt. Pasa de 200 tokens a 2,000. Nadie nota que el costo se multiplicó por 5 porque nadie mide tokens.
  • 🔄 Retry storms: Un rate limit causa retries. Cada retry es un request billable. Sin métricas de retries, pagas 3x por requests que ya fallaron.
  • 👤 Whale users: Un solo usuario envía 500 requests al día con contextos de 8,000 tokens. Cuesta más que los otros 999 usuarios combinados. Sin cost-per-user, no lo detectas.
  • 🔀 Model misrouting: Un endpoint debería usar gpt-4o-mini pero alguien lo configuró con gpt-4o. Sin tracking de modelo por endpoint, pagas 15x sin razón.

El costo de no saber el costo

COSTO_MENSUAL_ESTIMADO = {
    "sin_tracking": {
        "gasto_real_mensual_usd": 2400,
        "gasto_que_crees_usd": "~500 (estimación de servilleta)",
        "desperdicio_detectable_usd": 0,
        "accion_posible": "Revisar factura a fin de mes y sorprenderte",
    },
    "con_tracking_basico": {
        "gasto_real_mensual_usd": 2400,
        "gasto_que_crees_usd": 2400,
        "desperdicio_detectable_usd": 1200,
        "accion_posible": (
            "Identificar que /api/summarize cuesta 60% del total, "
            "cambiar a gpt-4o-mini (-80% costo), reducir chunks de RAG de 10 a 4 (-50%)"
        ),
    },
}

Gaps de Calidad: Outputs que Nadie Valida

Preguntas que no puedes responder

Pregunta                                        Sin observabilidad
──────────────────────────────────────────────────────────────────────
¿La calidad de outputs está degradando?          "No mido calidad. Si nadie
                                                  se queja, asumo que está bien."

¿Cuántas hallucinations hubo esta semana?         "No tengo detección de
                                                  hallucinations."

¿El model update afectó nuestros outputs?         "No puedo comparar antes
                                                  vs después."

¿Qué porcentaje de respuestas es útil             "No tengo feedback
 para el usuario?                                  del usuario."

¿El contexto RAG está ayudando o perjudicando?    "No sé si los chunks que
                                                  incluyo son relevantes."

Por qué los gaps de calidad son los más silenciosos

Un error HTTP 500 es ruidoso: tu monitoring lo detecta, tus alertas disparan, tu equipo responde. Una hallucination es silenciosa: el sistema devuelve 200 OK, el output se ve bien formateado, el usuario recibe una respuesta... que es incorrecta. Si el usuario no reporta (la mayoría no lo hace — simplemente deja de usar el producto), esa hallucination es invisible.

Error HTTP 500          →  Ruidoso  →  Detectable sin observabilidad AI
Timeout                 →  Ruidoso  →  Detectable con monitoring básico
Respuesta lenta         →  Notable  →  Detectable con métricas de latencia
Hallucination           →  SILENCIOSA  →  INDETECTABLE sin quality metrics
Output parcialmente     →  SILENCIOSA  →  INDETECTABLE sin quality metrics
  incorrecto
Respuesta correcta pero →  SILENCIOSA  →  INDETECTABLE sin quality metrics
  no útil
Drift de calidad        →  SILENCIOSA  →  INDETECTABLE sin baseline + tracking
  gradual

La ironía es que los problemas más dañinos para tu producto (respuestas incorrectas, hallucinations, degradación de calidad) son los que menos probabilidad tienen de ser detectados sin instrumentación.

Model drift: el enemigo invisible

Semana 1: Modelo X versión A → calidad baseline = 8.5/10
Semana 3: Proveedor actualiza modelo → calidad baja a 7.2/10
Semana 5: Usuarios se quejan → tú descubres que algo cambió
Semana 6: Investigas → no tienes datos históricos de calidad
Semana 7: "Pruebas a mano" → no puedes comparar con el baseline

CON OBSERVABILIDAD:
Semana 1: Modelo X versión A → calidad baseline = 8.5/10 [registrado]
Semana 3: Proveedor actualiza modelo → calidad baja a 7.2/10 [detectado]
Semana 3: Alerta dispara → ves exactamente cuándo empezó
Semana 3: Comparas outputs antes/después → identificas qué cambió
Semana 3: Decides: rollback de modelo, ajustar prompt, o aceptar

La diferencia no es tecnología sofisticada. Es haber registrado un quality_score por request y tener un baseline contra el cual comparar.


Gaps de Latencia: No Saber Qué Está Lento

Preguntas que no puedes responder

Pregunta                                        Sin observabilidad
──────────────────────────────────────────────────────────────────────
¿Qué está causando las respuestas lentas?        "¿El LLM? ¿El embedding?
                                                  ¿El vector search? No sé."

¿Cuál es nuestro TTFT vs TTI?                    "Solo mido end-to-end
                                                  (si acaso)."

¿Qué modelo es más rápido para este caso?        "No tengo datos comparativos
                                                  por modelo."

¿La latencia empeoró después del deploy?          "No sé, no tengo baseline."

¿Cuánto tiempo se gasta en el pipeline            "No tengo desglose
 antes del LLM?                                    por paso."

El problema del "end-to-end único"

La mayoría de equipos que miden algo de latencia miden solo end-to-end: cuánto tiempo desde que llega el request hasta que sale el response. Eso es como medir la duración de un vuelo sin saber cuánto fue boarding, cuánto fue taxiing, cuánto fue vuelo, y cuánto fue desembarque.

LO QUE MIDES:
  Request → → → → → → → → → → → → → Response
  |_______________ 4,200ms _________________|

LO QUE NECESITAS:
  Request →
    validation:     |── 5ms ──|
    embedding:      |──── 120ms ────|
    vector_search:  |──────── 230ms ────────|
    context:        |── 15ms ──|
    prompt:         |─ 2ms ─|
    llm_call:       |────────────────── 3,750ms ──────────────────|
    validation:     |──── 78ms ────|
                                                        → Response

Con end-to-end único, si la latencia sube, tu diagnóstico es: "está lento". Con desglose por paso, tu diagnóstico es: "el LLM call subió de 800ms a 3,750ms, probablemente por aumento de tokens en el prompt". Uno te deja adivinando. El otro te da una causa raíz.

TTFT: la métrica que olvidaste

Time To First Token es la métrica que más impacta la percepción del usuario en interfaces de streaming. Si TTFT es alto, el usuario ve "cargando..." durante 3 segundos antes de que aparezca el primer carácter. Si TTFT es bajo pero TTI es alto, el usuario ve texto fluyendo y se siente rápido aunque la respuesta total tarde.

TTFT bajo, TTI alto:     [........|████████████████████████]
  Percepción: "Rápido, genera mucho texto"

TTFT alto, TTI bajo:     [████████████████|██████]
  Percepción: "Lento, luego aparece de golpe"

TTFT alto, TTI alto:     [████████████████|████████████████████████]
  Percepción: "Lento en todo"

Sin métricas separadas de TTFT y TTI, no puedes distinguir entre estos escenarios ni optimizar para la percepción del usuario.


Gaps de Debugging: No Poder Investigar

Preguntas que no puedes responder

Pregunta                                        Sin observabilidad
──────────────────────────────────────────────────────────────────────
¿Qué prompt generó esa respuesta incorrecta?     "No logueo prompts."

¿Puedo reproducir la experiencia de este          "No, es no-determinístico
 usuario?                                          y no guardé el estado."

¿Esto afecta a un usuario o a muchos?             "No puedo hacer queries
                                                  sobre mis datos."

¿Cuándo empezó este problema?                     "No sé, no tengo
                                                  datos históricos."

¿Fue un cambio de prompt, de modelo, o            "No versiono prompts
 de datos lo que causó esto?                       ni trackeo cambios."

El escenario de debugging sin datos

Un usuario reporta: "El chatbot me dijo que el plan Pro cuesta $50/mes, pero en realidad cuesta $500/mes." Aquí tienes tu proceso de debugging con y sin observabilidad:

SIN OBSERVABILIDAD:
1. Abres los logs. Solo ves: "POST /api/chat 200 OK 2.3s"
2. No sabes qué prompt se envió.
3. No sabes qué contexto RAG recibió el modelo.
4. Intentas reproducir el bug: envías "¿cuánto cuesta el plan Pro?"
5. El modelo responde "$500/mes" — correcto esta vez.
6. No puedes reproducir el error. No puedes saber la causa.
7. Cierras el ticket: "No reproducible."
8. El problema sigue ocurriendo. Otros usuarios se van sin reportar.

CON OBSERVABILIDAD:
1. Buscas el trace del request del usuario (por user_id o timestamp).
2. Ves el prompt: incluía contexto de una página web vieja con el precio anterior.
3. Ves que el vector search devolvió un documento de 2024 con precio desactualizado.
4. Buscas traces similares: 47 requests en la última semana incluyeron ese documento.
5. Causa raíz: el índice de vectores tiene documentos desactualizados.
6. Fix: actualizar el índice, agregar filtro de fecha a la búsqueda.
7. Verificas: después del fix, 0 requests devuelven el precio incorrecto.

La diferencia no es inteligencia del debugger. Es acceso a datos.


Código: Una App AI sin Observabilidad vs Con Observabilidad

Versión 1: Zero observability (el estado actual de muchos sistemas)

"""
ai_app_v1_no_observability.py

Una app AI típica sin ninguna instrumentación.
Funciona, pero opera completamente a ciegas.
"""

import time
from openai import OpenAI
from dotenv import load_dotenv

load_dotenv()
client = OpenAI()


def get_rag_context(query: str) -> str:
    # Simula búsqueda en vector store
    time.sleep(0.2)
    return (
        "El plan Starter cuesta $10/mes. "
        "El plan Pro cuesta $500/mes. "
        "El plan Enterprise es custom pricing."
    )


def chat(user_message: str) -> str:
    print(f"Received: {user_message}")

    context = get_rag_context(user_message)

    response = client.chat.completions.create(
        model="gpt-4o-mini",
        messages=[
            {
                "role": "system",
                "content": f"Responde usando este contexto: {context}",
            },
            {"role": "user", "content": user_message},
        ],
        max_tokens=200,
    )

    answer = response.choices[0].message.content
    print(f"Response: {answer}")
    return answer


def summarize(text: str) -> str:
    print(f"Summarizing {len(text)} chars")

    response = client.chat.completions.create(
        model="gpt-4o",
        messages=[
            {"role": "system", "content": "Resume el siguiente texto en 2 oraciones."},
            {"role": "user", "content": text},
        ],
        max_tokens=100,
    )

    summary = response.choices[0].message.content
    print(f"Summary: {summary}")
    return summary


if __name__ == "__main__":
    chat("¿Cuánto cuesta el plan Pro?")
    chat("¿Qué incluye el plan Enterprise?")
    summarize("Este es un texto largo " * 200)

    print("\n--- Fin del demo ---")
    print("Preguntas que NO puedes responder:")
    print("  - ¿Cuánto costó cada request en USD?")
    print("  - ¿Cuántos tokens consumió cada request?")
    print("  - ¿Qué modelo se usó en cada endpoint?")
    print("  - ¿Cuál endpoint es más caro?")
    print("  - ¿Cuál fue la latencia del LLM vs la búsqueda RAG?")
    print("  - ¿El contexto RAG fue relevante?")
    print("  - ¿Cuánto gastaste en total hoy?")

Lo que ves en la terminal:

Received: ¿Cuánto cuesta el plan Pro?
Response: El plan Pro cuesta $500 al mes.
Received: ¿Qué incluye el plan Enterprise?
Response: El plan Enterprise tiene pricing personalizado...
Summarizing 4400 chars
Summary: El texto repite la frase...

--- Fin del demo ---
Preguntas que NO puedes responder:
  - ¿Cuánto costó cada request en USD?
  - ¿Cuántos tokens consumió cada request?
  - ¿Qué modelo se usó en cada endpoint?
  - ¿Cuál endpoint es más caro?
  ...

El sistema funciona. Los outputs son correctos (esta vez). Pero si alguien te pregunta "¿cuánto gastamos hoy?", tu respuesta es un encogimiento de hombros.

Versión 2: Con instrumentación básica

La misma app, pero ahora cada request se registra con los datos necesarios. No es un stack completo de observabilidad — es el mínimo viable que ya cambia todo:

"""
ai_app_v2_with_observability.py

La misma app, ahora con instrumentación básica.
Cada request registra: tokens, costo, latencia, modelo.
"""

import time
import uuid
import structlog
from dataclasses import dataclass, field
from collections import defaultdict
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.JSONRenderer(),
    ],
    wrapper_class=structlog.BoundLogger,
    context_class=dict,
    logger_factory=structlog.PrintLoggerFactory(),
)

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


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


@dataclass
class RequestMetrics:
    requests: list = field(default_factory=list)

    def record(self, data: dict):
        self.requests.append(data)

    def summary(self) -> dict:
        if not self.requests:
            return {}

        total_cost = sum(r["cost_usd"] for r in self.requests)
        total_tokens = sum(r["total_tokens"] for r in self.requests)

        cost_by_endpoint = defaultdict(float)
        tokens_by_endpoint = defaultdict(int)
        latency_by_endpoint = defaultdict(list)
        cost_by_model = defaultdict(float)

        for r in self.requests:
            ep = r["endpoint"]
            cost_by_endpoint[ep] += r["cost_usd"]
            tokens_by_endpoint[ep] += r["total_tokens"]
            latency_by_endpoint[ep].append(r["total_latency_ms"])
            cost_by_model[r["model"]] += r["cost_usd"]

        return {
            "total_requests": len(self.requests),
            "total_tokens": total_tokens,
            "total_cost_usd": round(total_cost, 6),
            "cost_by_endpoint": {
                k: round(v, 6) for k, v in sorted(
                    cost_by_endpoint.items(), key=lambda x: x[1], reverse=True
                )
            },
            "tokens_by_endpoint": dict(
                sorted(tokens_by_endpoint.items(), key=lambda x: x[1], reverse=True)
            ),
            "avg_latency_by_endpoint": {
                k: round(sum(v) / len(v), 1)
                for k, v in latency_by_endpoint.items()
            },
            "cost_by_model": {
                k: round(v, 6) for k, v in sorted(
                    cost_by_model.items(), key=lambda x: x[1], reverse=True
                )
            },
        }


metrics = RequestMetrics()


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 get_rag_context(query: str, trace_id: str) -> tuple[str, float]:
    log = logger.bind(trace_id=trace_id)
    start = time.time()
    time.sleep(0.2)
    context = (
        "El plan Starter cuesta $10/mes. "
        "El plan Pro cuesta $500/mes. "
        "El plan Enterprise es custom pricing."
    )
    rag_ms = (time.time() - start) * 1000
    log.info(
        "rag_search_completed",
        duration_ms=round(rag_ms, 1),
        context_length=len(context),
        chunks_returned=3,
    )
    return context, rag_ms


def chat(user_message: str, user_id: str = "anonymous") -> str:
    trace_id = str(uuid.uuid4())[:12]
    endpoint = "/api/chat"
    model = "gpt-4o-mini"
    log = logger.bind(trace_id=trace_id, endpoint=endpoint, user_id=user_id)

    log.info("request_started", query_length=len(user_message))
    request_start = time.time()

    context, rag_ms = get_rag_context(user_message, trace_id)

    llm_start = time.time()
    response = client.chat.completions.create(
        model=model,
        messages=[
            {
                "role": "system",
                "content": f"Responde usando este contexto: {context}",
            },
            {"role": "user", "content": user_message},
        ],
        max_tokens=200,
    )
    llm_ms = (time.time() - llm_start) * 1000
    total_ms = (time.time() - request_start) * 1000

    usage = response.usage
    cost = calculate_cost(model, usage.prompt_tokens, usage.completion_tokens)
    answer = response.choices[0].message.content

    log.info(
        "request_completed",
        model=model,
        prompt_tokens=usage.prompt_tokens,
        completion_tokens=usage.completion_tokens,
        total_tokens=usage.total_tokens,
        cost_usd=round(cost, 6),
        rag_latency_ms=round(rag_ms, 1),
        llm_latency_ms=round(llm_ms, 1),
        total_latency_ms=round(total_ms, 1),
        finish_reason=response.choices[0].finish_reason,
    )

    metrics.record({
        "trace_id": trace_id,
        "endpoint": endpoint,
        "model": model,
        "user_id": user_id,
        "prompt_tokens": usage.prompt_tokens,
        "completion_tokens": usage.completion_tokens,
        "total_tokens": usage.total_tokens,
        "cost_usd": cost,
        "rag_latency_ms": rag_ms,
        "llm_latency_ms": llm_ms,
        "total_latency_ms": total_ms,
    })

    return answer


def summarize(text: str, user_id: str = "anonymous") -> str:
    trace_id = str(uuid.uuid4())[:12]
    endpoint = "/api/summarize"
    model = "gpt-4o"
    log = logger.bind(trace_id=trace_id, endpoint=endpoint, user_id=user_id)

    log.info("request_started", input_length=len(text))
    request_start = time.time()

    llm_start = time.time()
    response = client.chat.completions.create(
        model=model,
        messages=[
            {"role": "system", "content": "Resume el siguiente texto en 2 oraciones."},
            {"role": "user", "content": text},
        ],
        max_tokens=100,
    )
    llm_ms = (time.time() - llm_start) * 1000
    total_ms = (time.time() - request_start) * 1000

    usage = response.usage
    cost = calculate_cost(model, usage.prompt_tokens, usage.completion_tokens)
    summary = response.choices[0].message.content

    log.info(
        "request_completed",
        model=model,
        prompt_tokens=usage.prompt_tokens,
        completion_tokens=usage.completion_tokens,
        total_tokens=usage.total_tokens,
        cost_usd=round(cost, 6),
        llm_latency_ms=round(llm_ms, 1),
        total_latency_ms=round(total_ms, 1),
        finish_reason=response.choices[0].finish_reason,
    )

    metrics.record({
        "trace_id": trace_id,
        "endpoint": endpoint,
        "model": model,
        "user_id": user_id,
        "prompt_tokens": usage.prompt_tokens,
        "completion_tokens": usage.completion_tokens,
        "total_tokens": usage.total_tokens,
        "cost_usd": cost,
        "rag_latency_ms": 0,
        "llm_latency_ms": llm_ms,
        "total_latency_ms": total_ms,
    })

    return summary


if __name__ == "__main__":
    chat("¿Cuánto cuesta el plan Pro?", user_id="user_001")
    chat("¿Qué incluye el plan Enterprise?", user_id="user_002")
    summarize("Este es un texto largo " * 200, user_id="user_001")

    print("\n" + "=" * 60)
    print("PREGUNTAS QUE AHORA SÍ PUEDES RESPONDER")
    print("=" * 60)
    s = metrics.summary()
    print(f"\n¿Cuánto gastamos en total?")
    print(f"  → ${s['total_cost_usd']} en {s['total_requests']} requests")
    print(f"\n¿Cuál endpoint es más caro?")
    for ep, cost in s["cost_by_endpoint"].items():
        print(f"  → {ep}: ${cost}")
    print(f"\n¿Cuántos tokens consume cada endpoint?")
    for ep, tokens in s["tokens_by_endpoint"].items():
        print(f"  → {ep}: {tokens} tokens")
    print(f"\n¿Cuál es la latencia promedio por endpoint?")
    for ep, lat in s["avg_latency_by_endpoint"].items():
        print(f"  → {ep}: {lat}ms")
    print(f"\n¿Cuánto cuesta cada modelo?")
    for model, cost in s["cost_by_model"].items():
        print(f"  → {model}: ${cost}")

Output esperado:

============================================================
PREGUNTAS QUE AHORA SÍ PUEDES RESPONDER
============================================================

¿Cuánto gastamos en total?
  → $0.014832 en 3 requests

¿Cuál endpoint es más caro?
  → /api/summarize: $0.014450
  → /api/chat: $0.000382

¿Cuántos tokens consume cada endpoint?
  → /api/summarize: 1180 tokens
  → /api/chat: 490 tokens

¿Cuál es la latencia promedio por endpoint?
  → /api/chat: 1450.3ms
  → /api/summarize: 2100.5ms

¿Cuánto cuesta cada modelo?
  → gpt-4o: $0.014450
  → gpt-4o-mini: $0.000382

Observa: /api/summarize con gpt-4o cuesta 37x más que /api/chat con gpt-4o-mini. Sin instrumentación, eso era invisible. Con 20 líneas extra de código, ahora es un dato que puede cambiar decisiones de producto.


El Gap Analysis Checklist

Este checklist te permite auditar tu propio sistema en 10 minutos. Para cada pregunta, responde honestamente si puedes contestarla hoy con datos (no con suposiciones).

Cómo usarlo

Ejecuta el checklist mentalmente o imprímelo. Cada "No" es un gap. Al final, cuenta los gaps por categoría para saber dónde priorizar.

"""
gap_analysis_checklist.py

Ejecuta este checklist contra tu sistema AI.
Cada "No" es un gap de observabilidad.
"""

CHECKLIST = {
    "COSTO": [
        "¿Puedes decir cuánto gastaste en tokens ayer?",
        "¿Sabes cuál endpoint es más caro en USD?",
        "¿Puedes calcular el costo de un request individual?",
        "¿Sabes cuánto cuesta servir a un usuario específico por día?",
        "¿Puedes comparar costo entre modelos para el mismo endpoint?",
        "¿Detectas si los tokens por request están creciendo?",
    ],
    "CALIDAD": [
        "¿Mides la calidad de los outputs de alguna forma?",
        "¿Puedes detectar hallucinations automáticamente?",
        "¿Tienes un baseline de calidad contra el cual comparar?",
        "¿Sabes si un model update degradó la calidad?",
        "¿Puedes comparar calidad entre modelos o prompts?",
        "¿Capturas feedback del usuario (thumbs up/down, rating)?",
    ],
    "LATENCIA": [
        "¿Mides TTFT (time to first token) por separado?",
        "¿Puedes desglosar latencia por paso del pipeline?",
        "¿Sabes si la latencia está empeorando con el tiempo?",
        "¿Puedes identificar el bottleneck de un request lento?",
        "¿Tienes SLOs de latencia definidos y monitoreados?",
        "¿Comparas latencia entre modelos?",
    ],
    "DEBUGGING": [
        "¿Puedes encontrar el prompt exacto de un request específico?",
        "¿Puedes reconstruir el contexto RAG que recibió el modelo?",
        "¿Puedes saber si un error afecta a un usuario o a muchos?",
        "¿Puedes rastrear un bug hasta un cambio de prompt/modelo/datos?",
        "¿Tienes trace_id que conecte logs, métricas y traces?",
        "¿Puedes responder '¿cuándo empezó este problema?'?",
    ],
}


def run_gap_analysis():
    print("=" * 60)
    print("GAP ANALYSIS — OBSERVABILIDAD AI")
    print("=" * 60)
    print("Responde 'sí' o 'no' para cada pregunta.\n")

    total_gaps = 0
    gaps_by_category = {}

    for category, questions in CHECKLIST.items():
        print(f"\n{'─' * 40}")
        print(f"  {category}")
        print(f"{'─' * 40}")

        category_gaps = 0
        for i, question in enumerate(questions, 1):
            while True:
                answer = input(f"  {i}. {question}\n     → (s/n): ").strip().lower()
                if answer in ("s", "n", "sí", "si", "no"):
                    break
                print("     Responde 's' o 'n'")

            is_gap = answer in ("n", "no")
            if is_gap:
                category_gaps += 1
                total_gaps += 1
                print(f"     ❌ GAP detectado")
            else:
                print(f"     ✅ Cubierto")

        gaps_by_category[category] = category_gaps

    total_questions = sum(len(qs) for qs in CHECKLIST.values())
    coverage = round((1 - total_gaps / total_questions) * 100, 1)

    print("\n" + "=" * 60)
    print("RESULTADOS")
    print("=" * 60)
    print(f"\nCobertura total: {coverage}% ({total_questions - total_gaps}/{total_questions})")
    print(f"Gaps totales: {total_gaps}\n")

    print("Gaps por categoría:")
    for cat, gaps in sorted(gaps_by_category.items(), key=lambda x: x[1], reverse=True):
        total_in_cat = len(CHECKLIST[cat])
        bar = "█" * gaps + "░" * (total_in_cat - gaps)
        status = "🔴 CRÍTICO" if gaps >= 4 else "🟡 MEDIO" if gaps >= 2 else "🟢 OK"
        print(f"  {cat:.<15} {gaps}/{total_in_cat} gaps [{bar}] {status}")

    print("\nPrioridad de instrumentación sugerida:")
    priority = sorted(gaps_by_category.items(), key=lambda x: x[1], reverse=True)
    for i, (cat, gaps) in enumerate(priority, 1):
        if gaps > 0:
            print(f"  {i}. {cat} ({gaps} gaps)")

    return gaps_by_category


if __name__ == "__main__":
    run_gap_analysis()

Output de ejemplo (para un sistema típico sin observabilidad):

============================================================
RESULTADOS
============================================================

Cobertura total: 16.7% (4/24)
Gaps totales: 20

Gaps por categoría:
  COSTO.......... 6/6 gaps [██████] 🔴 CRÍTICO
  CALIDAD........ 6/6 gaps [██████] 🔴 CRÍTICO
  DEBUGGING...... 5/6 gaps [█████░] 🔴 CRÍTICO
  LATENCIA....... 3/6 gaps [███░░░] 🟡 MEDIO

Prioridad de instrumentación sugerida:
  1. COSTO (6 gaps)
  2. CALIDAD (6 gaps)
  3. DEBUGGING (5 gaps)
  4. LATENCIA (3 gaps)

Si tu resultado se parece a esto, no te alarmes — es el punto de partida de la mayoría de equipos. Lo que importa es que ahora tienes un diagnóstico específico en lugar de una sensación vaga de "debería mejorar algo".


Comparación: App sin Observabilidad vs Con Instrumentación Básica

AspectoSin observabilidadCon instrumentación básica
Costo visibleSolo la factura mensual del proveedorCosto por request, endpoint, usuario, modelo
Debuggingprint() + leer logs manualmenteBuscar por trace_id, filtrar por endpoint/user
Calidad"Si nadie se queja, está bien"Baseline + tracking (aunque sea manual)
LatenciaEnd-to-end total (si acaso)Desglose por paso: RAG, LLM, validación
Incidente nocturno"Déjame ver los logs a ver qué encuentro""Déjame buscar el trace de ese request"
Decisiones de modelo"Creo que gpt-4o-mini es suficiente""Los datos muestran que gpt-4o-mini tiene 95% quality para este endpoint"
EscalaLos problemas crecen silenciosamenteLos problemas se detectan antes de ser críticos
Esfuerzo de implementación0 líneas~50-100 líneas con structlog + métricas básicas
Tiempo de debuggingHoras (si tienes suerte)Minutos (con trace_id)

La columna derecha no es un stack sofisticado de observabilidad. Es lo que obtienes con structlog, un cálculo de costo, y trace_ids. El salto de "nada" a "algo" es el más grande en valor por línea de código.


Conexión con Proyecto

Del checklist al Observability Assessment

El Gap Analysis Checklist de esta cápsula es una versión simplificada del Observability Assessment que construirás en la cápsula 08. La progresión es:

Cápsula 05 (esta):
  Gap checklist rápido → descubres tus puntos ciegos

Cápsula 06 (mentalidad):
  Aprendes a pensar en "¿puedo responder con datos?"

Cápsula 07 (priorización):
  Decides qué instrumentar primero según tu tipo de sistema

Cápsula 08 (proyecto):
  Assessment formal → documento con gaps, plan, y timeline

El checklist que ejecutaste aquí te da un preview del proyecto. Cuando llegues a la cápsula 08, ya sabrás cuáles son tus gaps — el assessment los formaliza y los convierte en un plan de acción.

Los gaps como roadmap personal

Cada gap que identificaste mapea directamente a un módulo de esta guía:

Gap de costo        →  Módulo 2 (Métricas: tokens, USD, cost tracking)
Gap de calidad      →  Módulo 6 (AI Monitoring: quality, drift, hallucinations)
Gap de latencia     →  Módulo 2 (Métricas: TTFT, TTI) + Módulo 3 (Traces: per-span)
Gap de debugging    →  Módulo 3 (OpenTelemetry) + Módulo 7 (Production Debugging)
Gap de visibilidad  →  Módulo 4 (Dashboards) + Módulo 5 (Alerting)

No necesitas resolver todos los gaps en orden. Pero sí necesitas saber cuáles tienes antes de empezar a instrumentar.


Troubleshooting

Problema: "Mi sistema es pequeño, ¿realmente necesito observabilidad?"

Síntoma: Tu app tiene 50 requests al día. Sientes que instrumentar es overkill.

Solución: El volumen de requests no determina la necesidad de observabilidad — la variabilidad sí. 50 requests a un LLM pueden costar entre $0.05 y $50 dependiendo de los prompts. Sin tracking, no sabes en qué extremo estás. Empieza con lo mínimo: structlog + costo por request. Son 20 líneas y te dan visibilidad inmediata.

Problema: "Ya tengo un APM (Datadog/New Relic), ¿eso no cubre mis gaps?"

Síntoma: Tu APM muestra latencia, error rate, y throughput. Piensas que tienes observabilidad.

Solución: APMs genéricos miden la infraestructura, no la lógica AI. Tu APM sabe que un request tomó 3 segundos, pero no sabe que 2.5 de esos segundos fueron una llamada LLM que consumió 5,000 tokens y costó $0.02. Necesitas instrumentación AI-específica encima de tu APM, no en lugar de él.

Problema: "No quiero loguear prompts por privacidad"

Síntoma: Tus prompts contienen datos de usuarios y no puedes loguearlo todo.

Solución: No necesitas loguear el contenido del prompt para tener observabilidad. Loguea la metadata: longitud, tokens, hash del prompt, versión del system prompt, número de chunks de RAG. Para debugging, implementa un modo "debug session" donde capturas prompts completos para un usuario específico con su consentimiento, o en ambientes de staging.

Problema: "El gap analysis muestra 20+ gaps, no sé por dónde empezar"

Síntoma: Todo es rojo. La sensación es abrumadora.

Solución: No intentes cerrar todos los gaps simultáneamente. Prioriza por impacto:

  1. Primero costo — si no sabes cuánto gastas, cualquier otra decisión es a ciegas
  2. Segundo debugging — structlog + trace_ids te salvan horas en el próximo incidente
  3. Tercero latencia — desglose por paso te permite optimizar lo que más importa
  4. Cuarto calidad — requiere más infraestructura (evals, baselines) así que déjalo para cuando lo demás esté estable

Problema: "Implementé logging pero mis logs son demasiado ruidosos"

Síntoma: Después de agregar structlog, la terminal se llena de JSON y es difícil encontrar información.

Solución: Esto es normal al principio. La solución no es loguear menos — es loguear a un sistema que permita buscar y filtrar. En desarrollo, configura structlog para que solo muestre logs de nivel warning+ en consola. En producción, exporta logs a un agregador (Elasticsearch, Loki, CloudWatch) donde puedes hacer queries.


Ejercicios

Ejercicio 1: Encuentra los gaps en estos logs

Tu compañero te muestra estos logs de producción y dice que "ya tiene observabilidad". Identifica al menos 6 gaps:

2026-03-08 10:15:23 INFO Starting server on port 8000
2026-03-08 10:15:45 INFO Request received: POST /api/chat
2026-03-08 10:15:47 INFO Response sent: 200 OK
2026-03-08 10:16:01 INFO Request received: POST /api/chat
2026-03-08 10:16:05 INFO Response sent: 200 OK
2026-03-08 10:16:12 INFO Request received: POST /api/summarize
2026-03-08 10:16:18 INFO Response sent: 200 OK
2026-03-08 10:17:00 ERROR Request failed: POST /api/chat - TimeoutError
2026-03-08 10:17:15 INFO Request received: POST /api/chat
2026-03-08 10:17:17 INFO Response sent: 200 OK
Ver solución

Gaps identificados:

  1. Sin tokens: Ningún log registra prompt_tokens ni completion_tokens. Imposible saber el costo de cada request.

  2. Sin modelo: No se indica qué modelo se usó. Si hay fallback logic (gpt-4o → gpt-4o-mini), no se puede distinguir.

  3. Sin costo: No hay campo cost_usd. La pregunta "¿cuánto gastamos hoy?" no tiene respuesta.

  4. Sin latencia desglosada: Solo se puede inferir latencia end-to-end por diferencia de timestamps (2 segundos, 4 segundos, 6 segundos). No hay desglose LLM vs RAG vs validación.

  5. Sin trace_id: No hay forma de correlacionar el error de las 10:17:00 con datos específicos del request. ¿Qué prompt causó el timeout? Imposible saberlo.

  6. Sin user_id: No se sabe qué usuario generó cada request. Imposible analizar cost-per-user o determinar si un error afecta a un usuario o a muchos.

  7. Formato no estructurado: Los logs son texto plano. No se pueden buscar por campos, filtrar, ni agregar programáticamente. Un script de grep es tu única herramienta de análisis.

  8. El error no tiene contexto: "TimeoutError" en /api/chat — ¿fue timeout del LLM, de la red, del vector search? Sin spans ni contexto adicional, "timeout" es una descripción pero no un diagnóstico.

  9. Sin métricas de calidad: Los requests 200 OK pueden incluir hallucinations. Sin logging de quality checks o finish_reason, todo "200 OK" se ve igual aunque el output sea basura.

  10. Sin system_prompt_version: Si alguien cambió el prompt y la calidad degradó, no hay forma de correlacionar el cambio con el efecto.

La respuesta correcta a tu compañero: "Tienes logs de infraestructura. No tienes observabilidad AI."

Ejercicio 2: Calcula el costo oculto

Tu sistema tiene estos endpoints. Calcula el costo diario total y determina qué endpoint necesita optimización urgente:

/api/chat
  - Modelo: gpt-4o-mini
  - Requests/día: 2,000
  - Prompt tokens promedio: 300
  - Completion tokens promedio: 150

/api/summarize
  - Modelo: gpt-4o
  - Requests/día: 200
  - Prompt tokens promedio: 4,000
  - Completion tokens promedio: 400

/api/analyze
  - Modelo: gpt-4o
  - Requests/día: 50
  - Prompt tokens promedio: 8,000
  - Completion tokens promedio: 2,000

Precios (por 1K tokens):
  gpt-4o:      prompt=$0.0025, completion=$0.01
  gpt-4o-mini: prompt=$0.00015, completion=$0.0006
Ver solución
endpoints = {
    "/api/chat": {
        "model": "gpt-4o-mini",
        "requests_day": 2000,
        "avg_prompt_tokens": 300,
        "avg_completion_tokens": 150,
    },
    "/api/summarize": {
        "model": "gpt-4o",
        "requests_day": 200,
        "avg_prompt_tokens": 4000,
        "avg_completion_tokens": 400,
    },
    "/api/analyze": {
        "model": "gpt-4o",
        "requests_day": 50,
        "avg_prompt_tokens": 8000,
        "avg_completion_tokens": 2000,
    },
}

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

total_daily = 0

print("ANÁLISIS DE COSTO DIARIO")
print("=" * 60)

for ep, data in endpoints.items():
    p = prices[data["model"]]
    cost_per_request = (
        data["avg_prompt_tokens"] / 1000 * p["prompt"]
        + data["avg_completion_tokens"] / 1000 * p["completion"]
    )
    daily_cost = cost_per_request * data["requests_day"]
    total_daily += daily_cost

    print(f"\n{ep} ({data['model']})")
    print(f"  Costo/request:  ${cost_per_request:.6f}")
    print(f"  Requests/día:   {data['requests_day']}")
    print(f"  Costo/día:      ${daily_cost:.2f}")

print(f"\n{'─' * 60}")
print(f"TOTAL DIARIO:       ${total_daily:.2f}")
print(f"TOTAL MENSUAL:      ${total_daily * 30:.2f}")

Resultado:

/api/chat (gpt-4o-mini)
  Costo/request:  $0.000135
  Requests/día:   2,000
  Costo/día:      $0.27

/api/summarize (gpt-4o)
  Costo/request:  $0.014000
  Requests/día:   200
  Costo/día:      $2.80

/api/analyze (gpt-4o)
  Costo/request:  $0.040000
  Requests/día:   50
  Costo/día:      $2.00

──────────────────────────────────────────────────────────
TOTAL DIARIO:       $5.07
TOTAL MENSUAL:      $152.10

Insights:

  • /api/chat tiene 2,000 requests/día pero solo cuesta $0.27 (gpt-4o-mini es barato)
  • /api/summarize tiene 10x menos requests que /api/chat pero cuesta 10x más ($2.80 vs $0.27)
  • /api/analyze con solo 50 requests/día cuesta $2.00 — cada request cuesta $0.04
  • Optimización urgente: /api/summarize. Opciones: cambiar a gpt-4o-mini (si la calidad lo permite) reduciría su costo de $2.80 a ~$0.20. Reducir prompt_tokens (menos chunks de RAG) también ayudaría.

Sin este análisis, solo verías "gastamos $152/mes en OpenAI" sin saber que el 55% viene de un solo endpoint con 200 requests.

Ejercicio 3: Diseña el logging para un caso real

Tienes un sistema RAG que:

  1. Recibe una pregunta del usuario
  2. Busca en un vector store
  3. Selecciona los chunks más relevantes
  4. Construye un prompt con contexto
  5. Llama a un LLM
  6. Valida el output
  7. Responde al usuario

Diseña el esquema de logging: qué evento loguearías en cada paso y qué campos incluiría cada log. No escribas código — describe el esquema.

Ver solución
ESQUEMA DE LOGGING PARA SISTEMA RAG
════════════════════════════════════

PASO 1: Request recibido
  Event: "request_started"
  Campos: trace_id, user_id, endpoint, query_length, timestamp

PASO 2: Vector search
  Event: "vector_search_completed"
  Campos: trace_id, embedding_model, embedding_tokens, embedding_latency_ms,
          index_name, top_k, results_returned, min_score, max_score,
          search_latency_ms

PASO 3: Chunk selection
  Event: "chunks_selected"
  Campos: trace_id, chunks_in, chunks_out, selection_strategy,
          total_context_tokens, avg_relevance_score, discarded_reason

PASO 4: Prompt construction
  Event: "prompt_assembled"
  Campos: trace_id, system_prompt_version, context_tokens, user_query_tokens,
          total_prompt_tokens, includes_examples (bool), template_version

PASO 5: LLM call
  Event: "llm_call_completed"
  Campos: trace_id, model, temperature, prompt_tokens, completion_tokens,
          total_tokens, cost_usd, ttft_ms, llm_latency_ms,
          finish_reason, max_tokens_set

  Event (si error): "llm_call_failed"
  Campos: trace_id, model, error_type, error_message, retry_count,
          latency_ms

PASO 6: Output validation
  Event: "output_validated"
  Campos: trace_id, hallucination_check (pass/fail/skip),
          format_check (pass/fail), confidence_score,
          validation_latency_ms, validation_method

PASO 7: Response enviado
  Event: "request_completed"
  Campos: trace_id, user_id, endpoint, total_latency_ms,
          total_tokens, total_cost_usd, output_length,
          status (success/error/partial)

CAMPO COMPARTIDO en TODOS los logs: trace_id
  Esto permite buscar TODOS los logs de un request con un solo filtro.

LOGS DE NIVEL ERROR (adicionales, si aplica):
  - "rate_limit_hit": trace_id, model, retry_after_ms, attempt
  - "context_too_large": trace_id, context_tokens, max_allowed, action_taken
  - "validation_failed": trace_id, check_name, reason, output_preview_hash

Este esquema te da visibilidad completa sobre un request sin loguear contenido sensible (solo metadata). Si necesitas debugging profundo, activas el logging de contenido (prompts, outputs) para un trace_id específico.

Ejercicio 4: Simula un incidente y debuggéalo

Modifica la versión con observabilidad (ai_app_v2_with_observability.py) para simular un escenario donde el costo se dispara. Introduce un "bug" donde uno de los endpoints accidentalmente usa gpt-4o en lugar de gpt-4o-mini. Luego usa las métricas para detectar el problema.

Ver solución
"""
Simulación de incidente: model misrouting.
El endpoint /api/chat debería usar gpt-4o-mini pero alguien
lo configuró con gpt-4o.
"""

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(),
)

logger = structlog.get_logger()

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


def calculate_cost(model, prompt_tokens, completion_tokens):
    rates = COST_PER_1K.get(model, {"prompt": 0, "completion": 0})
    return (prompt_tokens / 1000 * rates["prompt"]) + (
        completion_tokens / 1000 * rates["completion"]
    )


requests_log = []


def simulate_request(endpoint, model, prompt_tokens, completion_tokens):
    trace_id = str(uuid.uuid4())[:8]
    cost = calculate_cost(model, prompt_tokens, completion_tokens)
    latency = prompt_tokens * 0.5 + completion_tokens * 2

    log_entry = {
        "trace_id": trace_id,
        "endpoint": endpoint,
        "model": model,
        "prompt_tokens": prompt_tokens,
        "completion_tokens": completion_tokens,
        "cost_usd": cost,
        "latency_ms": latency,
    }
    requests_log.append(log_entry)

    logger.info("request_completed", **log_entry)
    return log_entry


import random

random.seed(42)

print("=== ANTES DEL BUG (modelo correcto) ===\n")
for _ in range(10):
    pt = random.randint(200, 400)
    ct = random.randint(100, 200)
    simulate_request("/api/chat", "gpt-4o-mini", pt, ct)

normal_cost = sum(r["cost_usd"] for r in requests_log)
print(f"\nCosto de 10 requests normales: ${normal_cost:.6f}")

# --- BUG: alguien cambió el modelo a gpt-4o ---
print("\n\n=== DESPUÉS DEL BUG (modelo incorrecto) ===\n")
bug_start_idx = len(requests_log)

for _ in range(10):
    pt = random.randint(200, 400)
    ct = random.randint(100, 200)
    simulate_request("/api/chat", "gpt-4o", pt, ct)  # BUG: gpt-4o en vez de mini

bug_cost = sum(r["cost_usd"] for r in requests_log[bug_start_idx:])
print(f"\nCosto de 10 requests con bug: ${bug_cost:.6f}")

# --- DETECCIÓN ---
print("\n\n=== DETECCIÓN DEL INCIDENTE ===\n")

cost_by_model = defaultdict(float)
requests_by_model = defaultdict(int)
for r in requests_log:
    if r["endpoint"] == "/api/chat":
        cost_by_model[r["model"]] += r["cost_usd"]
        requests_by_model[r["model"]] += 1

print("Costo de /api/chat por modelo:")
for model, cost in cost_by_model.items():
    count = requests_by_model[model]
    avg = cost / count if count > 0 else 0
    print(f"  {model}: ${cost:.6f} total ({count} requests, ${avg:.6f}/req)")

if "gpt-4o" in cost_by_model and "gpt-4o-mini" in cost_by_model:
    ratio = (
        cost_by_model["gpt-4o"]
        / requests_by_model["gpt-4o"]
        / (cost_by_model["gpt-4o-mini"] / requests_by_model["gpt-4o-mini"])
    )
    print(f"\n⚠️  ALERTA: gpt-4o es {ratio:.0f}x más caro que gpt-4o-mini por request")
    print("    en el endpoint /api/chat")
    print("    ACCIÓN: Verificar configuración de modelo para /api/chat")
    print(
        f"    AHORRO ESTIMADO si se corrige: "
        f"${bug_cost - normal_cost:.6f} por cada 10 requests"
    )

Output:

=== DETECCIÓN DEL INCIDENTE ===

Costo de /api/chat por modelo:
  gpt-4o-mini: $0.001319 total (10 requests, $0.000132/req)
  gpt-4o: $0.018450 total (10 requests, $0.001845/req)

⚠️  ALERTA: gpt-4o es 14x más caro que gpt-4o-mini por request
    en el endpoint /api/chat
    ACCIÓN: Verificar configuración de modelo para /api/chat
    AHORRO ESTIMADO si se corrige: $0.017131 por cada 10 requests

Sin el campo model en los logs, este incidente sería invisible. El endpoint seguiría funcionando (200 OK), las respuestas serían correctas (gpt-4o es más capaz), pero estarías pagando 14x más de lo necesario. A 2,000 requests/día, eso es ~$3.70/día de desperdicio → $111/mes.

Ejercicio 5: Crea tu propio gap report

Ejecuta el Gap Analysis Checklist (el script gap_analysis_checklist.py) contra tu sistema AI real. Si no tienes uno, usa un sistema hipotético basado en tu experiencia. Genera un reporte con:

  1. Gaps encontrados por categoría
  2. Los 3 gaps más críticos y por qué
  3. Plan de acción para cerrar esos 3 gaps (qué implementar, estimación de esfuerzo)
Ver solución (ejemplo para un chatbot FastAPI + OpenAI sin instrumentación)
GAP REPORT — Chatbot de Soporte (FastAPI + OpenAI)
═══════════════════════════════════════════════════

FECHA: 2026-03-08
SISTEMA: Chatbot FAQ con RAG, FastAPI, gpt-4o-mini, Pinecone

RESULTADOS DEL CHECKLIST:
  COSTO:     5/6 gaps  🔴 CRÍTICO
  CALIDAD:   6/6 gaps  🔴 CRÍTICO
  LATENCIA:  4/6 gaps  🔴 CRÍTICO
  DEBUGGING: 5/6 gaps  🔴 CRÍTICO

COBERTURA: 16.7% (4/24)

LOS 3 GAPS MÁS CRÍTICOS:

1. NO SÉ CUÁNTO CUESTA CADA REQUEST
   Impacto: La factura mensual de OpenAI subió 40% el mes pasado.
   No sé por qué. No puedo identificar qué endpoint o usuario
   causó el aumento. Estoy tomando decisiones de pricing de
   producto sin saber mis costos reales.

   Plan de acción:
   - Implementar structlog con campos de tokens y cost_usd (2h)
   - Agregar calculate_cost con tabla de precios (30min)
   - Agregar endpoint y user_id a cada log (1h)
   - Esfuerzo total: ~4 horas

2. NO PUEDO DEBUGGEAR UN REPORTE DE USUARIO
   Impacto: La semana pasada un usuario reportó una respuesta
   incorrecta. No pude encontrar qué prompt se envió ni qué
   contexto RAG recibió el modelo. Cerré el ticket como
   "no reproducible". El problema probablemente sigue.

   Plan de acción:
   - Agregar trace_id a todos los logs (1h)
   - Loguear metadata de RAG: chunks seleccionados, scores (2h)
   - Loguear system_prompt_version (30min)
   - Implementar safe_log_content para prompts en staging (1h)
   - Esfuerzo total: ~5 horas

3. NO DETECTO DEGRADACIÓN DE CALIDAD
   Impacto: El mes pasado, OpenAI actualizó gpt-4o-mini. No sé
   si mis outputs mejoraron o empeoraron. No tengo baseline ni
   tracking de calidad. Podría estar perdiendo usuarios sin
   saberlo.

   Plan de acción:
   - Definir quality metrics básicas (relevance, format) (2h)
   - Implementar LLM-as-judge para sampling de outputs (4h)
   - Crear baseline con datos actuales (2h)
   - Esfuerzo total: ~8 horas (más complejo, priorizar después de 1 y 2)

TIMELINE PROPUESTO:
  Semana 1: Gap 1 (costo) — structlog + cost tracking
  Semana 2: Gap 2 (debugging) — trace_ids + RAG metadata
  Semana 3-4: Gap 3 (calidad) — quality metrics + baseline

Este tipo de reporte es lo que producirás formalmente en el proyecto del módulo (cápsula 08), pero más detallado y con evidencia de datos.


Resumen

  • Los gaps de observabilidad en AI se organizan en cuatro categorías: costo, calidad, latencia, y debugging. Cada categoría representa preguntas de producción que no puedes responder sin instrumentación.
  • Gaps de costo son los más peligrosos financieramente: prompt creep, retry storms, whale users, y model misrouting pueden multiplicar tu factura sin que lo notes.
  • Gaps de calidad son los más silenciosos: hallucinations, model drift, y degradación gradual no generan errores HTTP — solo usuarios insatisfechos que se van sin reportar.
  • Gaps de latencia con medición end-to-end única te dicen "está lento" sin decirte dónde. El desglose por paso (embedding, RAG, LLM, validación) es lo que convierte un síntoma en un diagnóstico.
  • Gaps de debugging hacen que cada incidente sea una aventura: sin trace_ids, logs estructurados, y metadata AI, "investigar" significa "adivinar".
  • La diferencia entre zero y basic observability es ~50-100 líneas de código. El salto de "nada" a "algo" es el de mayor impacto.
  • El Gap Analysis Checklist te da un diagnóstico rápido de 24 preguntas. Cada "No" es un gap concreto que mapea a un módulo específico de esta guía.
  • No intentes cerrar todos los gaps a la vez. Prioriza: costo primero (impacto financiero directo), debugging segundo (te salva en incidentes), latencia tercero (mejora UX), calidad cuarto (requiere más infraestructura).

Recursos Adicionales

  1. Charity Majors — Observability is About Confidence — Por qué observabilidad no es herramientas sino confianza operativa
  2. OpenAI API Pricing — Precios actualizados para calcular gaps de costo
  3. Anthropic Token Counting — Cómo contar tokens para tracking de costos
  4. structlog — Structured Logging for Python — La librería base para cerrar gaps de logging
  5. Google SRE — Practical Alerting — Cómo pasar de "no sé qué pasa" a alertas accionables
  6. Hamel Husain — Your AI Product Needs Evals — Por qué los gaps de calidad son los más costosos en producto
  7. OpenTelemetry — Getting Started — La herramienta que usarás en el módulo 3 para cerrar gaps de tracing
  8. LangSmith — Tracing and Debugging — Herramienta específica para debugging de LLM flows