Cómo implementar un loop de agente efectivo para LLM en producción
El loop de agente que sí funciona en producción
Tiempo estimado de lectura: 5 min
- Contrato y salida clara: fuerza una condición de salida semántica explícita (herramienta
finalizar/AgentFinishconrespuesta_final). - Validación antes de actuar: validar payloads con Pydantic y devolver errores estructurados como observaciones.
- Errores como contexto: convertir fallos de herramientas en observaciones legibles (p. ej.
ERROR_TOOL: nombre — detalle). - Límites y circuit breakers: limitar iteraciones con
MAX_STEPSy cortar si el mismo error aparece repetidamente.
¿Quieres que tu agente deje de repetir el mismo error a las 3 a. m. y empiece a resolver problemas reales? El loop de agente que sí funciona en producción es exactamente eso: una celda de contención determinista alrededor de un motor estocástico.
En las primeras líneas: El loop de agente que sí funciona en producción obliga al agente a operar con contratos claros (plan), validar antes de actuar (act), convertir fallos en contexto (observe) y respetar límites duros (reflect). Si no aplicas estas reglas, el agente acabará en bucles infinitos, consumiendo tokens y creando deuda técnica.
Resumen rápido (lectores con prisa)
Qué es: un patrón de control que encierra un LLM estocástico en un loop determinista con contratos claros y límites.
Cuándo usarlo: flujos síncronos y de corta duración (consultas DB, enriquecimiento, generación de artefactos pequeños).
Por qué importa: evita bucles infinitos, reduce deuda técnica y mejora la capacidad de recuperación ante errores de herramienta.
Cómo funciona: usar una herramienta de cierre (finalizar), validar entradas con Pydantic, registrar errores como observaciones y aplicar límites de iteración.
El loop de agente que sí funciona en producción: diseño y principios
Los LLMs no tienen intención ni memoria fiable. Son modelos probabilísticos que requieren disciplina del lado del ingeniero. Esto significa aplicar cuatro reglas innegociables:
1. Forzar una condición de salida semántica explícita
Define la herramienta de cierre del flujo. Llama a la función finalizar o AgentFinish y exige que incluya respuesta_final. Ese es el interruptor lógico que separa “sigo pensando” de “he terminado”.
2. Validar todos los argumentos antes de ejecutar herramientas
Valida con Pydantic cada payload que el modelo devuelve. Si no pasa validación, no ejecutes nada: devuelve el error al agente como observación estructurada.
3. Inyectar errores técnicos al historial como observaciones
Si una herramienta falla (timeout, DB deadlock, error de tipo), captura la excepción y agrega al historial un mensaje tipo: “ERROR_TOOL: nombre — detalle”. No ocultes fallos: convertirlos en contexto mejora la capacidad del agente para corregir su siguiente plan.
4. Imponer límites duros de iteración y circuit breakers semánticos
Finalmente, limita las iteraciones con MAX_STEPS. Si el agente alcanza ese límite, devuelve un error controlado y registra el incidente. Implementa circuit breakers si el mismo error aparece repetidamente.
Ejemplo práctico en Python
Aquí tienes un esqueleto funcional, pensado para integrarse con cualquier cliente LLM que soporte tool-calling. Incluye validación Pydantic y la inyección de errores al historial.
import json
from pydantic import BaseModel, ValidationError
MAX_STEPS = 5
# Ejemplo de validator para una tool
class CreateTicketArgs(BaseModel):
title: str
priority: str # 'low'|'medium'|'high'
# Simulación de herramientas
def create_ticket(title: str, priority: str):
# lógica real aquí (DB, API)
return {"ticket_id": "TCK-123", "title": title, "priority": priority}
TOOLS = {
"create_ticket": (create_ticket, CreateTicketArgs),
# "finalizar" no necesita validator complejo; solo espera respuesta_final
}
def run_agent(user_prompt: str, llm_client) -> str:
messages = [
{"role": "system", "content":
"Eres un agente. Devuelve siempre JSON con 'tool_call' o 'final_text'. "
"Usa la herramienta 'finalizar' con {'respuesta_final': '...'} para terminar."},
{"role": "user", "content": user_prompt}
]
seen_errors = {}
for step in range(MAX_STEPS):
response = llm_client.chat(messages=messages, tools=get_tool_schemas(TOOLS))
# respuesta textual sin tool_call
if response.finish_reason == "stop" and not response.tool_calls:
return response.content
for tc in response.tool_calls:
name = tc.name
args_raw = json.loads(tc.arguments)
if name == "finalizar":
return args_raw.get("respuesta_final", "")
tool_fn, validator = TOOLS[name]
# VALIDACIÓN (Act)
try:
args = validator(**args_raw)
except ValidationError as ve:
observation = f"VALIDATION_ERROR: {ve.json()}"
messages.append({"role": "assistant", "content": response.content})
messages.append({"role": "tool", "tool_call_id": tc.id, "content": observation})
# registro para circuit breaker
seen_errors.setdefault(str(ve), 0)
seen_errors[str(ve)] += 1
if seen_errors[str(ve)] >= 2:
return f"Interrumpido: error recurrente de validación: {ve}"
continue
# EJECUCIÓN segura (Observe)
try:
result = tool_fn(**args.dict())
observation = f"RESULT: {json.dumps(result)}"
except Exception as e:
observation = f"ERROR_TOOL: {name} — {str(e)}"
seen_errors.setdefault(str(e), 0)
seen_errors[str(e)] += 1
if seen_errors[str(e)] >= 2:
return f"Interrumpido: error recurrente en herramienta: {e}"
messages.append({"role": "assistant", "content": response.content})
messages.append({"role": "tool", "tool_call_id": tc.id, "content": observation})
return "Error: el agente superó el límite de pasos permitidos."
Operaciones que debes instrumentar desde el día uno
- Logging estructurado por paso: step, tool, args, resultado, tokens consumidos.
- Conteo de tokens del historial; si supera el 75–80% de la ventana del modelo, ejecuta resumen (ver técnicas de resumen).
- Circuit breaker semántico: si el mismo error aparece dos veces, corta y alerta.
- Métricas SLIs: tiempo por petición, reintentos por herramienta, tasa de finalización exitosa.
Cuándo no usar este loop y optar por orquestación
Usa este loop en procesos síncronos y de corta duración: consultas DB, enriquecimiento de datos, generación de artefactos pequeños. Si tu flujo dura horas/días, incluye pasos humanos o requiere durabilidad de estado, cambia a un orquestador (n8n, LangGraph) que persista estado y permita reprogramación.
Cierre práctico
El loop de agente que sí funciona en producción no es una trick de prompts: es ingeniería. Forzar salida semántica (finalizar), validar antes de actuar (Pydantic), convertir fallos en contexto y aplicar límites duros son las piezas que convierten un prototipo ruidoso en un agente fiable. Implementa esto, instrumenta y repite: la estabilidad viene de la disciplina, no de la magia.
Para recursos experimentales y proyectos relacionados con automatización y agentes, considera explorar los trabajos y laboratorios prácticos en Dominicode Labs. Es una referencia contextual para implementar pipelines y pruebas prácticas en entornos de ingeniería.
FAQ
- ¿Cuál es la función de la herramienta
finalizar? - ¿Por qué usar Pydantic para validar payloads?
- ¿Qué hacer si una herramienta externa falla repetidamente?
- ¿Cómo definir
MAX_STEPS? - ¿Cuándo prefiero un orquestador en lugar de este loop?
- ¿Qué métricas debo recolectar desde el día uno?
¿Cuál es la función de la herramienta finalizar?
Es el interruptor semántico que indica que el agente ha terminado su trabajo. Debe devolver un objeto que incluya respuesta_final, que el loop interpreta como salida definitiva.
¿Por qué usar Pydantic para validar payloads?
Porque ofrece validación estructurada y errores claros. Validar evita ejecuciones inseguras y permite convertir fallos en observaciones reutilizables por el agente.
¿Qué hacer si una herramienta externa falla repetidamente?
Registrar el fallo como ERROR_TOOL: nombre — detalle, aplicar un circuit breaker y, si el error se repite, interrumpir el loop con un error controlado para evitar consumo indefinido de recursos.
¿Cómo definir MAX_STEPS?
Depende del dominio y del coste por iteración. Empieza con un valor conservador (p. ej. 5) y monitoriza reintentos y tiempo por petición para ajustarlo.
¿Cuándo prefiero un orquestador en lugar de este loop?
Si el flujo requiere durabilidad, pasos humanos o puede durar horas/días, usa un orquestador como n8n o LangGraph para persistir estado y reintentar tareas.
¿Qué métricas debo recolectar desde el día uno?
Tiempo por petición, reintentos por herramienta, tasa de finalización exitosa, conteo de tokens y registros estructurados por paso (tool, args, resultado).
