← Arquitectura de Agentes de IA

Capítulo 1

El loop de control: qué convierte a un LLM en agente

1. Concepto central

Tener herramientas no te hace agente. Lo que te hace agente es quién decide el flujo.

Definición operativa que uso en todo el curso:

Un sistema es un agente en la medida en que el LLM decide, en runtime, (a) qué acción ejecutar a continuación, (b) con qué argumentos, y (c) cuándo terminar. Si esas tres decisiones están fijadas en el código, es un pipeline con pasos LLM — por sofisticado que sea.

Con esa vara, un pipeline RAG típico (embed → retrieve → generate) llama a un LLM y usa una “tool” (el retriever), pero el orden está fijo: el LLM nunca decide si buscar, qué buscar de nuevo, ni cuándo la respuesta está completa. Es un pipeline excelente, pero pipeline. La agencia no es binaria — es un continuo de cuántas decisiones de control le cedes al modelo.

En el mundo RPA/BPM/n8n el flujo es el artefacto: se dibuja, se versiona, se audita. En un agente, el flujo es emergente: existe solo como la trayectoria que el modelo eligió en esa ejecución particular. Dos ejecuciones con el mismo input pueden producir trayectorias distintas. Esa inversión de control es el concepto central de este capítulo.

El patrón ReAct (Reasoning + Acting, Yao et al. 2022) es la implementación mínima de esa inversión: un loop donde el modelo alterna Thought (razonamiento sobre qué hacer), Action (llamada a una tool con argumentos) y Observation (el resultado vuelve al contexto), hasta que decide emitir una respuesta final.

el modelo decide que ya tiene la respuesta

Thought

Action: llamar tool

Observation: resultado

Respuesta final

El “framework” de un agente ReAct es un while de treinta líneas. Todo lo demás — LangGraph, CrewAI — es andamiaje alrededor de ese while.

2. Por qué importa en producción

Cada decisión que se le cede al modelo es una decisión que se deja de poder garantizar. En banca regulada esto no es filosofía, es arquitectura de riesgo:

3. Ejercicio práctico

Stack: Python puro + Ollama (Qwen 2.5 7B) + pgvector existente. Sin LangChain, sin LangGraph. Un script, no un servicio.

Etapa A — ReAct textual

El modelo emite Thought: ... / Action: nombre_tool / Action Input: {json}, parseado con regex, ejecutado, y devuelto como Observation: .... Este era el ReAct original, pre-tool-calling nativo — y el sufrimiento del parsing es el punto: ahí se entiende por qué existe el tool calling estructurado.

Etapa B — Tool calling nativo

El mismo loop, usando el parámetro tools de la API de chat de Ollama:

import json, time, uuid
import ollama


MODEL = "qwen2.5:7b"
MAX_ITERS = 8


SYSTEM = """Eres un asistente de análisis bancario con herramientas.
Reglas:
- Usa herramientas para obtener datos; nunca inventes cifras.
- Si la información no está disponible con tus herramientas, di explícitamente "no puedo responder esto".
- Cuando tengas la respuesta final, respóndela sin llamar más herramientas."""


def obtener_transacciones(cliente_id: str, desde: str, hasta: str) -> str:
    # Consulta parametrizada (NO SQL libre) a una tabla dummy en Postgres.
    ...


def buscar_documentos(query: str) -> str:
    # Reutiliza el retriever pgvector existente; devuelve top-3 pasajes.
    ...


import ast, operator
_OPS = {ast.Add: operator.add, ast.Sub: operator.sub,
        ast.Mult: operator.mul, ast.Div: operator.truediv,
        ast.USub: operator.neg}


def calcular(expresion: str) -> str:
    def ev(n):
        if isinstance(n, ast.Constant) and isinstance(n.value, (int, float)):
            return n.value
        if isinstance(n, ast.BinOp):
            return _OPS[type(n.op)](ev(n.left), ev(n.right))
        if isinstance(n, ast.UnaryOp):
            return _OPS[type(n.op)](ev(n.operand))
        raise ValueError("expresión no permitida")
    return str(ev(ast.parse(expresion, mode="eval").body))


TOOLS = {
    "obtener_transacciones": {"fn": obtener_transacciones, "schema": {"type": "function", "function": {
        "name": "obtener_transacciones",
        "description": "Obtiene las transacciones de un cliente en un rango de fechas.",
        "parameters": {"type": "object", "properties": {
            "cliente_id": {"type": "string"},
            "desde": {"type": "string", "description": "YYYY-MM-DD"},
            "hasta": {"type": "string", "description": "YYYY-MM-DD"}},
            "required": ["cliente_id", "desde", "hasta"]}}}},
    "buscar_documentos": {"fn": buscar_documentos, "schema": {"type": "function", "function": {
        "name": "buscar_documentos",
        "description": "Busca en la base documental interna y devuelve pasajes relevantes.",
        "parameters": {"type": "object", "properties": {"query": {"type": "string"}},
            "required": ["query"]}}}},
    "calcular": {"fn": calcular, "schema": {"type": "function", "function": {
        "name": "calcular",
        "description": "Evalúa una expresión aritmética simple (+ - * / paréntesis).",
        "parameters": {"type": "object", "properties": {"expresion": {"type": "string"}},
            "required": ["expresion"]}}}},
}


def run_agent(task: str) -> dict:
    run_id = str(uuid.uuid4())[:8]
    messages = [{"role": "system", "content": SYSTEM},
                {"role": "user", "content": task}]
    trace, llm_calls, t0 = [], 0, time.time()


    for step in range(MAX_ITERS):
        resp = ollama.chat(model=MODEL, messages=messages,
                            tools=[t["schema"] for t in TOOLS.values()],
                            options={"temperature": 0.1})
        llm_calls += 1
        msg = resp["message"]
        messages.append(dict(msg))
        tool_calls = msg.get("tool_calls") or []


        # >>> AQUÍ vive la inversión de control: el MODELO decidió
        # si el loop continúa (emitió tool_calls) o termina (no emitió).
        if not tool_calls:
            trace.append({"step": step, "type": "final", "content": msg.get("content")})
            return {"run_id": run_id, "answer": msg.get("content"),
                    "terminated": True, "llm_calls": llm_calls,
                    "seconds": time.time() - t0, "trace": trace}


        for call in tool_calls:
            name = call["function"]["name"]
            args = call["function"]["arguments"]
            if name not in TOOLS:
                result, error_type = f"ERROR: la herramienta '{name}' no existe.", "tool_inexistente"
            else:
                try:
                    result, error_type = TOOLS[name]["fn"](**args), None
                except Exception as e:
                    result, error_type = f"ERROR: {e}", "argumentos_invalidos"
            trace.append({"step": step, "type": "tool", "name": name,
                          "args": args, "result": str(result)[:500], "error": error_type})
            messages.append({"role": "tool", "content": str(result)})


    # El modelo NO decidió terminar: lo cortamos nosotros. Esto también es dato.
    return {"run_id": run_id, "answer": None, "terminated": False,
            "llm_calls": llm_calls, "seconds": time.time() - t0, "trace": trace}

Notas de implementación honestas: (a) la API del cliente Python de Ollama ha cambiado entre versiones — en versiones recientes resp.message.tool_calls es un atributo de objeto, no una key de dict; ajustar según la versión instalada. (b) los errores de tool se devuelven al modelo como Observation en vez de romper el loop — darle al agente la oportunidad de auto-corregirse es una decisión de diseño con costo (más iteraciones) y beneficio (resiliencia) medibles. (c) messages y trace completos se guardan en un JSONL por ejecución — es la evidencia de auditoría embrionaria y el insumo del harness.

El baseline determinista

El contendor del experimento es un pipeline al estilo de lo que ya se haría hoy: clasificación de intención (una sola llamada al LLM con categorías cerradas) + tres rutas fijas.

def pipeline(task: str) -> str:
    intent = clasificar_intencion(task)   # 1 llamada LLM, opciones cerradas
    if intent == "consulta_documental":
        return rag_fijo(task)             # retrieve → generate
    if intent == "consulta_transacciones":
        return ruta_transacciones(task)   # extraer params → query → resumir
    if intent == "calculo":
        return ruta_calculo(task)
    return "No puedo procesar esta solicitud."

El pipeline codifica supuestos sobre qué tareas llegarán. Las tareas compuestas (“busca la comisión vigente Y calcula cuánto pagó el cliente X”) caen entre las rutas por diseño — no es un bug del baseline, es la limitación estructural que el agente pretende resolver, y el experimento cuantifica si lo logra y a qué precio.

4. Experimento comparativo

Pregunta: ¿en qué categoría de tareas el agente justifica su sobrecosto frente al pipeline, con un modelo local de 7B?

Diseño: 25 tareas en tasks.jsonl, tres categorías — 10 simples (1 paso, dentro de las rutas del pipeline), 10 compuestas (2–3 pasos, requieren combinar tools), y 5 trampa (irresolubles con las tools disponibles; la respuesta correcta es rehusar). Cada tarea lleva verificación automática (contiene / numerico con tolerancia / rechazo).

Métricas por sistema y por categoría: tasa de éxito, llamadas LLM promedio, latencia p50/p95, tasa de no-terminación (hit MAX_ITERS), tool calls malformados o a tools inexistentes, y en trampa: tasa de alucinación.

Hipótesis registradas antes de correr:

Corrida de control: las 10 compuestas contra Qwen 2.5 14B local y Claude Sonnet vía API, para separar “límite de la arquitectura” de “límite del modelo”.

5. Resultados reales

210 ejecuciones corridas (150 de la comparación principal + 60 de la corrida de control). Código, datos sintéticos y trazas completas — repositorio público, enlace próximamente.

Análisis completo, con tabla de resultados, contraste hipótesis-vs-realidad, un trace real citado y los cuatro modos de falla observados: Solté el control de flujo a un LLM de 7B. Esto es lo que midió.

Comparación principal (pipeline vs. agente Etapa B, Qwen 2.5 7B, 25 tareas × 3 repeticiones):

CategoríaPipelineAgente Etapa B
Simple90%70%
Compuesta0%0%
Trampa80%100%

Corrida de control (solo compuestas, mismo agente y tools, cuatro modelos):

SistemaTasa de éxito en compuestas
Pipeline determinista0%
Qwen 2.5 7B (agente)0%
Qwen 2.5 14B (agente)16.7%
Claude Sonnet (agente)100%

Contra las hipótesis

6. Preguntas de cierre

Respuestas completas, con citas de datos concretos, en el repositorio del proyecto (enlace próximamente). Resumen:

  1. ¿Dónde vive la decisión de continuar/terminar? agente_etapa_b.py:61 (if not tool_calls:). La toma el modelo en cada llamada — y en las 210 ejecuciones, 0 llegaron a MAX_ITERS, incluso cuando “decidió terminar” prematuramente (rendirse sin intentar una tool disponible). Auditoría: terminated: true no distingue “resolví la tarea” de “me rendí”.
  2. ¿Es un agente un RAG con orden fijo? No — y los datos lo confirman: convertirlo en agente no habría ayudado en tareas simples (el pipeline de este experimento sacó 90% vs. 70% del agente equivalente, mismo modelo).
  3. ¿Qué se gana/pierde forzando el orden? Es literalmente pipeline vs. agente en este experimento: se gana fiabilidad en simples y trampa (90%/80% vs. 70%/100%… el agente igual ganó en trampa), se pierde adaptabilidad — y el pipeline forzado no eliminó la alucinación, la trasladó a un bug de código silencioso (el año hardcodeado).
  4. ¿Qué le falla más al 7B? Ni elegir tool ni formatear argumentos (0 errores en sus 52 llamadas reales, las tres categorías) — falla en saber cuándo detenerse: 0/30 veces intentó buscar_documentos en compuestas antes de rendirse o alucinar.
  5. ¿Por qué temperature=0.1 no es determinismo? Porque sigue siendo muestreo, no greedy puro — y lo confirma un dato real: la misma tarea (compuesta-06), mismo modelo (14B), misma temperatura, dio 2 respuestas razonables y 1 respuesta degenerada ("sourceMapping" repetido) en 3 repeticiones idénticas.
  6. ¿Alcanza el JSONL para un regulador? Parcialmente — tiene run_id, model, mensajes y trace completo, pero falta timestamp firmado dentro del JSON (hoy solo hay fecha de archivo), digest inmutable del modelo, e identidad de quién disparó la ejecución. Queda abierto para el Capítulo 8, como estaba previsto.