Agregué una Caja Negra de 71 Líneas a Mi Agente Python y Consulté la Falla de $200 con DuckDB

El incidente comenzó con una aburrida tarea de automatización de soporte.

Tomar una solicitud de usuario, buscar en un índice privado de documentos, resumir la respuesta y entregar el resultado a un revisor. Nada heroico. El tipo de agente Python que construyes cuando la demo terminó y el flujo de trabajo real comienza.

Entonces una ejecución se quedó atrapada en un bucle de reintentos.

No quemó $200 antes de que lo detectara. La ejecución de prueba real fue más barata. El problema era la proyección: el mismo bucle defectuoso, la misma búsqueda de documentos, las mismas llamadas al modelo, ejecutándose dentro del lote nocturno. La estimación se acercó a $200 por una falla completamente evitable.

La respuesta que produjo parecía lo suficientemente pulida como para pasar una revisión somnolienta. La traza detrás de ella no era pulida en absoluto. El agente había llamado a la herramienta correcta con el input equivocado, reintentado contra contexto obsoleto, resumido resultados viejos y seguido pagando por cada turno.

Fue entonces cuando dejé de tratar al agente como una funcionalidad de chat.

Empecé a tratarlo como un sistema que necesita una caja negra.

No un dashboard. No un stack completo de observabilidad. No otro servicio hosteado.

Simplemente un archivo local que pueda responder:

• ¿Qué intentó hacer el agente?
• ¿Qué herramienta llamó?
• ¿Qué input recibió la herramienta?
• ¿Falló la herramienta?
• ¿Cuánto tardó?
• ¿La ejecución cruzó un límite de costo o de turnos?
• ¿Puedo consultar la ejecución después de que todo terminó?

Vamos a construir esa caja negra en Python puro, y luego usaremos DuckDB para inspeccionarla como una pequeña base de datos de crash.

Antes y Después

Antes de la solución, depurar se veía así:

La respuesta final está mal.
El modelo probablemente alucinó.
Quizás la herramienta de búsqueda devolvió datos incorrectos.
Quizás el bucle de reintentos reutilizó un mensaje viejo.
Quizás el pico de costo vino de la llamada al modelo.

Eso no es depurar. Eso es adivinar con resaltado de sintaxis.

Después de la solución, depurar se veía así:

El turno 1 llamó a search_docs con la query incorrecta.
La herramienta expiró después de 147.82 ms.
El reintento usó contexto obsoleto.
El guardia detuvo la ejecución en $0.0124.
DuckDB muestra un tool_error y un guard_stop.

Mismo bug. Día muy diferente.

La Forma del Problema

Un script Python normal generalmente falla en un solo lugar.

Un agente falla a lo largo de una cadena.

Solicitud del Usuario -> Decisión del Modelo -> Llamada a Herramienta -> Resultado de Herramienta -> Siguiente Turno -> Respuesta Final

Si solo registras la respuesta final, tienes una entrada de diario.

Si registras la cadena, tienes evidencia.

El formato útil más simple es JSONL. Un evento por línea.

{"type":"tool_start","tool":"search_docs","input":{"query":"límites de tasa"}}
{"type":"tool_end","tool":"search_docs","duration_ms":83.4,"ok":true}
{"type":"turn_end","turn":2,"total_cost_usd":0.0041}

JSONL es aburrido exactamente de la manera correcta. Agrega líneas limpiamente, sobrevive crashes mejor que un gran documento JSON y puede consultarse con herramientas normales.

Un Pequeño Recorder Que Hace Trabajo Real

Aquí está el recorder.

Hace cuatro cosas:

• asigna a cada ejecución un id único
• escribe eventos JSONL en modo append-only
• mide la duración de las herramientas
• sanitiza secretos obvios antes de escribir cualquier cosa a disco

from __future__ import annotations

import json
import re
import time
import traceback
from contextlib import contextmanager
from dataclasses import asdict, dataclass, field
from pathlib import Path
from typing import Any, Iterator
from uuid import uuid4


SECRET_KEYS = re.compile(
    r"(api[_-]?key|token|password|secret|authorization|cookie)",
    re.IGNORECASE,
)


@dataclass
class Event:
    run_id: str
    event_id: str
    type: str
    timestamp: float
    data: dict[str, Any] = field(default_factory=dict)


def sanitize(value: Any) -> Any:
    if isinstance(value, dict):
        cleaned = {}
        for key, item in value.items():
            if SECRET_KEYS.search(str(key)):
                cleaned[key] = "[redacted]"
            else:
                cleaned[key] = sanitize(item)
        return cleaned

    if isinstance(value, list):
        return [sanitize(item) for item in value]

    return value


class AgentBlackBox:
    def __init__(self, path: str | Path, run_id: str | None = None) -> None:
        self.path = Path(path)
        self.run_id = run_id or uuid4().hex
        self.path.parent.mkdir(parents=True, exist_ok=True)

    def record(self, event_type: str, **data: Any) -> None:
        event = Event(
            run_id=self.run_id,
            event_id=uuid4().hex,
            type=event_type,
            timestamp=time.time(),
            data=sanitize(data),
        )

        with self.path.open("a", encoding="utf-8") as file:
            file.write(json.dumps(asdict(event), default=str) + "\n")

    @contextmanager
    def tool(self, name: str, **tool_input: Any) -> Iterator[None]:
        started = time.perf_counter()
        self.record("tool_start", tool=name, input=tool_input)

        try:
            yield
        except Exception as exc:
            self.record(
                "tool_error",
                tool=name,
                error_type=type(exc).__name__,
                error=str(exc),
                traceback=traceback.format_exc(limit=6),
                duration_ms=round((time.perf_counter() - started) * 1000, 2),
            )
            raise
        else:
            self.record(
                "tool_end",
                tool=name,
                ok=True,
                duration_ms=round((time.perf_counter() - started) * 1000, 2),
            )

La función sanitize() no es perfecta. Es un cinturón de seguridad, no una bóveda.

Aun así, previene la versión más embarazosa de este patrón: construir una traza de depuración útil que silenciosamente almacena API keys.

Envuelve Una Herramienta Primero

Comienza con una herramienta. No instrumentes todo el primer día.

import random
import time


def search_docs(query: str, api_key: str) -> list[str]:
    time.sleep(random.uniform(0.05, 0.2))

    if "timeout" in query:
        raise TimeoutError("Document search timed out")

    return [
        "JSONL funciona bien para trazas append-only.",
        "Los context managers son útiles alrededor de llamadas a herramientas.",
        "DuckDB puede consultar archivos JSON sin un servidor.",
    ]

Ahora registra la llamada:

box = AgentBlackBox("traces/run.jsonl")

query = "python agent trace format"

with box.tool("search_docs", query=query, api_key="***"):
    docs = search_docs(query=query, api_key="***")

box.record("tool_result", tool="search_docs", result_count=len(docs))

Abre traces/run.jsonl y la key está redactada.

{"tool":"search_docs","input":{"query":"python agent trace format","api_key":"[redacted]"}}

Ese pequeño detalle importa. Depurar no debería crear un segundo incidente.

Agrega un Guardia de Ejecución Barato

La mayoría de las historias de agentes descontrolados comienzan con un bucle que parecía inofensivo.

Así que la caja negra no solo debería registrar lo que sucedió. Debería registrar cuándo se negó a continuar.

class RunStopped(RuntimeError):
    pass


def stop_if_needed(
    box: AgentBlackBox,
    *,
    turn: int,
    max_turns: int,
    spent_usd: float,
    max_usd: float,
) -> None:
    box.record(
        "guard_check",
        turn=turn,
        max_turns=max_turns,
        spent_usd=round(spent_usd, 6),
        max_usd=round(max_usd, 6),
    )

    if turn > max_turns:
        box.record("guard_stop", reason="max_turns", turn=turn)
        raise RunStopped(f"Stopped at turn {turn}. Max turns is {max_turns}.")

    if spent_usd > max_usd:
        box.record("guard_stop", reason="budget", spent_usd=spent_usd)
        raise RunStopped(f"Stopped at ${spent_usd:.4f}. Budget is ${max_usd:.4f}.")

Esto no es facturación exacta. Usa la respuesta de tu proveedor para conteos reales de tokens cuando los tengas.

El objetivo aquí es un cable trampa local. Quieres que la ejecución deje una razón clara cuando se detiene.

Un Pequeño Bucle de Agente

Este bucle simulado mantiene las partes móviles pequeñas.

Reemplaza la sección del modelo simulado con tu llamada real al modelo.

def estimate_cost(input_tokens: int, output_tokens: int) -> float:
    return input_tokens * 0.0000005 + output_tokens * 0.0000015


def run_agent(question: str) -> str:
    box = AgentBlackBox("traces/run.jsonl")
    messages = [{"role": "user", "content": question}]
    spent_usd = 0.0
    max_turns = 3
    max_usd = 0.01

    box.record("run_start", question=question, max_turns=max_turns, max_usd=max_usd)

    for turn in range(1, max_turns + 1):
        stop_if_needed(
            box,
            turn=turn,
            max_turns=max_turns,
            spent_usd=spent_usd,
            max_usd=max_usd,
        )

        box.record("turn_start", turn=turn, message_count=len(messages))

        # Simula que el modelo eligió este input de herramienta.
        query = question if turn == 1 else "python jsonl duckdb traces"

        with box.tool("search_docs", query=query, api_key="***"):
            docs = search_docs(query=query, api_key="***")

        messages.append({"role": "tool", "content": "\n".join(docs)})

        turn_cost = estimate_cost(
            input_tokens=sum(len(message["content"].split()) for message in messages),
            output_tokens=120,
        )
        spent_usd += turn_cost

        box.record(
            "turn_end",
            turn=turn,
            message_count=len(messages),
            turn_cost_usd=round(turn_cost, 6),
            total_cost_usd=round(spent_usd, 6),
        )

    answer = "Record every tool call as JSONL, then query failures after the run."
    box.record("run_end", answer=answer, total_cost_usd=round(spent_usd, 6))
    return answer

Ejecútalo una vez con una pregunta normal.

print(run_agent("How should I debug Python agent tools?"))

Luego ejecútalo con una mala.

print(run_agent("timeout during document search"))

La segunda ejecución debería fallar, pero ahora falla con un rastro.

Para forzar una parada por presupuesto durante las pruebas, establece temporalmente max_usd = 0.0001. La siguiente verificación del guardia escribirá un evento guard_stop en lugar de dejar que el bucle continúe silenciosamente.

Consulta el Crash con DuckDB

Esta es la parte que hace que JSONL se sienta menos como logging y más como una herramienta de debugging.

Instala DuckDB:

pip install duckdb

Luego consulta la traza:

import duckdb


def query_trace(path: str = "traces/run.jsonl") -> None:
    con = duckdb.connect()

    con.sql(
        f"""
        create or replace view events as
        select *
        from read_json_auto('{path}');
        """
    )

    print("Conteo de eventos")
    con.sql(
        """
        select type, count(*) as events
        from events
        group by type
        order by events desc;
        """
    ).show()

    print("Errores de herramientas")
    con.sql(
        """
        select
            data.tool as tool,
            data.error_type as error_type,
            data.error as error,
            data.duration_ms as duration_ms
        from events
        where type = 'tool_error';
        """
    ).show()

    print("Herramientas lentas")
    con.sql(
        """
        select
            data.tool as tool,
            data.duration_ms as duration_ms
        from events
        where type = 'tool_end'
        order by data.duration_ms desc
        limit 5;
        """
    ).show()

Ahora ejecuta:

query_trace()

El resultado debería verse así:

Conteo de eventos
+-------------+--------+
| type        | events |
+-------------+--------+
| guard_check |      4 |
| turn_start  |      3 |
| tool_start  |      3 |
| tool_end    |      2 |
| tool_error  |      1 |
| guard_stop  |      1 |
+-------------+--------+

Y la fila del crash ahora es un resultado de query, no un misterio:

Errores de herramientas
+-------------+--------------+---------------------------+-------------+
| tool        | error_type   | error                     | duration_ms |
+-------------+--------------+---------------------------+-------------+
| search_docs | TimeoutError | Document search timed out |      147.82 |
+-------------+--------------+---------------------------+-------------+

Puedes responder preguntas que los logs de print normales hacen molestas:

• ¿Qué herramientas fallaron con más frecuencia?
• ¿Qué herramienta fue la más lenta?
• ¿Qué turno cruzó el presupuesto?
• ¿El mismo input falló repetidamente?
• ¿El guardia detuvo la ejecución, o la herramienta crasheó primero?

Esa es la mejora.

No “tengo logs”.

“Puedo interrogar la ejecución.”

Qué Registraría en un Proyecto Real

Para una demo, la traza anterior es suficiente.

Para un proyecto real, agregaría estos campos:

• model
• provider
• prompt_hash
• tool_schema_version
• input_tokens
• output_tokens
• finish_reason
• retry_count
• user_id_hash
• environment

No registraría esto por defecto:

• tokens de acceso sin procesar
• documentos privados
• prompts completos de clientes
• respuestas completas de herramientas con datos sensibles
• cookies o headers de solicitud

La regla de seguridad aburrida es simple:

Registra suficiente para depurar el comportamiento. No registres suficiente para dañar a alguien.

El Patrón en Una Sola Frase

Cada ejecución de agente debería producir un flujo de eventos local, append-only, que sea seguro de conservar, fácil de consultar y útil después de que el proceso falle.

Esa frase es menos emocionante que un nuevo truco de prompts.

Pero es más probable que te salve el fin de semana.

Archivo Completo

Aquí está el ejemplo completo en un solo lugar.

from __future__ import annotations

import json
import random
import re
import time
import traceback
from contextlib import contextmanager
from dataclasses import asdict, dataclass, field
from pathlib import Path
from typing import Any, Iterator
from uuid import uuid4


SECRET_KEYS = re.compile(
    r"(api[_-]?key|token|password|secret|authorization|cookie)",
    re.IGNORECASE,
)


@dataclass
class Event:
    run_id: str
    event_id: str
    type: str
    timestamp: float
    data: dict[str, Any] = field(default_factory=dict)


def sanitize(value: Any) -> Any:
    if isinstance(value, dict):
        return {
            key: "[redacted]" if SECRET_KEYS.search(str(key)) else sanitize(item)
            for key, item in value.items()
        }

    if isinstance(value, list):
        return [sanitize(item) for item in value]

    return value


class AgentBlackBox:
    def __init__(self, path: str | Path, run_id: str | None = None) -> None:
        self.path = Path(path)
        self.run_id = run_id or uuid4().hex
        self.path.parent.mkdir(parents=True, exist_ok=True)

    def record(self, event_type: str, **data: Any) -> None:
        event = Event(
            run_id=self.run_id,
            event_id=uuid4().hex,
            type=event_type,
            timestamp=time.time(),
            data=sanitize(data),
        )

        with self.path.open("a", encoding="utf-8") as file:
            file.write(json.dumps(asdict(event), default=str) + "\n")

    @contextmanager
    def tool(self, name: str, **tool_input: Any) -> Iterator[None]:
        started = time.perf_counter()
        self.record("tool_start", tool=name, input=tool_input)

        try:
            yield
        except Exception as exc:
            self.record(
                "tool_error",
                tool=name,
                error_type=type(exc).__name__,
                error=str(exc),
                traceback=traceback.format_exc(limit=6),
                duration_ms=round((time.perf_counter() - started) * 1000, 2),
            )
            raise
        else:
            self.record(
                "tool_end",
                tool=name,
                ok=True,
                duration_ms=round((time.perf_counter() - started) * 1000, 2),
            )


class RunStopped(RuntimeError):
    pass


def stop_if_needed(
    box: AgentBlackBox,
    *,
    turn: int,
    max_turns: int,
    spent_usd: float,
    max_usd: float,
) -> None:
    box.record(
        "guard_check",
        turn=turn,
        max_turns=max_turns,
        spent_usd=round(spent_usd, 6),
        max_usd=round(max_usd, 6),
    )

    if turn > max_turns:
        box.record("guard_stop", reason="max_turns", turn=turn)
        raise RunStopped(f"Stopped at turn {turn}. Max turns is {max_turns}.")

    if spent_usd > max_usd:
        box.record("guard_stop", reason="budget", spent_usd=spent_usd)
        raise RunStopped(f"Stopped at ${spent_usd:.4f}. Budget is ${max_usd:.4f}.")


def search_docs(query: str, api_key: str) -> list[str]:
    time.sleep(random.uniform(0.05, 0.2))

    if "timeout" in query:
        raise TimeoutError("Document search timed out")

    return [
        "JSONL works well for append-only traces.",
        "Context managers are useful around tool calls.",
        "DuckDB can query JSON files without a server.",
    ]


def estimate_cost(input_tokens: int, output_tokens: int) -> float:
    return input_tokens * 0.0000005 + output_tokens * 0.0000015


def run_agent(question: str) -> str:
    box = AgentBlackBox("traces/run.jsonl")
    messages = [{"role": "user", "content": question}]
    spent_usd = 0.0
    max_turns = 3
    max_usd = 0.01

    box.record("run_start", question=question, max_turns=max_turns, max_usd=max_usd)

    for turn in range(1, max_turns + 1):
        stop_if_needed(
            box,
            turn=turn,
            max_turns=max_turns,
            spent_usd=spent_usd,
            max_usd=max_usd,
        )

        box.record("turn_start", turn=turn, message_count=len(messages))

        query = question if turn == 1 else "python jsonl duckdb traces"

        with box.tool("search_docs", query=query, api_key="***"):
            docs = search_docs(query=query, api_key="***")

        messages.append({"role": "tool", "content": "\n".join(docs)})

        turn_cost = estimate_cost(
            input_tokens=sum(len(message["content"].split()) for message in messages),
            output_tokens=120,
        )
        spent_usd += turn_cost

        box.record(
            "turn_end",
            turn=turn,
            message_count=len(messages),
            turn_cost_usd=round(turn_cost, 6),
            total_cost_usd=round(spent_usd, 6),
        )

    answer = "Record every tool call as JSONL, then query failures after the run."
    box.record("run_end", answer=answer, total_cost_usd=round(spent_usd, 6))
    return answer


if __name__ == "__main__":
    print(run_agent("How should I debug Python agent tools?"))

Hay una línea en ese archivo completo que vale la pena observar:

box.record("run_start", question=question, max_turns=max_turns, max_usd=max_usd)

Esa línea cambia la postura del programa.

La ejecución ya no es una conversación privada con un modelo. Es una ejecución registrada con una traza que puedes inspeccionar, consultar y mejorar.

Esa es la diferencia entre una demo y algo en lo que puedes confiar.

¿Qué agregarías después: hashes de prompts, conteos de tokens, screenshots, checkpoints o fixtures de herramientas reproducibles?