#La pieza más pequeña que ya enseña algo
Un bucle agéntico repite cinco pasos: prepara mensajes, obtiene una propuesta, valida su forma, ejecuta las herramientas autorizadas y añade las observaciones. Se detiene por una condición explícita. No necesita concurrencia ni una base de datos para empezar; añadirlas ahora ocultaría errores de protocolo detrás de errores de infraestructura.
Vamos a investigar un pedido. Un modelo simulado pedirá su estado y después elaborará una respuesta. Simulado significa que una función devuelve decisiones previsibles: no realiza inferencia ni demuestra calidad de un modelo real. Sí permite demostrar que el programa conecta correctamente decisiones y efectos.
Usaremos un contrato interno sencillo: una decisión de tipo tool contiene id, name y arguments; una decisión final contiene content. Es un formato definido por nosotros, no la API de ningún proveedor. Un adaptador traducirá después su protocolo a este.
#Construir el programa en tres piezas
Los siguientes bloques forman un único programa. Cópialos en orden en primer_agente.py; cada uno añade una responsabilidad. Al final lo ejecutarás con python3 primer_agente.py. No necesitas paquetes ni credenciales.
#1 · Un registro conecta nombres con funciones
En Python una función puede guardarse como valor de un diccionario. HERRAMIENTAS relaciona un nombre público con el ejecutor que controlamos. El modelo no decide qué importar ni ejecuta texto mediante eval; solo propone un nombre entre los que permitimos.
La herramienta comprueba forma e identidad del pedido antes de consultar. Un identificador ausente produce un error explícito, que el bucle convertirá después en una observación.
import json
PEDIDOS = {"p-17": {"estado": "enviado", "transportista": "Norte"}}
def consultar_pedido(arguments):
if not isinstance(arguments, dict) or set(arguments) != {"pedido_id"}:
raise ValueError("Se requiere únicamente pedido_id")
pedido_id = arguments["pedido_id"]
if not isinstance(pedido_id, str):
raise ValueError("pedido_id debe ser texto")
if pedido_id not in PEDIDOS:
raise ValueError("Pedido no encontrado")
return PEDIDOS[pedido_id]
HERRAMIENTAS = {"consultar_pedido": consultar_pedido}
#2 · La propuesta cambia al recibir una observación
El simulador primero pide consultar_pedido y asigna c-1 a la llamada. Cuando el último mensaje ya es una observación, usa su resultado para responder. La rama de error evita afirmar que un pedido se comprobó cuando la consulta falló.
def modelo_simulado(messages):
if messages[-1]["role"] == "user":
return {"type": "tool", "id": "c-1", "name": "consultar_pedido",
"arguments": {"pedido_id": "p-17"}}
observation = messages[-1]
if not observation["ok"]:
return {"type": "final", "content": "No pude comprobar el pedido."}
estado = observation["result"]["estado"]
return {"type": "final", "content": f"El pedido p-17 está {estado}."}
#3 · El harness conserva el protocolo y sus salidas
El conjunto ids permite detectar identidades repetidas. El identificador enlaza propuesta y resultado; una llamada nueva no debe reutilizarlo. Más adelante añadiremos una identidad de operación para deduplicar efectos entre reintentos, que resuelve otro problema.
Cada iteración guarda primero la propuesta, ejecuta la función seleccionada y añade una observación con el mismo call_id. **observation incorpora las claves de otro diccionario: mantiene role y call_id comunes y añade ok/result u ok/error.
Solo capturamos errores esperables de petición o permisos. Capturar cualquier Exception y convertirla en un error normal ocultaría fallos de programación. El límite de iteraciones se aplica en código: escribirlo en el prompt no impediría una quinta vuelta.
def ejecutar(modelo, max_steps=4):
messages = [{"role": "user", "content": "Comprueba el pedido p-17"}]
ids = set()
for step in range(max_steps):
decision = modelo(messages)
if not isinstance(decision, dict):
raise ValueError("El modelo debe devolver un objeto")
if decision.get("type") == "final":
content = decision.get("content")
if not isinstance(content, str) or not content.strip():
raise ValueError("Entrega vacía")
messages.append({"role": "assistant", "content": content})
return {"status": "candidate", "messages": messages}
if decision.get("type") != "tool":
raise ValueError("Tipo de decisión desconocido")
call_id = decision.get("id")
if not isinstance(call_id, str) or not call_id or call_id in ids:
raise ValueError("Identidad de llamada ausente o repetida")
ids.add(call_id)
messages.append({"role": "assistant", "call": decision})
try:
name = decision.get("name")
if not isinstance(name, str) or name not in HERRAMIENTAS:
raise ValueError("Herramienta desconocida")
result = HERRAMIENTAS[name](decision.get("arguments"))
observation = {"ok": True, "result": result}
except (ValueError, PermissionError) as error:
observation = {"ok": False, "error": str(error)}
messages.append({"role": "tool", "call_id": call_id, **observation})
return {"status": "budget_exhausted", "messages": messages}
#4 · Ejecutar y comprobar la trayectoria
El resultado debe contener cuatro mensajes: petición, llamada, observación y propuesta final. candidate significa que el modelo propone una entrega; la aplicación todavía debe verificar sus requisitos. En una tarea de código esa comprobación se haría sobre la versión concreta del artefacto.
resultado = ejecutar(modelo_simulado)
assert resultado["status"] == "candidate"
assert resultado["messages"][2]["call_id"] == "c-1"
print(json.dumps(resultado, indent=2, ensure_ascii=False))
En este ejemplo puedes comparar el estado de PEDIDOS con la respuesta: la observación dice enviado y el texto final conserva ese dato. Lo que se comprueba es la conexión entre acción y observación, no una capacidad de razonamiento del simulador.
#Cómo observar errores sin engañarte
Cambia en el simulador p-17 por p-99. Debe aparecer una observación con ok: false; no debe inventarse un pedido. Después cambia el nombre de la herramienta. El registro debe rechazarla antes de ejecutar nada.
Prueba también un modelo que nunca termina:
def insistente(messages):
return {"type": "tool", "id": f"c-{len(messages)}",
"name": "consultar_pedido", "arguments": {"pedido_id": "p-17"}}
assert ejecutar(insistente, max_steps=3)["status"] == "budget_exhausted"
Esta prueba comprueba una propiedad relevante: incluso un modelo que sigue pidiendo acciones tiene un final controlado. No comprueba que tres pasos sean un presupuesto adecuado; esa decisión depende de tareas representativas.
#Varias herramientas en una respuesta
Algunos proveedores permiten pedir varias herramientas a la vez. El harness debe guardar la propuesta completa y asociar cada resultado con su identificador. Solo puede ejecutar en paralelo llamadas independientes: consultar dos documentos puede serlo; escribir un archivo y probar esa escritura tiene una dependencia.
Si hay tres llamadas y la segunda falla, no borres la primera ni inventes un resultado para la tercera. Registra cada estado. Si se reinicia el proceso, esa distinción será necesaria para saber qué repetir. El contrato serial de este ejemplo evita inicialmente esa complejidad, pero no cambia el principio.
#Conectar un modelo sin rehacer el sistema
Un adaptador es una función que transforma un contrato en otro. Recibe nuestros mensajes, construye la petición al proveedor y convierte su respuesta en tool o final. La lógica de permisos y persistencia permanece en el harness.
No basta con hacer json.loads() y confiar en el contenido: JSON válido puede pedir una herramienta inexistente o incluir un argumento peligroso. En el miniproyecto encontrarás un adaptador HTTP opcional y la validación que lo rodea. Primero necesitamos definir mejor qué representa una herramienta.
#Referencias
- Funciones y estructuras de datos, tutorial oficial de Python: diccionarios y conjuntos utilizados en el ejemplo.
- Tools, MCP 2025-11-25: referencia de contratos y resultados; nuestro formato interno simplificado no implementa MCP.