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 a 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 utilizamos 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-20250514",
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 de la mañana.
Definiciones de herramientas: diseño para la confiabilidad
La calidad de tus herramientas determina la calidad de tu agente. Cada herramienta debe tener una descripción precisa, un esquema de entrada estricto y un comportamiento de error predecible. Aquí tienes un conjunto de herramientas de ejemplo para un agente de atención al cliente:
SUPPORT_TOOLS = [
{
"name": "search_knowledge_base",
"description": "Busca en la base de conocimientos interna artículos relevantes para el problema de un cliente. Devuelve los 5 artículos más coincidentes con títulos y contenido. Usa 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 limitar la búsqueda"
}
},
"required": ["query"]
}
},
{
"name": "lookup_customer",
"description": "Busca 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. Usa esto para entender 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": "Crea un ticket de soporte para problemas que requieren seguimiento humano. Usa 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": "Transfiere inmediatamente la conversación a un agente humano. Usa esto cuando: el cliente solicite explícitamente un humano, el problema involucre reembolsos de más de $100, o no estés seguro de tu 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 el momento"},
"suggested_team": {
"type": "string",
"enum": ["billing", "engineering", "account_management"]
}
},
"required": ["reason", "summary"]
}
}
]
Dos principios de diseño que hemos aprendido por las malas: primero, incluye siempre una herramienta de escalación. Un agente que no puede escalar alucinará soluciones cuando esté atascado. Segundo, escribe las descripciones de las herramientas como directivas: "Usa esto PRIMERO antes de..." le da al modelo un marco de decisión claro.
Patrones que funcionan en producción
Intervención humana. Ningún agente debe operar sin un mecanismo de escalada. 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 y disyuntores. El agente debe ser capaz de diagnosticar el error e intentar una ruta alternativa.
Aquí está nuestro envoltorio 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):
"""Decorator that adds retry logic and circuit breaking to tool handlers."""
def decorator(func):
_failure_count = {"value": 0}
_circuit_open_until = {"value": 0}
@wraps(func)
def wrapper(**kwargs):
# Circuit breaker: if too many recent failures, fail fast
if time.time() < _circuit_open_until["value"]:
if fallback:
logger.warning(f"Circuit open for {func.__name__}, using fallback")
return fallback(**kwargs)
return {"error": "Service temporarily unavailable", "retry_after": 60}
for attempt in range(max_retries):
try:
result = func(**kwargs)
_failure_count["value"] = 0 # Reset on success
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"Timeout after {max_retries} retries"}
except Exception as e:
logger.error(f"Tool {func.__name__} failed: {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:
"""Look up customer from CRM."""
response = crm_client.get(f"/customers?email={email}", timeout=10)
response.raise_for_status()
return response.json()
El patrón de disyuntor 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 disyuntor, después de 5 fallos consecutivos, la herramienta falla rápidamente 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, costes 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": "
## 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:
Prueba con entradas adversarias. Los usuarios enviarán a tu agente cosas que nunca imaginaste: mensajes vacíos, mensajes en el idioma equivocado, capturas de pantalla cuando espera texto o instrucciones deliberadamente engañosas. Crea un conjunto de pruebas de más de 50 casos adversarios y ejecútalo en cada cambio de prompt.
Mantén el prompt del sistema por debajo de los 2,000 tokens. Los prompts más largos le dan al modelo más información, pero también aumentan la latencia y el coste por paso. El bucle del agente multiplica este coste entre 5 y 25 veces. Hemos descubierto que los prompts de sistema concisos y bien estructurados superan a los prolijos.
Registra cada llamada a herramientas, no solo los errores. Cuando algo sale mal en el paso 15 de la ejecución de un agente, necesitas el rastro completo para entender cómo llegó allí. Registrar solo los errores te da el síntoma. Registrar cada paso te da el diagnóstico.
