Dentro del Grafo: la máquina de estados de 10 nodos detrás de un agente agnóstico

"En la era dorada de los prototipos de IA, casi todo el mundo programa sus agentes con un bucle while True que reinyecta respuestas del modelo hasta que este decide detenerse. En un cuaderno interactivo parece magia; en un entorno de producción regulado es una receta para el desastre: bucles infinitos que devoran presupuestos de tokens, alucinaciones no auditadas y mutaciones destructivas sin control de acceso. Un agente fiable no es un bucle libre: es un autómata finito de estados formalmente compilado, donde cada transición está gobernada por políticas matemáticas y cada paso crítico está protegido por contratos deterministas."

1. Resumen Ejecutivo: La Ilusión del Bucle Infinito

La mayoría de los frameworks populares de agentes presentan el ciclo cognitivo como un patrón simple de tipo ReAct (Reason + Act): el modelo de lenguaje razona, decide invocar una herramienta, lee el resultado y repite el proceso de forma recursiva hasta concluir. Si bien este planteamiento resulta intuitivo para demostraciones rápidas, carece de las garantías requeridas en sistemas distribuidos corporativos: predictibilidad temporal, control de costes, contención de fallos y soberanía sobre los efectos secundarios.

En jg_agente, el runtime rechaza por completo el patrón de bucle descontrolado. En su lugar, el flujo de ejecución se formaliza como un grafo dirigido de estados acíclico y determinista montado sobre LangGraph (StateGraph). Este grafo no es estático ni contiene herramientas acopladas en su núcleo: se compila dinámicamente mediante GraphBuilder en función de las capacidades registradas en el CapabilityRegistry.

En este artículo desgranamos la anatomía completa de esta máquina de estados: la topología canónica de 10 nodos de app/core/graph.py, las funciones de enrutamiento condicional que gobiernan las bifurcaciones de seguridad, el mecanismo de interrupción asíncrona con persistencia de checkpoints para aprobación humana (interrupt()), la compensación automática mediante el patrón Saga y el guardarraíl determinista de evaluación que impide que cualquier ejecución exceda tres iteraciones.

2. Puntos de Fricción: El Antipatrón del while(True)

Al auditar implementaciones de agentes que fallan en producción, se observan sistemáticamente tres problemas estructurales derivados del patrón ReAct descontrolado:

1. Desbordamiento de Contexto y Explosión de Costes (Token Runaway): Un agente que entra en un ciclo de razonamiento repetitivo ante un error no manejado continúa invocando al modelo, acumulando mensajes en el prompt y agotando rápidamente la ventana de contexto. En cuestión de minutos, una sola tarea puede facturar cientos de miles de tokens sin haber completado su objetivo.

2. Falta de Separación entre Intención, Riesgo y Ejecución: Si el mismo nodo que infiere la respuesta ejecuta inmediatamente la herramienta, no existe oportunidad de aplicar políticas de seguridad intermedias, auditar argumentos ni requerir confirmación humana ante operaciones destructivas.

3. Topologías Monolíticas No Adaptativas: Un flujo conversacional básico no necesita planificadores complejos de múltiples pasos ni nodos de verificación de infraestructura; por el contrario, una automatización de despliegues exige validación de precondiciones, análisis de riesgos y transacciones de compensación. Acoplar una única topología fija para todos los casos de uso degrada la latencia en tareas simples y compromete la robustez en tareas críticas.

3. Arquitectura y Solución Técnica: Ensamblado Dinámico con GraphBuilder

El principio rector del microkernel jg_agente es el Agnosticismo Radical: el núcleo no conoce dominios de negocio, no tiene herramientas preinstaladas y no asume una topología rígida.

Cuando el agente arranca, inspecciona el CapabilityRegistry. En función de los plugins y herramientas cargados, GraphBuilder compila una instancia de CompiledStateGraph activando o desactivando nodos específicos de forma condicional.

Diagrama General de la Máquina de Estados (10 Nodos)

               ┌───────────────┐
               │     START     │
               └───────┬───────┘
                       │
                       ▼
               ┌───────────────┐
               │    observe    │  (Opcional: Sensores y telemetría)
               └───────┬───────┘
                       │
                       ▼
               ┌───────────────┐
               │    reason     │  (Obligatorio: Inferencia e intención)
               └───────┬───────┘
                       │
        ┌──────────────┴────────────────────────┬──────────────────────┐
        │ [intención directa: respond/conclude] │                      │ [requiere plan]
        ▼                                       ▼                      ▼
┌───────────────┐                       ┌───────────────┐      ┌───────────────┐
│   conclude    │                       │  risk_assess  │      │     plan      │ (Opcional)
└───────▲───────┘                       └───────┬───────┘      └───────┬───────┘
        │                                       │                      │
        │ [DENY / Rechazo]                      │                      ▼
        ├───────────────────────────────────────┤              ┌───────────────┐
        │                                       │              │  risk_assess  │
        │                                       ▼              └───────┬───────┘
        │                               ┌───────────────┐              │
        │                               │human_approval │ (Opcional)   │
        │                               └───────┬───────┘              │
        │                                       │                      │
        │                                       ▼                      ▼
        │                               ┌──────────────────────────────┴───────┐
        │                               │                execute               │ (Obligatorio)
        │                               └──────────────────┬───────────────────┘
        │                                                  │
        │                         ┌────────────────────────┴─────────────┐
        │                         │ [Fallo y reversible: requires_rollback]
        │                         ▼                                      ▼
        │                 ┌───────────────┐                      ┌───────────────┐
        │                 │    verify     │ (Opcional)           │   rollback    │ (Condicional Saga)
        │                 └───────┬───────┘                      └───────┬───────┘
        │                         │                                      │
        │                         ▼                                      ▼
        │                 ┌───────────────┐                      ┌───────────────┐
        │                 │   evaluate    │ (Obligatorio) ◄──────┤   evaluate    │
        │                 └───────┬───────┘                      └───────────────┘
        │                         │
        │ [Satisfecho / Fallo]    │ [Iteración < 3 y Needs More Info]
        ├─────────────────────────┴──────────────────────────────────────┐
        │                                                                │
        ▼                                                                ▼
┌───────────────┐                                                (Ciclo a reason)
│   conclude    │ ──► END
└───────────────┘

4. Análisis Detallado de los 10 Nodos del Grafo

A continuación se detalla la responsabilidad, condiciones de activación y diseño de cada uno de los 10 nodos definidos en app/core/graph.py:

1. observe (Opcional)

• Función: Captura el estado físico del entorno antes de que el modelo comience a razonar. Recopila variables de sistema, telemetría, archivos de contexto o estado de conexiones externas delegando en herramientas de tipo ToolType.OBSERVER.
• Condición de activación: Se enlaza en el grafo si el registro contiene herramientas con rol observador (has_observer = True). Si no existen observadores, el nodo START conecta directamente con reason.

2. reason (Obligatorio)

• Función: Es el motor cognitivo central. Invoca al LLM configurado pasándole el estado actual, el historial conversacional y el catálogo de capacidades. Su salida es un modelo estructurado IntentModel que tipifica la acción: responder directamente (respond), concluir (conclude) o solicitar una operación externa.
• Condición de activación: Siempre activo.

3. plan (Opcional)

• Función: Cuando la orden del usuario es compleja o involucra múltiples acciones secuenciales, este nodo genera un grafo acíclico de pasos estructurados (PlanModel). Desglosa la meta en tareas individuales con dependencias explícitas.
• Condición de activación: Se enlaza si el registro incluye motores de planificación (has_planner = True).

4. risk_assess (Opcional)

• Función: Evalúa el nivel de riesgo de la operación propuesta (LOW, MEDIUM, HIGH, CRITICAL) contrastándola con la matriz de seguridad y el SecurityGuard. Detecta si la herramienta es destructiva, si opera sobre el sistema de archivos fuera del Workspace Jail o si el checkpointer se encuentra en modo degradado.
• Condición de activación: Activo siempre que existan herramientas registradas con niveles de riesgo declarados (has_risk_assessment = True).

5. human_approval (Opcional)

• Función: Si risk_assess determina que la operación requiere autorización (PolicyDecision.REQUIRE_HUMAN_APPROVAL), este nodo congela el flujo mediante la primitiva interrupt() de LangGraph. El estado se persiste de forma atómica en PostgreSQL y el sistema queda a la espera de que un operador humano resuelva la compuerta desde el Control Plane.
• Condición de activación: Activo si has_human_approval = True.

6. execute (Obligatorio)

• Función: Despacha la herramienta autorizada bajo un entorno de sandboxing estricto custodiado por SecurityGuard.execute_tool(). Impone un timeout asíncrono duro (asyncio.wait_for, por defecto 30 segundos) para evitar bloqueos en el Event Loop y registra un ExecutionRecord estructurado con los resultados y el contexto de rollback.
• Condición de activación: Siempre activo si existen herramientas ejecutables registradas.

7. verify (Opcional)

• Función: Comprobación física de post-condiciones. No confía en el código de retorno de la herramienta: ejecuta verificaciones deterministas independientes (por ejemplo, comprobar si un puerto de red quedó escuchando o si un archivo efectivamente existe en disco).
• Condición de activación: Se conecta si existen herramientas de verificación registradas (has_verifier = True).

8. rollback (Condicional)

• Función: Implementa la orquestación del patrón Saga. Si un paso de ejecución o verificación falla de forma irrecuperable y existen pasos previos reversibles completados, este nodo recorre en orden inverso (LIFO) los registros históricos ejecutando las funciones rollback_func con su contexto original.
• Condición de activación: Se activa condicionalmente durante la ejecución cuando state["requires_rollback"] == True.

9. evaluate (Obligatorio)

• Función: Realiza un juicio semántico sobre el grado de cumplimiento del objetivo tras la ejecución o compensación. Determina si la meta está satisfecha (SATISFIED), si faltan capacidades (BLOCKED_MISSING_CAPABILITY) o si se requiere información adicional (NEEDS_MORE_INFO).
• Guardarraíl Determinista: Comprueba el contador state["iteration"]. Si la iteración alcanza el límite máximo estricto (3 iteraciones), fuerza la transición inmediata a conclude, impidiendo matemáticamente cualquier bucle infinito.

10. conclude (Obligatorio)

• Función: Nodo terminal del ciclo. Consolida los resultados parciales, formatea la respuesta final depurando metacomentarios técnicos innecesarios y construye el contrato canónico AgentOutput listo para ser entregado al cliente emisor.
• Condición de activación: Siempre activo. Su salida apunta directamente a END.

5. Lógica de Enrutamiento Condicional en Código Fuente

Las transiciones del grafo no están grabadas de forma estática: se calculan en tiempo de ejecución mediante funciones deterministas en app/core/graph.py.

Enrutamiento tras el Razonamiento (route_after_reason)

# jg_agente/app/core/graph.py

def route_after_reason(
    state: AgentState,
    has_planner: bool,
    has_tools: bool,
    has_risk_assessment: bool = False,
) -> str:
    """Enruta desde 'reason' según la intención detectada y las capacidades disponibles."""
    intent = state.get("intent")
    action_type = ""
    if isinstance(intent, IntentModel):
        action_type = intent.action_type
    elif isinstance(intent, dict):
        action_type = intent.get("action_type", "")

    # Respuestas puramente conversacionales van directo a conclusión
    if action_type in ("conclude", "respond"):
        return "conclude"
    
    # Si la tarea requiere plan y existe planificador registrado
    if has_planner:
        return "plan"
        
    # Si requiere evaluación de riesgo previa
    if has_risk_assessment:
        return "risk_assess"
        
    # Ejecución directa si hay herramientas disponibles
    if has_tools:
        return "execute"
        
    return "conclude"

Enrutamiento tras la Evaluación de Riesgo (route_after_risk_assess)

# jg_agente/app/core/graph.py

def route_after_risk_assess(state: AgentState, has_human_approval: bool) -> str:
    """Enruta desde 'risk_assess' evaluando la política de seguridad determinista."""
    risk_info = state.get("risk_assessment") or {}
    decision = risk_info.get("decision") if isinstance(risk_info, dict) else getattr(risk_info, "decision", None)

    # Si SecurityGuard emite DENY (política denegada o jail escape), abortar
    if decision in (PolicyDecision.DENY, PolicyDecision.DENY.value):
        return "conclude"

    # Si requiere autorización humana, transicionar a compuerta HITL
    if decision in (PolicyDecision.REQUIRE_HUMAN_APPROVAL, PolicyDecision.REQUIRE_HUMAN_APPROVAL.value):
        if has_human_approval:
            return "human_approval"
        return "conclude"

    return "execute"

Enrutamiento tras la Evaluación con Guardarraíl de Iteración (route_after_evaluation)

# jg_agente/app/core/graph.py

def route_after_evaluation(state: AgentState, has_rollback: bool = False) -> str:
    """Aplica guardarraíles deterministas de iteración y estado de objetivo."""
    # 1. Guardarraíl de compensación Saga si la evaluación detecta fallo explícito
    if state.get("requires_rollback") and has_rollback:
        return "rollback"

    # 2. Guardarraíl estricto contra bucles infinitos: Máximo 3 iteraciones
    iteration = state.get("iteration", 1)
    max_iterations = state.get("max_iterations", 3)
    if iteration >= max_iterations:
        return "conclude"

    # 3. Inspección del juicio del evaluador
    eval_records = state.get("evaluation_records") or []
    if not eval_records:
        return "conclude"

    last_eval = eval_records[-1]
    status_val = str(getattr(last_eval, "goal_status", "")).lower().strip()

    # Si el objetivo se cumplió, falló definitivamente o faltan capacidades -> Concluir
    if status_val in ("satisfied", "failed", "blocked_missing_capability"):
        return "conclude"

    # Si el evaluador determina que se requiere más información con capacidades activas -> Ciclo a reason
    if status_val == "needs_more_info":
        return "reason"

    return "conclude"

6. Consideraciones de Producción, Trade-offs y Límites

1. Latencia Acumulada vs. Rigor Estructural: Un grafo de 10 nodos ofrece garantías de seguridad inigualables, pero introduce latencia inherente debido a la serialización de checkpoints en PostgreSQL y las sucesivas evaluaciones del modelo. Para tareas conversacionales simples, la compilación condicional de GraphBuilder es fundamental: si una petición solo requiere responder una pregunta, el grafo se reduce en tiempo de ejecución a la ruta directa START → reason → conclude → END, eliminando los 7 nodos restantes y ejecutándose en milisegundos.

2. Límites del Juicio Semántico en evaluate: El nodo de evaluación semántica delega en el LLM la decisión de si la meta fue alcanzada. Aunque existe el guardarraíl determinista de 3 iteraciones que previene el desbordamiento infinito de tokens, un modelo de lenguaje puede evaluar erróneamente que una tarea no fue satisfecha debido a discrepancias cosméticas en la salida de una herramienta. Por ello, las herramientas críticas deben incorporar verificadores físicos deterministas (verify) que inyecten hechos incontestables en el estado antes de la evaluación.

3. Atomicidad de Checkpoints en Nodos Interrumpidos: Cuando el nodo human_approval suspende la ejecución mediante interrupt(), el estado exacto del grafo se congela en PostgreSQL. Si durante el tiempo de espera el clúster sufre un reinicio, la reanudación es completamente determinista: el runner de LangGraph reconstruye el thread desde el último snapshot válido sin volver a ejecutar los nodos precedentes.

7. Reflexión de Arquitectura: La Belleza del Autómata Finito

En los albores de la computación aprendimos que un autómata finito con transiciones bien definidas es la única base sólida sobre la que construir sistemas predecibles. La explosión de los modelos de lenguaje nos tentó a olvidar esa lección, pretendiendo que la flexibilidad del lenguaje natural podía sustituir la rigidez del diseño formal de sistemas.

El tiempo ha demostrado lo contrario. La verdadera potencia de un agente de IA no reside en dejarlo libre en un bucle ciego, sino en enjaular su capacidad de razonamiento dentro de una máquina de estados implacable. Deja que el LLM proponga el razonamiento en reason; pero haz que el riesgo lo juzgue risk_assess, que los cambios los custodie SecurityGuard, que la pausa la controle un motor de base de datos y que el bucle lo corte un guardarraíl matemático inmutable.

8. Enlaces y Referencias en el Repositorio

• Ensamblador de Grafos Agnóstico: jg_agente/app/core/graph.py
• Registro Dinámico de Capacidades: jg_agente/app/core/registry.py
• Definición de Estados y Modelos de Dominio: jg_agente/app/models/state.py
• Nodo de Aprobación Humana con interrupt(): jg_agente/app/nodes/human_approval.py
• Nodo de Compensación Rollback (Sagas): jg_agente/app/nodes/rollback.py
• Suite de Pruebas de Transiciones y Topología: jg_agente/tests/test_graph_builder.py