Observabilidad y trazas

Al terminar esta lección cada petición a tu agente dejará una traza completa: qué preguntó el usuario, qué llamadas al modelo hubo, qué herramientas se ejecutaron, cuánto tardó y cuánto costó cada paso. Sabrás qué métricas sacar de ahí y qué alertas poner.

La idea

La Observabilidad es la capacidad de responder preguntas sobre tu sistema en producción sin tener que desplegar código nuevo para averiguarlo. En un agente las preguntas típicas son: ¿por qué respondió esto?, ¿por qué tardó veinte segundos?, ¿por qué esta conversación costó diez veces más que la media?, ¿desde cuándo falla esta herramienta?

Para responderlas necesitas trazas: un registro estructurado de cada petición como un árbol de tramos (spans). La raíz es la petición del usuario. Debajo cuelgan las llamadas al modelo y las herramientas. En un sistema multiagente, cada subagente es una rama con sus propios tramos.

flowchart TD
  P[peticion trace_id=t1] --> M1[modelo paso 1]
  P --> H1[herramienta buscar_pedido]
  P --> M2[modelo paso 2]
  P --> S[subagente]
  S --> M3[modelo sub paso 1]
  S --> H2[herramienta buscar_docs]

Todos los tramos comparten trace_id. Cada uno tiene su span_id y el parent_id de quien lo creó. Con esos tres campos reconstruyes el árbol en cualquier herramienta de análisis.

Qué registrar

Tres campos se olvidan a menudo y luego se echan de menos:

• stop_reason. Un aumento de max_tokens indica respuestas cortadas; uno de refusal, peticiones que el modelo rechaza. Ninguno lanza una excepción, así que solo los ves si los registras.
• El ID de la petición del proveedor. Si tienes que abrir una incidencia con el proveedor, es lo que te pedirán. El SDK de Python lo expone en la respuesta como _request_id.
• La versión del prompt. Sin ella no puedes relacionar un cambio de comportamiento con un despliegue (Despliegue).

Trazas con contextvars

No necesitas una plataforma para empezar. Una línea JSON por tramo, escrita en la salida estándar o en un archivo, se puede enviar después a cualquier sistema de registros. contextvars propaga el tramo actual a través de funciones y de tareas de asyncio sin pasarlo como argumento.

import contextvars, json, sys, time, uuid
from contextlib import contextmanager

_traza = contextvars.ContextVar("traza", default=None)
_tramo = contextvars.ContextVar("tramo", default=None)

def emitir(registro: dict) -> None:
    sys.stdout.write(json.dumps(registro, ensure_ascii=False, default=str) + "\n")

@contextmanager
def tramo(tipo: str, **atributos):
    trace_id = _traza.get() or uuid.uuid4().hex
    span_id, parent_id = uuid.uuid4().hex[:16], _tramo.get()
    t_traza, t_tramo = _traza.set(trace_id), _tramo.set(span_id)
    datos = {"tipo": tipo, **atributos}
    inicio = time.perf_counter()
    try:
        yield datos            # quien usa el tramo añade campos a este diccionario
    except Exception as e:
        datos["error"] = repr(e)
        raise
    finally:
        emitir({"trace_id": trace_id, "span_id": span_id, "parent_id": parent_id,
                "ms": round((time.perf_counter() - inicio) * 1000), **datos})
        _tramo.reset(t_tramo); _traza.reset(t_traza)

Y así se usa dentro del bucle del agente:

import anthropic
from trazas import tramo

MODEL = "claude-sonnet-5-5"
PROMPT_VERSION = "soporte-v14"
client = anthropic.Anthropic()

def llamar_modelo(paso: int, **kwargs):
    with tramo("modelo", modelo=MODEL, paso=paso) as t:
        resp = client.messages.create(model=MODEL, **kwargs)
        u = resp.usage
        t.update(stop_reason=resp.stop_reason, message_id=resp.id,
                 request_id=getattr(resp, "_request_id", None),
                 input_tokens=u.input_tokens, output_tokens=u.output_tokens,
                 cache_creation=u.cache_creation_input_tokens or 0,
                 cache_read=u.cache_read_input_tokens or 0)
        return resp

def llamar_herramienta(nombre: str, entrada: dict, ejecutar) -> tuple[str, bool]:
    with tramo("herramienta", nombre=nombre, entrada=redactar(entrada)) as t:
        try:
            salida, error = ejecutar(nombre, entrada), False
        except Exception as e:
            salida, error = f"Error: {e}", True
        t.update(is_error=error, bytes_resultado=len(salida))
        return salida, error

def atender(usuario_id: str, pregunta: str) -> str:
    with tramo("peticion", usuario=seudonimo(usuario_id), prompt=PROMPT_VERSION) as t:
        respuesta = bucle_agente(pregunta)  # usa llamar_modelo y llamar_herramienta
        t["resultado"] = "ok"
        return respuesta

redactar y seudonimo son tuyas: la primera quita o enmascara datos personales de los argumentos; la segunda convierte el ID del usuario en un hash con sal para poder agrupar sin identificar. Con subagentes en paralelo, cada tarea de asyncio hereda una copia del contexto, así que sus tramos cuelgan del tramo que las lanzó sin hacer nada más.

# salida de ejemplo
{"trace_id": "9f2c…", "span_id": "a1b2…", "parent_id": "77de…", "ms": 2140, "tipo": "modelo", "modelo": "claude-sonnet-5-5", "paso": 1, "stop_reason": "tool_use", "input_tokens": 3120, "output_tokens": 88, "cache_creation": 0, "cache_read": 2900}
{"trace_id": "9f2c…", "span_id": "c3d4…", "parent_id": "77de…", "ms": 180, "tipo": "herramienta", "nombre": "buscar_pedido", "entrada": {"pedido_id": "4512"}, "is_error": false, "bytes_resultado": 412}

Cuando el volumen crezca, OpenTelemetry es el estándar abierto para trazas y la mayoría de plataformas de observabilidad lo aceptan. El modelo de datos es el mismo: tramos con IDs y atributos.

Métricas que salen de las trazas

Agregando los tramos obtienes lo necesario para operar:

• Coste por petición (media y percentil alto), por versión de prompt y por tipo de petición. Sale de usage y de tus precios (Control de costes y prompt caching).
• Latencia total y del primer token, percentiles 50 y 95. Los usuarios recuerdan las esperas largas, y la media las esconde (Latencia: streaming, modelo adecuado y menos pasos).
• Pasos por petición y su distribución. Una cola larga indica bucles que no convergen.
• Errores de herramienta por nombre. Un salto repentino suele ser un cambio en un sistema externo.
• Distribución de stop_reason, con atención a max_tokens y refusal.
• Tasa de caché: cache_read frente al total de entrada.

Pon alertas sobre pocas cosas y bien elegidas: coste por hora por encima de un presupuesto, tasa de errores de una herramienta, percentil 95 de latencia y aumento de max_tokens. Cada alerta debe llevar a una acción concreta; si nadie sabe qué hacer cuando salta, sobra.

De la traza a la eval

Las trazas cierran el círculo con Nivel 80 · Evals. Cuando un usuario se queja o una alerta salta, buscas la traza, entiendes el fallo y la conviertes en un caso de tu conjunto de prueba. Si tu formato de trazas coincide con el de la trayectoria de tus evals (Evals de trayectoria: cómo llegó, no solo qué dijo), el paso es casi automático.

Errores frecuentes

• Registrar solo errores y no poder explicar una respuesta que no falló pero estaba mal.
• No registrar usage y estimar el coste con la factura mensual.
• Usar IDs distintos en cada servicio y no poder unir los tramos de una misma petición.
• Guardar prompts completos con datos personales sin redacción ni retención.
• Mirar solo medias de latencia y coste.

Ejercicio

Resumen

• Una traza es un árbol de tramos con trace_id, span_id y parent_id.
• Registra por llamada al modelo stop_reason, usage, duración e IDs; por herramienta, nombre, argumentos redactados, error y duración.
• contextvars propaga la traza por funciones y tareas de asyncio sin tocar firmas.
• Saca de las trazas coste, latencia, pasos, errores y stop_reason, y alerta solo sobre lo accionable.
• Redacta datos personales y fija una retención.

Para seguir

• Control de costes y prompt caching
• Observabilidad
• Cómo leer la traza de un agente