Depurar sistemas multiagente

Al terminar esta lección podrás reconstruir qué hizo cada agente de una ejecución, en qué orden y con qué información, y volver a ejecutar una parte concreta con los mismos datos para encontrar el fallo. Sin esto, depurar un sistema multiagente es adivinar.

La idea

Depurar un agente ya es difícil: el mismo prompt puede tomar caminos distintos. Con varios agentes se suma otro problema: el síntoma aparece en un sitio y la causa está en otro. La respuesta final omite un dato porque la síntesis lo descartó, o porque el subagente no lo encontró, o porque su brief no lo pedía. Necesitas ver la trayectoria de cada agente y cómo se conectan.

El artículo de Anthropic How we built our multi-agent research system lo resume desde la experiencia: los agentes toman decisiones dinámicas y no son deterministas entre ejecuciones, así que depurar es más difícil, y añadir trazado completo en producción les permitió diagnosticar por qué fallaban y corregirlo de forma sistemática. Dos herramientas resuelven la mayor parte: trazas enlazadas y reproducción.

Trazas enlazadas

Cada evento que registras lleva tres identificadores: traza_id (la ejecución completa), span_id (este agente o esta llamada) y padre_id (quién lo lanzó). Con eso reconstruyes el árbol de la ejecución.

flowchart TD
  T[traza 9c1: orquestador] --> A[span a1: subagente proveedores]
  T --> B[span b2: subagente autoalojado]
  A --> A1[llamada modelo]
  A --> A2[herramienta buscar]
  B --> B1[llamada modelo]
  T --> S[span s9: síntesis]

Qué registrar en cada llamada al modelo: agente, paso, stop_reason, herramientas pedidas con sus argumentos, usage y duración. En cada herramienta: nombre, argumentos, resultado (recortado) y si fue error. Y al lanzar un subagente, el brief completo: es el primer sitio donde mirar cuando un subagente hace algo raro.

import json, time, uuid, contextvars, pathlib

_span_actual = contextvars.ContextVar("span_actual", default=None)

class Trazador:
    def __init__(self, ruta: str = "trazas.jsonl"):
        self.ruta = pathlib.Path(ruta)
        self.traza_id = uuid.uuid4().hex[:8]

    def evento(self, tipo: str, **datos) -> None:
        reg = {"t": time.time(), "traza_id": self.traza_id, "span_id": _span_actual.get(),
               "tipo": tipo, **datos}
        with self.ruta.open("a", encoding="utf-8") as f:
            f.write(json.dumps(reg, ensure_ascii=False, default=str) + "\n")

    def span(self, agente: str, **datos):
        trazador = self
        class _Span:
            def __enter__(self_):
                self_.padre = _span_actual.get()
                self_.id = uuid.uuid4().hex[:8]
                self_.token = _span_actual.set(self_.id)
                trazador.evento("inicio", agente=agente, padre_id=self_.padre, **datos)
                return self_
            def __exit__(self_, tipo_exc, exc, tb):
                trazador.evento("fin", agente=agente, error=repr(exc) if exc else None)
                _span_actual.reset(self_.token)
        return _Span()

tr = Trazador()

def registrar_respuesta(agente: str, paso: int, resp) -> None:
    tr.evento("modelo", agente=agente, paso=paso, stop_reason=resp.stop_reason,
              herramientas=[{"name": b.name, "input": b.input}
                            for b in resp.content if b.type == "tool_use"],
              usage=resp.usage.model_dump())

Se usa envolviendo cada subagente: with tr.span("subagente-proveedores", brief=asdict(brief)): .... Como contextvars se copia en cada tarea de asyncio, los subagentes lanzados con asyncio.gather dentro del span del orquestador heredan su padre_id correctamente.

Un archivo JSONL se filtra con cualquier herramienta. Para producción, el mismo modelo de datos encaja con estándares de trazado distribuido y con las plataformas de observabilidad (Observabilidad).

Reproducir una ejecución

Un fallo que no puedes repetir no lo puedes arreglar con confianza. Las respuestas del modelo varían, pero buena parte de la variación viene del entorno: una búsqueda devuelve resultados distintos hoy que ayer. La técnica: graba las respuestas de las herramientas y reprodúcelas.

import hashlib, json, pathlib

class Grabadora:
    """modo='grabar' ejecuta y guarda; modo='reproducir' devuelve lo guardado."""
    def __init__(self, ruta: str, modo: str = "grabar"):
        self.ruta, self.modo = pathlib.Path(ruta), modo
        self.cinta = json.loads(self.ruta.read_text()) if self.ruta.exists() else {}

    def _clave(self, nombre: str, entrada: dict) -> str:
        return hashlib.sha256(f"{nombre}:{json.dumps(entrada, sort_keys=True)}".encode()).hexdigest()[:16]

    def envolver(self, nombre: str, funcion):
        async def envuelta(**entrada):
            clave = self._clave(nombre, entrada)
            if self.modo == "reproducir":
                if clave not in self.cinta:
                    raise KeyError(f"{nombre}({entrada}) no está grabada")
                return self.cinta[clave]
            salida = await funcion(**entrada)
            self.cinta[clave] = salida
            self.ruta.write_text(json.dumps(self.cinta, ensure_ascii=False))
            return salida
        return envuelta

Con las herramientas grabadas y el brief guardado en la traza, puedes relanzar solo el subagente sospechoso con el mismo brief y el mismo entorno. Si el fallo se repite, cambia una cosa (el brief, la descripción de una herramienta, el modelo) y compara. Si se repite a veces, ejecútalo diez veces y cuenta: estás ante un problema de fiabilidad, y la herramienta adecuada es una eval (Evals para agentes).

Guarda también la versión exacta del modelo y de los prompts de cada ejecución. Sin eso, «ayer funcionaba» no se puede comprobar.

Fallos típicos y dónde mirar

Una pauta útil: busca el primer evento donde la información deja de ser correcta. Si el subagente encontró el dato y la síntesis no lo usó, el problema está en la síntesis. Si el subagente nunca lo buscó, mira su brief. Si lo buscó y la herramienta no lo devolvió, mira la herramienta.

Errores frecuentes

• Trazar solo el orquestador. Ves el síntoma y nunca la causa.
• Registrar sin enlazar. Mil eventos sin padre_id son una lista, no un árbol.
• Depurar con ejecuciones nuevas cada vez. Cambia el entorno y no sabes si tu arreglo funcionó o tuviste suerte.
• Arreglar el prompt de la síntesis cuando el fallo está en un brief. Sigue la información hacia atrás.

Ejercicio

Resumen

• En multiagente el síntoma y la causa suelen estar en agentes distintos.
• Traza cada agente con traza_id, span_id y padre_id, y guarda el brief.
• Graba las respuestas de herramientas para reproducir una ejecución y aislar el fallo.
• Busca el primer punto donde la información deja de ser correcta.
• Los fallos intermitentes se miden con evals, no con una ejecución suelta.

Para seguir

• Observabilidad y trazas
• Evals de trayectoria: cómo llegó, no solo qué dijo
• Sistema multiagente de investigación