Un agente de IA no es un chatbot con más contexto. Es un sistema autónomo que recibe un objetivo, planifica los pasos necesarios, ejecuta acciones a través de herramientas externas y se adapta cuando algo falla. La diferencia es operativa: un chatbot responde preguntas, un agente completa tareas.
En Cloudstudio llevamos más de un año diseñando y desplegando agentes de IA para clientes que necesitan automatizar procesos complejos. En este artículo compartimos la arquitectura que utilizamos, los patrones que funcionan y los errores que hemos aprendido a evitar.
El bucle fundamental del agente
Cada agente sigue un ciclo: observar el estado actual, decidir la siguiente acción, ejecutarla y evaluar el resultado. Este bucle se repite hasta que se cumple el objetivo o se escala a un humano. La clave es que cada paso es una llamada al modelo con contexto actualizado: el agente no memoriza un plan fijo, sino que reevalúa en cada iteración.
Utilizamos Claude como el cerebro de nuestros agentes. Su capacidad nativa de uso de herramientas permite definir herramientas como funciones que el modelo puede invocar directamente: consultar bases de datos, enviar correos electrónicos, actualizar CRMs, generar documentos. El modelo decide cuándo y cómo usar cada herramienta basándose en el contexto de la tarea.
Aquí está el bucle central del agente que usamos en producción. Es engañosamente simple: la complejidad reside en las herramientas y en el prompt del sistema:
import anthropic
import json
import time
from dataclasses import dataclass, field
from typing import Any
@dataclass
class AgentState:
goal: str
messages: list = field(default_factory=list)
steps_taken: int = 0
total_tokens: int = 0
total_cost: float = 0.0
max_steps: int = 25
max_cost: float = 5.00 # USD budget limit
class Agent:
def __init__(self, system_prompt: str, tools: list, tool_handlers: dict):
self.client = anthropic.Anthropic()
self.system_prompt = system_prompt
self.tools = tools
self.tool_handlers = tool_handlers
def run(self, goal: str) -> AgentState:
state = AgentState(goal=goal)
state.messages = [{"role": "user", "content": goal}]
while state.steps_taken < state.max_steps:
# Check budget before each step
if state.total_cost >= state.max_cost:
state.messages.append({
"role": "user",
"content": "Budget limit reached. Summarize what you have accomplished so far."
})
response = self.client.messages.create(
model="claude-sonnet-4-6-6",
max_tokens=4096,
system=self.system_prompt,
tools=self.tools,
messages=state.messages,
)
# Track usage
state.total_tokens += response.usage.input_tokens + response.usage.output_tokens
state.total_cost += self._calculate_cost(response.usage)
state.steps_taken += 1
# If Claude is done (no more tool calls), return
if response.stop_reason == "end_turn":
state.messages.append({"role": "assistant", "content": response.content})
return state
# Process tool calls
if response.stop_reason == "tool_use":
state.messages.append({"role": "assistant", "content": response.content})
tool_results = self._execute_tools(response.content)
state.messages.append({"role": "user", "content": tool_results})
return state # Max steps reached
def _execute_tools(self, content) -> list:
results = []
for block in content:
if block.type == "tool_use":
result = self._safe_execute(block.name, block.input)
results.append({
"type": "tool_result",
"tool_use_id": block.id,
"content": json.dumps(result),
})
return results
def _safe_execute(self, tool_name: str, params: dict) -> dict:
handler = self.tool_handlers.get(tool_name)
if not handler:
return {"error": f"Unknown tool: {tool_name}"}
try:
return handler(**params)
except Exception as e:
return {"error": f"{type(e).__name__}: {str(e)}"}
def _calculate_cost(self, usage) -> float:
# Sonnet pricing per million tokens
input_cost = (usage.input_tokens / 1_000_000) * 3.0
output_cost = (usage.output_tokens / 1_000_000) * 15.0
return input_cost + output_cost
Los límites max_steps y max_cost no son opcionales: son la diferencia entre un agente controlado y un proceso desbocado que agota tu presupuesto de API a las 3 AM.
Alternativas de frameworks: Si prefieres no construir un bucle de agente personalizado desde cero, ahora existen opciones maduras. El Anthropic Agent SDK proporciona un framework de Python listo para producción para construir agentes impulsados por Claude con gestión de herramientas integrada, guardrails y orquestación multi-agente. Para equipos de TypeScript, Vercel AI SDK 6 ofrece un framework agnóstico al proveedor con aprobación de herramientas human-in-the-loop, soporte para MCP y un AI Gateway que funciona con Claude, GPT-5.4 y Gemini 3.1. Y para conectar agentes a herramientas externas y fuentes de datos, el Model Context Protocol (MCP) —ahora un estándar de la industria bajo la Linux Foundation con el respaldo de Anthropic, OpenAI, Google, Microsoft y AWS— proporciona una interfaz universal que elimina la necesidad de escribir integraciones personalizadas para cada herramienta. Seguimos prefiriendo los bucles personalizados para un control máximo en escenarios de producción complejos, pero estos frameworks reducen significativamente el tiempo de puesta en producción para patrones de agentes estándar.
Definiciones de herramientas: diseño para la confiabilidad
La calidad de sus herramientas determina la calidad de su agente. Cada herramienta debe tener una descripción precisa, un esquema de entrada estricto y un comportamiento de error predecible. Aquí hay un conjunto de herramientas de ejemplo para un agente de soporte al cliente:
SUPPORT_TOOLS = [
{
"name": "search_knowledge_base",
"description": "Busque en la [base de conocimientos](/services/rag-systems) interna artículos relevantes para un problema del cliente. Devuelve los 5 artículos más coincidentes con títulos y contenido. Use esto PRIMERO antes de intentar responder cualquier pregunta técnica.",
"input_schema": {
"type": "object",
"properties": {
"query": {
"type": "string",
"description": "Consulta de búsqueda en lenguaje natural que describe el problema del cliente"
},
"category": {
"type": "string",
"enum": ["billing", "technical", "account", "product"],
"description": "Categoría para restringir la búsqueda"
}
},
"required": ["query"]
}
},
{
"name": "lookup_customer",
"description": "Busque los detalles de la cuenta de un cliente por correo electrónico. Devuelve el plan de suscripción, el estado de la cuenta, los tickets recientes y el historial de facturación. Use esto para comprender el contexto del cliente.",
"input_schema": {
"type": "object",
"properties": {
"email": {
"type": "string",
"description": "Dirección de correo electrónico del cliente"
}
},
"required": ["email"]
}
},
{
"name": "create_ticket",
"description": "Cree un ticket de soporte para problemas que requieran seguimiento humano. Use esto cuando el problema no se pueda resolver automáticamente o requiera permisos elevados.",
"input_schema": {
"type": "object",
"properties": {
"customer_email": {"type": "string"},
"subject": {"type": "string", "description": "Resumen breve del problema"},
"body": {"type": "string", "description": "Descripción detallada que incluye los pasos seguidos"},
"priority": {
"type": "string",
"enum": ["low", "medium", "high", "urgent"]
},
"category": {
"type": "string",
"enum": ["billing", "technical", "account", "product"]
}
},
"required": ["customer_email", "subject", "body", "priority", "category"]
}
},
{
"name": "escalate_to_human",
"description": "Transfiera inmediatamente la conversación a un agente humano. Use esto cuando: el cliente solicite explícitamente a un humano, el problema involucre reembolsos de más de $100, o no tenga confianza en su resolución.",
"input_schema": {
"type": "object",
"properties": {
"reason": {"type": "string", "description": "Por qué es necesaria la escalación"},
"summary": {"type": "string", "description": "Resumen de la conversación hasta ahora"},
"suggested_team": {
"type": "string",
"enum": ["billing", "engineering", "account_management"]
}
},
"required": ["reason", "summary"]
}
}
]
Dos principios de diseño que hemos aprendido por las malas: primero, incluya siempre una herramienta de escalación. Un agente que no puede escalar alucinará soluciones cuando esté atascado. Segundo, escriba las descripciones de las herramientas como directivas: "Use esto PRIMERO antes de..." le da al modelo un marco de decisión claro.
Patrones que funcionan en producción
Humano en el bucle (Human-in-the-loop). Ningún agente debe operar sin un mecanismo de escalación. Definimos umbrales de confianza: si el agente no está seguro de su decisión, pausa la ejecución y notifica a un humano. Esto es crítico en procesos con impacto financiero o legal.
Recuperación de errores. Los agentes en producción encuentran errores constantemente: APIs que fallan, datos inesperados, tiempos de espera agotados. Diseñamos cada herramienta con reintentos automáticos, alternativas (fallbacks) e interruptores automáticos (circuit breakers). El agente debe ser capaz de diagnosticar el error e intentar una ruta alternativa.
Aquí está nuestro envoltorio (wrapper) de recuperación de errores que envuelve cada controlador de herramientas:
import time
import random
import logging
from functools import wraps
logger = logging.getLogger(__name__)
def resilient_tool(max_retries=3, timeout=30, fallback=None):
"""Decorador que añade lógica de reintento e interruptor automático a los controladores de herramientas."""
def decorator(func):
_failure_count = {"value": 0}
_circuit_open_until = {"value": 0}
@wraps(func)
def wrapper(**kwargs):
# Interruptor automático: si hay demasiados fallos recientes, fallar rápido
if time.time() < _circuit_open_until["value"]:
if fallback:
logger.warning(f"Circuito abierto para {func.__name__}, usando alternativa")
return fallback(**kwargs)
return {"error": "Servicio temporalmente no disponible", "retry_after": 60}
for attempt in range(max_retries):
try:
result = func(**kwargs)
_failure_count["value"] = 0 # Restablecer en caso de éxito
return result
except TimeoutError:
if attempt < max_retries - 1:
wait = (2 ** attempt) + random.uniform(0, 1)
time.sleep(wait)
else:
_failure_count["value"] += 1
if _failure_count["value"] >= 5:
_circuit_open_until["value"] = time.time() + 60
return {"error": f"Tiempo de espera agotado después de {max_retries} reintentos"}
except Exception as e:
logger.error(f"La herramienta {func.__name__} falló: {e}")
_failure_count["value"] += 1
if _failure_count["value"] >= 5:
_circuit_open_until["value"] = time.time() + 60
return {"error": str(e)}
return wrapper
return decorator
@resilient_tool(max_retries=3, timeout=10)
def lookup_customer(email: str) -> dict:
"""Buscar cliente en el CRM."""
response = crm_client.get(f"/customers?email={email}", timeout=10)
response.raise_for_status()
return response.json()
El patrón de interruptor automático es esencial. Sin él, una interrupción del servicio descendente hace que su agente gaste todo su presupuesto reintentando una herramienta que nunca tendrá éxito. Con el interruptor automático, después de 5 fallos consecutivos, la herramienta falla rápido durante 60 segundos, dándole al agente una señal de error clara con la que trabajar.
Observabilidad. Cada acción del agente se registra con marcas de tiempo, costos de tokens, duración y resultado. Sin una observabilidad completa, depurar un agente en producción es imposible. Utilizamos trazas estructuradas que permiten reconstruir cada decisión del agente paso a paso.
Configuración de monitoreo y observabilidad
No se puede operar un agente que no se puede observar. Registramos cada paso en un formato estructurado que nos permite reconstruir la cadena de decisión completa:
import time
import json
import logging
from contextlib import contextmanager
from dataclasses import dataclass, asdict
logger = logging.getLogger("agent.trace")
@dataclass
class AgentTrace:
agent_id: str
session_id: str
step: int
action: str # "llm_call", "tool_call", "tool_result", "error", "escalation"
tool_name: str | None = None
input_summary: str | None = None
output_summary: str | None = None
input_tokens: int = 0
output_tokens: int = 0
cost_usd: float = 0.0
duration_ms: int = 0
success: bool = True
error: str | None = None
class AgentTracer:
def __init__(self, agent_id: str, session_id: str):
self.agent_id = agent_id
self.session_id = session_id
self.step = 0
self.traces: list[AgentTrace] = []
@contextmanager
def trace_step(self, action: str, tool_name: str = None):
self.step += 1
trace = AgentTrace(
agent_id=self.agent_id,
session_id=self.session_id,
step=self.step,
action=action,
tool_name=tool_name,
)
start = time.monotonic()
try:
yield trace
trace.success = True
except Exception as e:
trace.success = False
trace.error = str(e)
raise
finally:
trace.duration_ms = int((time.monotonic() - start) * 1000)
self.traces.append(trace)
# Emit as structured JSON log
logger.info(json.dumps(asdict(trace)))
def summary(self) -> dict:
return {
"total_steps": self.step,
"total_tokens": sum(t.input_tokens + t.output_tokens for t in self.traces),
"total_cost": sum(t.cost_usd for t in self.traces),
"total_duration_ms": sum(t.duration_ms for t in self.traces),
"errors": [t.error for t in self.traces if not t.success],
"tools_used": [t.tool_name for t in self.traces if t.tool_name],
}
Enviamos estos registros JSON a nuestra pila de observabilidad (Datadog o similar) y creamos paneles que muestran: tasa de éxito del agente, promedio de pasos por tarea, costo por tarea, herramientas más utilizadas y frecuencia de errores por herramienta. El panel ha sido la herramienta de depuración más valiosa: cuando un agente comienza a comportarse de manera inesperada, las trazas indican exactamente dónde falló el razonamiento.
Control de costos: límites de presupuesto y seguimiento de tokens
Los agentes pueden ser costosos porque realizan múltiples llamadas a LLM por tarea. Sin controles de costos, una sola entrada patológica puede desencadenar docenas de iteraciones. Así es como aplicamos los presupuestos:
@dataclass
class BudgetConfig:
max_cost_per_session: float = 2.00 # USD
max_cost_per_step: float = 0.50 # USD
max_tokens_per_session: int = 100_000
max_steps: int = 25
alert_threshold: float = 0.75 # Alert at 75% of budget
class BudgetManager:
def __init__(self, config: BudgetConfig):
self.config = config
self.total_cost = 0.0
self.total_tokens = 0
self.steps = 0
def check_budget(self, estimated_input_tokens: int = 0) -> dict:
"""Check if we can afford another step. Returns status and remaining budget."""
if self.steps >= self.config.max_steps:
return {"allowed": False, "reason": "max_steps_reached"}
if self.total_cost >= self.config.max_cost_per_session:
return {"allowed": False, "reason": "cost_limit_reached"}
if self.total_tokens + estimated_input_tokens > self.config.max_tokens_per_session:
return {"allowed": False, "reason": "token_limit_reached"}
remaining = self.config.max_cost_per_session - self.total_cost
if remaining / self.config.max_cost_per_session < (1 - self.config.alert_threshold):
logger.warning(f"Budget alert: only ${remaining:.2f} remaining")
return {
"allowed": True,
"remaining_cost": remaining,
"remaining_steps": self.config.max_steps - self.steps,
}
def record_usage(self, input_tokens: int, output_tokens: int, model: str):
self.steps += 1
self.total_tokens += input_tokens + output_tokens
self.total_cost += self._price(input_tokens, output_tokens, model)
def _price(self, input_tokens: int, output_tokens: int, model: str) -> float:
pricing = {
"claude-haiku-4-5-5": (0.80, 4.0),
"claude-sonnet-4-6-6": (3.0, 15.0),
"claude-opus-4-6-6": (15.0, 75.0),
}
input_rate, output_rate = pricing.get(model, (3.0, 15.0))
return (input_tokens / 1_000_000 * input_rate) + (output_tokens / 1_000_000 * output_rate)
Establecemos diferentes configuraciones de presupuesto para distintos tipos de agentes. Un agente de clasificación simple recibe $0.50 por sesión. Un agente de investigación complejo que necesita buscar en múltiples fuentes recibe $5.00. Estos límites detectan a tiempo a los agentes fuera de control y ofrecen una economía unitaria predecible.
Lo que aprendimos por las malas
El mayor error es asumir que el agente siempre tomará la decisión correcta. En producción, los casos extremos son la norma. Un agente de soporte que clasifica tickets funcionará perfectamente el 95% de las veces, pero ese 5% restante puede generar respuestas incorrectas para clientes importantes.
La solución no es más ingeniería de prompts. Es diseñar el sistema para que los fallos sean detectables, reversibles y escalables. Limitar el radio de impacto de cada acción del agente. Y medir continuamente la calidad de las decisiones frente a un conjunto de evaluación.
Tres lecciones más tras ejecutar agentes en producción durante un año:
Pruebe con entradas adversarias. Los usuarios enviarán a su agente cosas que nunca imaginó: mensajes vacíos, mensajes en el idioma equivocado, capturas de pantalla cuando espera texto o instrucciones deliberadamente engañosas. Cree una suite de pruebas con más de 50 casos adversarios y ejecútela con cada cambio de prompt.
Mantenga el prompt del sistema por debajo de los 2,000 tokens. Los prompts más largos dan más información al modelo, pero también aumentan la latencia y el costo por paso. El bucle del agente multiplica este costo de 5 a 25 veces. Hemos descubierto que los prompts de sistema concisos y bien estructurados superan a los detallados.
Registre cada llamada a herramientas, no solo los errores. Cuando algo sale mal en el paso 15 de la ejecución de un agente, necesita la traza completa para entender cómo llegó allí. Registrar solo los errores le da el síntoma. Registrar cada paso le da el diagnóstico.
