Logs estructurados como JSON

Descripción

La lección anterior confirmó, ejecutado, que print() produce texto libre: legible para un humano, ilegible para un programa. Esta lección construye la solución concreta: cada evento de log deja de ser una frase suelta y se convierte en un objeto JSON completo, con campos fijos y nombrados, en una sola línea. Un archivo con una línea de JSON por evento —el formato conocido como JSON Lines o NDJSON— se puede leer con un editor de texto como cualquier archivo, pero también se puede procesar con json.loads línea por línea, filtrar por campo, agregar, o cargar en cualquier herramienta que entienda JSON. Esa doble naturaleza —legible y parseable a la vez— es la base de todo lo que este módulo construye a partir de aquí.

Esta lección no toca todavía run_reservo_agent ni ninguna tool de Reservo — eso empieza en la lección 05. Aquí se construye, y se prueba, la maquinaria del logging estructurado en sí misma: un formatter propio, dos dataclasses de eventos, y la función que las convierte en una línea de log real.

Conexión con el módulo

Esta lección entrega la primera pieza real de observability/run_logger.py: el formatter JSON y la función log_event, que todas las lecciones que siguen —04 en adelante— van a reusar sin cambios. RunEvent, la primera dataclass de eventos, se completa aquí; ToolCallEvent, la segunda, llega en la lección 05, cuando hay pasos de loop que registrar.

Analogía: una hoja de admisión, contra una nota al margen

Cuando alguien llega a la sala de urgencias de un hospital, no se anota lo que pasó en una libreta con prosa libre ("llegó un señor con dolor, parecía grave"). Se llena una hoja de admisión: campos fijos —nombre, hora de llegada, síntoma principal, nivel de urgencia, signos vitales—, cada uno en su casilla, sin importar quién la llene ni en qué turno. Esa estructura fija es lo que permite que, más tarde, cualquiera —un médico distinto, un sistema de estadísticas del hospital, una auditoría— pueda tomar cien hojas de admisión y contar cuántos casos fueron urgentes, sin tener que leer cien párrafos de prosa y decidir, caso por caso, qué querían decir.

Una línea de log estructurado es esa hoja de admisión. {"trace_id": "run-abc123", "event": "tool_use", "tool": "get_quote", ...} no es más "informativo" que la línea equivalente en texto libre de la lección anterior —de hecho, dice exactamente lo mismo—. Lo que cambia es que cualquier programa, no solo un humano leyendo con atención, puede extraer tool sin ambigüedad, sin necesidad de adivinar dónde empieza y dónde termina cada dato dentro de una frase.

Ejemplo trabajado: el formatter, y las primeras dos líneas de JSON reales

El logger, configurado para escribir JSON

import json
import logging
import sys
from dataclasses import dataclass, asdict

logger = logging.getLogger("reservo.observability")
logger.setLevel(logging.DEBUG)
_handler = logging.StreamHandler(sys.stdout)
_handler.setFormatter(logging.Formatter("%(message)s"))  # el message YA es la línea JSON completa
logger.handlers = [_handler]
logger.propagate = False

Fíjate en la decisión de diseño: en vez de escribir un logging.Formatter complejo que arme el JSON leyendo atributos del LogRecord (el objeto interno que logging construye por cada llamada), esta guía arma el JSON antes de llamar al logger, con json.dumps directo, y le pasa esa línea ya completa como el message. El formatter, entonces, se vuelve trivial —"%(message)s", sin ningún campo adicional—, porque el mensaje ya es exactamente lo que se quiere escribir. Esta es la opción "json.dumps directo" que el diseño de esta guía deja abierta junto a la opción de un formatter propio más elaborado — más simple de leer, más fácil de probar, y con el mismo resultado final.

logger.propagate = False evita que este logger reenvíe sus mensajes al logger raíz de Python (que podría tener su propia configuración, y duplicaría cada línea). Es una línea que casi nunca se nota hasta que falta, y entonces cada evento aparece dos veces.

RunEvent, la primera dataclass de eventos

@dataclass
class RunEvent:
    """Un evento de nivel de RUN: abre o cierra un run completo."""
    seq: int
    trace_id: str
    event: str            # "run_started" | "run_finished" | "run_failed"
    question: str = ""
    error: str = ""


def log_event(level, event):
    """Serializa CUALQUIER evento (un dataclass) como una sola línea JSON,
    y la emite con el nivel dado. Una línea == un objeto JSON completo: el
    formato NDJSON/JSON Lines, parseable con json.loads línea por línea."""
    logger.log(level, json.dumps(asdict(event), ensure_ascii=False))

seq es el reemplazo honesto del reloj real que ya adelantó la lección 01 de este módulo: un entero que crece de a uno por cada evento generado, para poder ordenar eventos sin depender de datetime.now(). Todavía no está conectado a nada en esta lección —llega con un contador real en la lección 04—; aquí se pasa a mano para confirmar que el formato funciona.

asdict(event) convierte la dataclass en un diccionario normal de Python — {"seq": 1, "trace_id": "...", "event": "...", ...} —, y json.dumps(..., ensure_ascii=False) lo convierte en la línea de texto que logger.log termina emitiendo. El parámetro ensure_ascii=False importa: sin él, cualquier acento en la prosa ("Sofía", por ejemplo) sale escapado como í, válido pero mucho más difícil de leer a simple vista.

Las dos primeras líneas reales

log_event(logging.INFO, RunEvent(seq=1, trace_id="run-demo", event="run_started", question="Reserva Focus pro 3h para Ana"))
log_event(logging.INFO, RunEvent(seq=2, trace_id="run-demo", event="run_finished", question="Reserva Focus pro 3h para Ana"))

Qué esperar:

{"seq": 1, "trace_id": "run-demo", "event": "run_started", "question": "Reserva Focus pro 3h para Ana", "error": ""}
{"seq": 2, "trace_id": "run-demo", "event": "run_finished", "question": "Reserva Focus pro 3h para Ana", "error": ""}

Dos líneas, cada una un objeto JSON completo, autocontenido — no necesitas la línea anterior ni la siguiente para entender lo que dice una de ellas. Confirma que son JSON válido y parseable, de vuelta a un diccionario de Python, con la misma librería estándar que las escribió:

parsed = json.loads('{"seq": 1, "trace_id": "run-demo", "event": "run_started", "question": "Reserva Focus pro 3h para Ana", "error": ""}')
print(parsed["trace_id"], "->", parsed["event"])

run-demo -> run_started

Esta es, con precisión, la diferencia con la lección 02: aquella línea de print() —llamando list_rooms con {}— también es texto, pero ningún json.loads la puede parsear como un objeto con campos. Esta línea sí.

Por qué esto no es solo print(json.dumps(...))

Vale la pena notar, con precisión, qué agrega logging por encima de simplemente imprimir JSON a mano — porque en este ejemplo, la línea que sale a pantalla es, byte por byte, la misma que produciría print(json.dumps(asdict(event), ensure_ascii=False)). La diferencia no está en el formato de la línea — está en la maquinaria alrededor:

• Nivel de severidad, heredado de la lección 02: logger.log(logging.DEBUG, ...) puede quedar filtrado sin tocar ni una línea de código, algo que un print() nunca puede hacer.
• Handlers independientes del destino: el mismo logger puede escribir a la terminal (como aquí) y, sin cambiar log_event ni una dataclass, también a un archivo — exactamente lo que hace la lección 07 con RUN_LOG.jsonl.
• Un espacio de nombres ("reservo.observability"), que permite silenciar o amplificar esta parte del sistema sin afectar a otros loggers.

json.dumps arma la línea; logging decide si esa línea se emite, y a dónde. Las dos piezas juntas son las que hacen falta — ninguna alcanza sola.

Errores comunes

1. Olvidar ensure_ascii=False. Sin ese argumento, json.dumps es técnicamente correcto —el JSON sigue siendo válido—, pero cualquier acento sale escapado (Sofía en vez de Sofía), mucho más difícil de leer a simple vista en un archivo de logs. Confírmalo:

   print(json.dumps({"question": "Reserva Boardroom pro 1h para Sofía"}, ensure_ascii=False))
   print(json.dumps({"question": "Reserva Boardroom pro 1h para Sofía"}, ensure_ascii=True))

   {"question": "Reserva Boardroom pro 1h para Sofía"}
   {"question": "Reserva Boardroom pro 1h para Sofía"}
   

2. Olvidar logger.propagate = False. Sin esa línea, si el logger raíz de Python también tiene un handler configurado (algo común si otra parte del programa llamó a logging.basicConfig()), cada evento puede aparecer duplicado — una vez por el handler propio de "reservo.observability", otra por el handler heredado del logger raíz.

3. Intentar loguear un objeto que json.dumps no sabe serializar. Un dataclass con solo str/int/bool/dict/list en sus campos siempre es serializable — pero si alguien agrega un campo con un tipo no estándar (un set, una instancia de una clase propia, un objeto datetime), json.dumps lanza un TypeError en el momento de loguear, no antes. El Ejercicio 3 de esta lección lo confirma.

4. Pensar que una línea de JSON "ya es" un trace_id funcional. El campo trace_id de RunEvent en este ejemplo es el string fijo "run-demo", escrito a mano — no un identificador real, determinista, calculado a partir de las entradas del run. Eso es, con precisión, el trabajo de la lección 04.

5. Confundir el formato JSON Lines (una línea, un objeto) con un archivo JSON normal. Un archivo .json típico contiene un solo objeto o arreglo, que puede ocupar muchas líneas con indentación. Un archivo .jsonl (o RUN_LOG.jsonl, como en la lección 07) contiene muchos objetos independientes, uno por línea, sin ninguna coma ni corchete que los una — cada línea se parsea por separado. Intentar cargar un archivo .jsonl completo con un solo json.loads(archivo.read()) falla, porque no es un único documento JSON válido.

Ejercicios

Ejercicio 1: Parsea una línea de vuelta y extrae un campo (Fácil)

Genera una línea de log con log_event para un RunEvent con event="run_failed" y error="RuntimeError: max_iterations alcanzado (2)". Captura la línea que produce (puedes usar json.dumps(asdict(...)) directamente para esto, sin pasar por el logger, ya que el resultado es el mismo string), parséala de vuelta con json.loads, y extrae solo el campo error.

event = RunEvent(seq=5, trace_id="run-xyz", event="run_failed", question="Reserva algo", error="RuntimeError: max_iterations alcanzado (2)")
line = json.dumps(asdict(event), ensure_ascii=False)
print("línea generada:", line)

parsed = json.loads(line)
print("campo error   :", parsed["error"])

línea generada: {"seq": 5, "trace_id": "run-xyz", "event": "run_failed", "question": "Reserva algo", "error": "RuntimeError: max_iterations alcanzado (2)"}
campo error   : RuntimeError: max_iterations alcanzado (2)

Ejercicio 2: Confirma el problema de ensure_ascii=True con datos reales de Reservo (Medio)

Usando RunEvent, genera un evento con question="Reserva Boardroom pro 1h para Sofía" (nota la tilde). Serialízalo dos veces: una con ensure_ascii=False y otra con ensure_ascii=True (el valor por defecto de json.dumps si no se especifica). Compara las dos líneas.

event = RunEvent(seq=1, trace_id="run-sofia", event="run_started", question="Reserva Boardroom pro 1h para Sofía")

legible = json.dumps(asdict(event), ensure_ascii=False)
escapado = json.dumps(asdict(event))  # ensure_ascii=True por defecto

print("legible :", legible)
print("escapado:", escapado)
print("ambas son JSON valido, se parsean igual:", json.loads(legible) == json.loads(escapado))

legible : {"seq": 1, "trace_id": "run-sofia", "event": "run_started", "question": "Reserva Boardroom pro 1h para Sofía", "error": ""}
escapado: {"seq": 1, "trace_id": "run-sofia", "event": "run_started", "question": "Reserva Boardroom pro 1h para Sofía", "error": ""}
ambas son JSON valido, se parsean igual: True

Ejercicio 3: Provoca el TypeError de un campo no serializable (Difícil)

Define una dataclass BadEvent con un campo trace_id: str y un campo weird: set. Crea una instancia con weird={1, 2, 3} y trata de loguearla con log_event. Confirma el TypeError, y explica en una frase por qué las dataclasses de este módulo (RunEvent, y ToolCallEvent de la lección 05) se restringen deliberadamente a str/int/bool/dict/list.

from dataclasses import dataclass, asdict

@dataclass
class BadEvent:
    trace_id: str
    weird: set

event = BadEvent(trace_id="run-x", weird={1, 2, 3})
try:
    json.dumps(asdict(event))
except TypeError as exc:
    print(f"TypeError: {exc}")

TypeError: Object of type set is not JSON serializable

Resumen y siguiente paso

• Construimos la primera pieza real de observability/run_logger.py: un logger configurado para JSON (_handler + Formatter("%(message)s")), la dataclass RunEvent, y log_event, la función que convierte cualquier evento en una línea de JSON completa.
• Ejecutamos las dos primeras líneas de log reales de este módulo, y confirmamos, con json.loads, que son parseables de vuelta a un diccionario de Python — la diferencia central con el texto libre de print() de la lección anterior.
• Confirmamos, ejecutado, por qué ensure_ascii=False importa para prosa en español, y por qué un campo no serializable produce un TypeError en el momento de loguear.

Siguiente lección: 04 — El trace_id: correlacionando un run. Con el formato JSON resuelto, construimos la pieza que hace que cada línea sepa a qué run pertenece: un identificador determinista, nunca uuid4(), y el contextlib.contextmanager que abre y cierra un run con ese id.

Recursos adicionales

1. Python — json — json.dumps/json.loads, el par de funciones detrás de cada línea de este módulo, incluidos ensure_ascii y el TypeError de esta lección.
2. Python — dataclasses — @dataclass y asdict, la forma en que este módulo representa cada evento antes de serializarlo.
3. Python — logging.Formatter — La clase detrás de _handler.setFormatter(...), y por qué "%(message)s" es suficiente cuando el mensaje ya llega armado.
4. JSON Lines — El formato de "una línea, un objeto JSON" que este módulo usa desde esta lección, y que RUN_LOG.jsonl (lección 07) adopta como su formato de archivo.
5. Python 3.14 — What's New — La versión con la que se ejecutó cada línea de código de esta lección.