Observabilidad: ver qué hace tu IA por dentro

Definición

La observabilidad (observability) es la capacidad de entender qué pasó dentro de un sistema a partir de los datos que emite mientras funciona. En una app de IA significa poder responder, para cualquier pregunta de un usuario, qué entró, qué prompt recibió el modelo, qué documentos se recuperaron, qué contestó, cuánto tardó, cuánto costó y en qué paso falló.

Se apoya en tres tipos de datos o señales: registros (logs, eventos con fecha y hora), métricas (metrics, números agregados como latencia o errores por minuto) y rastros (traces). Un rastro es el recorrido completo de una petición; cada paso dentro de él (buscar documentos, llamar al modelo, usar una herramienta) es un tramo (span) con su hora de inicio, duración, atributos y estado. Los tramos se anidan: el de la pregunta contiene al de la búsqueda y al de la generación. Un modelo de lenguaje no da siempre la misma respuesta (día 9) y una falla puede venir de la búsqueda, del prompt, del modelo o de una herramienta. Sin rastros solo ves la respuesta final; con ellos ves el paso exacto donde se torció.

Qué se registra en una app de LLM

OpenTelemetry y sus convenciones para IA generativa

OpenTelemetry (OTel) es el estándar abierto de la Cloud Native Computing Foundation para generar y exportar rastros, métricas y registros sin atarse a un proveedor. Sus convenciones semánticas para IA generativa (GenAI semantic conventions) fijan los nombres de los atributos para que cualquier herramienta entienda los rastros de cualquier app:

Las convenciones están en estado de desarrollo (Development) y pueden cambiar; capturar el contenido de los mensajes es opt-in por privacidad. Instrumentar con OTel permite cambiar de herramienta sin reescribir el código.

Herramientas

Todas reciben rastros por SDK u OpenTelemetry. Se eligen por dónde deben vivir los datos (nube o servidor propio), integración con tu stack, costo y qué tan fuerte es su parte de evaluación.

Evaluación: de los rastros a las pruebas

Una evaluación (eval) es una prueba automática de calidad para una app de IA: un conjunto de casos (entradas con el resultado esperado o un criterio), una forma de calificar cada respuesta y un reporte comparable entre versiones.

El ciclo que une observabilidad y evaluación:

1. Los rastros de producción revelan respuestas malas (retroalimentación negativa, errores, puntajes bajos).
2. Cada caso malo se convierte en un caso de prueba con su resultado esperado.
3. Antes de cambiar un prompt, un modelo o la búsqueda, se corre el conjunto completo; si algo que antes pasaba ahora falla, es una regresión (regression) y el cambio no se publica.

Caso real: Notion. Según el caso publicado por Braintrust el 9 de octubre de 2024, Notion dejó atrás un flujo con archivos JSONL y calificadores humanos y pasó a conjuntos de datos curados con ejemplos de uso real registrados automáticamente, calificados con reglas y LLM como juez. Con ese flujo, el equipo pasó de clasificar y corregir unas 3 fallas al día a unas 30.

Privacidad: datos personales en los rastros

Un rastro completo guarda lo que el usuario escribió y lo que los documentos dicen: nombres, teléfonos, contratos, datos de salud. Buenas prácticas:

• Enmascarar (masking) antes de guardar: sustituir correos, teléfonos, RFC o números de tarjeta por marcadores ([CORREO]), también si los rastros van a un servicio externo.
• Guardar el contenido completo solo donde haga falta, con retención limitada y acceso por roles.
• Respetar el aviso de privacidad y la ley aplicable (en México, la Ley Federal de Protección de Datos Personales en Posesión de los Particulares).

Cuándo empezar

Práctica: rastreador mínimo y mini eval

El script implementa un rastreador con un context manager de Python: cada with span(...) registra un tramo con su rastro, su padre, la duración, los tokens estimados y el costo, y lo escribe en un archivo JSONL (un JSON por línea). Lo aplica a un flujo RAG simulado (sin llamar a ninguna API) y corre un mini eval de 5 casos que dice cuál falló y en qué paso. Los precios son de ejemplo y los tokens se estiman con la regla aproximada de 4 caracteres por token.

1. Guarda el script como mini_rastreo.py:

"""Rastreo mínimo (spans en JSONL) y mini eval sobre un RAG simulado. Solo biblioteca estándar; precios de ejemplo."""
import json
import time
import uuid
from contextlib import contextmanager
from contextvars import ContextVar

ARCHIVO = "rastros.jsonl"
PRECIO_ENTRADA, PRECIO_SALIDA = 2.00, 10.00   # USD por millón de tokens (ejemplo)
_padre: ContextVar[str | None] = ContextVar("padre", default=None)
_traza: ContextVar[str | None] = ContextVar("traza", default=None)

def tokens(texto: str) -> int:
    return max(1, len(texto) // 4)   # estimación burda: ~4 caracteres por token

@contextmanager
def span(nombre: str, **atributos):
    """Registra un paso: a qué rastro pertenece, quién es su padre, cuánto tardó y si falló."""
    raiz = _traza.get() is None
    if raiz:
        _traza.set(uuid.uuid4().hex[:8])
    datos = {"traza": _traza.get(), "span": uuid.uuid4().hex[:6], "padre": _padre.get(), "nombre": nombre, **atributos}
    marca = _padre.set(datos["span"])
    inicio = time.perf_counter()
    try:
        yield datos
        datos["estado"] = "ok"
    except Exception as e:
        datos["estado"], datos["error"] = "error", f"{type(e).__name__}: {e}"
        raise
    finally:
        datos["ms"] = round((time.perf_counter() - inicio) * 1000, 1)
        _padre.reset(marca)
        with open(ARCHIVO, "a", encoding="utf-8") as f:
            f.write(json.dumps(datos, ensure_ascii=False) + "\n")
        if raiz:
            _traza.set(None)

DOCUMENTOS = {
    "garantia.txt": "La garantía de los equipos es de 12 meses a partir de la fecha de compra.",
    "envios.txt": "Los envíos nacionales tardan de 3 a 5 días hábiles.",
    "devoluciones.txt": "Se aceptan devoluciones dentro de los 30 días naturales con ticket.",
}

def recuperar(pregunta: str) -> list[str]:
    with span("recuperar", entrada=pregunta) as s:
        time.sleep(0.01)
        clave = {w.strip("¿?.,").lower() for w in pregunta.split() if len(w) > 4}
        s["documentos"] = [d for d, t in DOCUMENTOS.items() if clave & {w.strip(".,").lower() for w in t.split()}]
        return s["documentos"]

def generar(pregunta: str, docs: list[str]) -> str:
    prompt = "Responde solo con las fuentes.\n" + "\n".join(DOCUMENTOS[d] for d in docs) + "\nPregunta: " + pregunta
    with span("generar", modelo="modelo-ejemplo") as s:
        time.sleep(0.02)   # simula la llamada al LLM
        respuesta = DOCUMENTOS[docs[0]] if docs else "No aparece en los documentos."
        s["tokens_entrada"], s["tokens_salida"] = tokens(prompt), tokens(respuesta)
        s["costo_usd"] = round((s["tokens_entrada"] * PRECIO_ENTRADA + s["tokens_salida"] * PRECIO_SALIDA) / 1e6, 6)
        return respuesta

def responder(pregunta: str) -> dict:
    with span("pregunta", entrada=pregunta) as raiz:
        docs = recuperar(pregunta)
        raiz["salida"] = generar(pregunta, docs)
        return {"traza": raiz["traza"], "docs": docs, "respuesta": raiz["salida"]}

CASOS = [  # (pregunta, documento esperado, texto esperado en la respuesta)
    ("¿Cuántos meses dura la garantía?", "garantia.txt", "12 meses"),
    ("¿Cuánto tardan los envíos nacionales?", "envios.txt", "3 a 5"),
    ("¿Puedo hacer devoluciones sin ticket?", "devoluciones.txt", "30 días"),
    ("¿En cuánto tiempo llega mi paquete?", "envios.txt", "3 a 5"),
    ("¿Tienen tienda en Monterrey?", None, "No aparece"),
]

if __name__ == "__main__":
    open(ARCHIVO, "w").close()
    aprobados = 0
    for pregunta, doc, texto in CASOS:
        r = responder(pregunta)
        if doc and doc not in r["docs"]:
            veredicto = f"FALLA en recuperar: esperaba {doc}, llegó {r['docs'] or 'nada'}"
        elif texto not in r["respuesta"]:
            veredicto = f"FALLA en generar: la respuesta no contiene «{texto}»"
        else:
            veredicto, aprobados = "ok", aprobados + 1
        print(f"[{r['traza']}] {pregunta:<40} {veredicto}")
    filas = [json.loads(linea) for linea in open(ARCHIVO, encoding="utf-8")]
    gen = [f for f in filas if f["nombre"] == "generar"]
    print(f"\nAprobados: {aprobados}/{len(CASOS)} · {len(filas)} spans en {ARCHIVO}")
    print(f"Tokens: {sum(f['tokens_entrada'] for f in gen)} de entrada, {sum(f['tokens_salida'] for f in gen)} de salida"
          f" · costo (precios de ejemplo): ${sum(f['costo_usd'] for f in gen):.6f} USD")
    fallo = next(f["traza"] for f in filas if f["nombre"] == "recuperar" and not f["documentos"])
    print(f"\nRastro {fallo} (el caso que falló), span por span:")
    for f in (f for f in filas if f["traza"] == fallo):
        print(json.dumps(f, ensure_ascii=False))

2. Ejecútalo: python mini_rastreo.py. Salida real (Python 3.12, Windows); los identificadores y los milisegundos cambian en cada ejecución:

[4b9b917d] ¿Cuántos meses dura la garantía?         ok
[c64d845a] ¿Cuánto tardan los envíos nacionales?    ok
[16f4ce1a] ¿Puedo hacer devoluciones sin ticket?    ok
[cf16347a] ¿En cuánto tiempo llega mi paquete?      FALLA en recuperar: esperaba envios.txt, llegó nada
[ad17a7c7] ¿Tienen tienda en Monterrey?             ok

Aprobados: 4/5 · 15 spans en rastros.jsonl
Tokens: 140 de entrada, 60 de salida · costo (precios de ejemplo): $0.000880 USD

Rastro cf16347a (el caso que falló), span por span:
{"traza": "cf16347a", "span": "014783", "padre": "be978f", "nombre": "recuperar", "entrada": "¿En cuánto tiempo llega mi paquete?", "documentos": [], "estado": "ok", "ms": 10.5}
{"traza": "cf16347a", "span": "2345ad", "padre": "be978f", "nombre": "generar", "modelo": "modelo-ejemplo", "tokens_entrada": 19, "tokens_salida": 7, "costo_usd": 0.000108, "estado": "ok", "ms": 20.5}
{"traza": "cf16347a", "span": "be978f", "padre": null, "nombre": "pregunta", "entrada": "¿En cuánto tiempo llega mi paquete?", "salida": "No aparece en los documentos.", "estado": "ok", "ms": 32.3}

3. Interpreta el resultado. Cuatro casos pasan. El cuarto falla y el eval dice dónde: la búsqueda no trajo envios.txt porque «¿En cuánto tiempo llega mi paquete?» no comparte palabras con «Los envíos nacionales tardan…» (el límite de la búsqueda léxica del día 11). El rastro lo confirma: el tramo recuperar devolvió "documentos": [] y el modelo, correctamente, contestó que no aparece. El problema no es el modelo ni el prompt: es la búsqueda.

4. Convierte la falla en mejora. Ese caso ya es parte del conjunto de pruebas. Si mejoras la búsqueda (sinónimos, embeddings o búsqueda híbrida), vuelve a correr el script: el cambio se acepta solo si pasa 5/5 sin romper los otros cuatro. Para producción, sustituye span() por el SDK de OpenTelemetry o de tu herramienta de rastreo, con los atributos gen_ai.*.

Preguntas frecuentes

¿Qué es la observabilidad en IA en palabras simples?

Es registrar cada paso que da una app de IA (qué llegó, qué buscó, qué respondió, cuánto tardó y cuánto costó) para poder explicar después cualquier respuesta, buena o mala.

¿Qué diferencia hay entre un log y un trace?

Un log es un evento suelto. Un trace (rastro) une todos los pasos de una misma petición, con su orden, su duración y quién llamó a quién.

¿Qué es un eval?

Es una prueba automática de calidad: un conjunto de casos con su resultado esperado que se corre antes de cambiar un prompt o un modelo, para detectar si algo empeoró.

En el reel lo explicamos así

Una IA le dio una respuesta absurda a un cliente. ¿Cómo averiguas por qué? La observabilidad es como la caja negra de un avión: no evita el problema, pero explica qué pasó. Cada pregunta deja un rastro (trace): qué llegó, qué buscó, qué respondió, cuánto tardó y cuánto costó; con él ves el paso exacto que falló. Notion convierte los rastros reales en pruebas y pasó de corregir 3 fallas al día a 30. Antes de cambiar una instrucción o un modelo, se corren esas pruebas: una evaluación.

Referencias

• OpenTelemetry, Observability primer y Traces: señales, rastros y tramos.
• OpenTelemetry, GenAI Semantic Conventions: atributos gen_ai.* para spans, métricas y eventos.
• Braintrust, «How Notion develops world-class AI features» (9 de octubre de 2024, copia archivada): el caso de 3 a 30 fallas al día. La versión actual del caso está en braintrust.dev/customers/notion.
• Langfuse, LangSmith, Braintrust y Arize Phoenix: documentación oficial.
• Anthropic, Define success criteria and build evaluations: cómo diseñar evaluaciones.
• OWASP, Top 10 for LLM Applications: LLM02 Sensitive Information Disclosure.