Error Recovery en AI Agents: El Framework que Transforma el 40% de Fallos en Aprendizaje
Error Recovery en AI Agents: Framework paso a paso para detectar, clasificar y recuperarse de errores. Mantén 95% de correctness con validación estructurada.
El 95% de los AI Agents Se Rompen en el Primer Error Real
Tu agent ejecuta cien tareas perfectamente.
El centésimo primer request tiene un formato inesperado.
Tu agent crashea. Se detiene. Necesita intervención manual.
*El problema real no es que los agents fallen. Es que no tienen sistema para levantarse.*
La mayoría de developers construye AI agents como si fueran scripts deterministas. Cuando algo falla, saltan excepciones, se imprime un error en consola, y el sistema queda en estado indefinido.
Esto no es cómo se construyen sistemas resilientes. Es cómo se construyen sistemas que fallan elegantemente hacia abajo.
Los datos cuentan otra historia: cuando implementas un framework de validación estructurada con checkpoints humanos, los AI agents mantienen 95% de code correctness mientras reducen deuda técnica un 40%.
No por magia. Por arquitectura.
Por Qué los Try-Catch No Son Suficiente
❌ LO QUE HACE LA MAYORÍA:
```python
try:
result = agent.execute(task)
except Exception as e:
print(f"Error: {e}")
return None
```
✅ LO QUE DEBERÍAS HACER:
```python
class ErrorRecoveryAgent:
def __init__(self):
self.error_classifier = ErrorClassifier()
self.recovery_strategies = RecoveryStrategies()
self.checkpoint_validator = CheckpointValidator()
def execute_with_recovery(self, task):
checkpoint = self.checkpoint_validator.create_checkpoint()
try:
result = self.execute_task(task)
self.checkpoint_validator.validate(checkpoint, result)
return result
except TransientError as e:
return self.recovery_strategies.retry(e, backoff=True)
except LogicError as e:
return self.recovery_strategies.alternative_path(task, e)
except FatalError as e:
return self.recovery_strategies.escalate_to_human(e)
```
La diferencia no es solo código. Es mentalidad.
Basic exception handling espera que fallen cosas. Un framework de error recovery espera que fallen cosas, pero tiene un plan sistemático para cada tipo de fallo.
El Sistema de Clasificación que Cambia Todo
No todos los errores son iguales. Tratar todos los errores con el mismo retry logic es como usar el mismo antibiótico para un resfriado y una infección bacteriana.
Necesitas un clasificador.
Step 1: Implementa el Error Classifier
```python
from enum import Enum
from dataclasses import dataclass
from typing import Optional
import re
class ErrorType(Enum):
TRANSIENT = "transient" # Timeout, conexión, rate limit
LOGIC = "logic" # Malformed input, invalid state
RESOURCE = "resource" # Out of memory, disk full
FATAL = "fatal" # Corruption, security breach
@dataclass
class ClassifiedError:
error_type: ErrorType
original_exception: Exception
retry_count: int = 0
context: dict = None
class ErrorClassifier:
TRANSIENT_PATTERNS = [
r"timeout",
r"connection.*reset",
r"429",
r"rate.*limit",
r"temporarily.*unavailable"
]
def classify(self, exception: Exception, context: dict) -> ClassifiedError:
error_msg = str(exception).lower()
for pattern in self.TRANSIENT_PATTERNS:
if re.search(pattern, error_msg):
return ClassifiedError(
error_type=ErrorType.TRANSIENT,
original_exception=exception,
context=context
)
if self.is_logic_error(exception, context):
return ClassifiedError(
error_type=ErrorType.LOGIC,
original_exception=exception,
context=context
)
if self.is_resource_error(exception):
return ClassifiedError(
error_type=ErrorType.RESOURCE,
original_exception=exception,
context=context
)
return ClassifiedError(
error_type=ErrorType.FATAL,
original_exception=exception,
context=context
)
```
Este classifier transforma errores genéricos en categorías accionables. Un timeout no se maneja igual que un input malformado. Un problema de memoria no se maneja igual que una violación de seguridad.
El Decorator que Valida Todo
El checkpoint validator es donde la magia sucede. Antes de cada acción crítica, guardas estado. Después de ejecutar, verificas que el estado resultado es válido.
```python
import functools
import logging
from datetime import datetime
class CheckpointValidator:
def __init__(self, agent):
self.agent = agent
self.checkpoints = []
self.logger = logging.getLogger("checkpoint_validator")
def checkpoint(self, operation_name: str):
def decorator(func):
@functools.wraps(func)
def wrapper(args, *kwargs):
checkpoint_id = f"{operation_name}_{datetime.utcnow().isoformat()}"
pre_state = self.capture_state()
self.logger.info(f"Checkpoint {checkpoint_id} - Pre-state captured")
try:
result = func(args, *kwargs)
post_state = self.capture_state()
if self.validate_transition(pre_state, post_state):
self.checkpoints.append({
"id": checkpoint_id,
"operation": operation_name,
"pre": pre_state,
"post": post_state,
"status": "success"
})
return result
else:
self.logger.warning(f"Checkpoint {checkpoint_id} - Invalid transition")
raise InvalidStateTransition(pre_state, post_state)
except Exception as e:
self.logger.error(f"Checkpoint {checkpoint_id} - Failed: {e}")
self.rollback(checkpoint_id, pre_state)
raise
return wrapper
return decorator
def capture_state(self) -> dict:
return {
"memory": self.agent.memory.snapshot(),
"tool_states": self.agent.tool_manager.get_states(),
"timestamp": datetime.utcnow().isoformat()
}
def validate_transition(self, pre: dict, post: dict) -> bool:
Lógica de validación específica para tu agent
return True
def rollback(self, checkpoint_id: str, state: dict):
self.agent.memory.restore(state["memory"])
self.agent.tool_manager.restore_states(state["tool_states"])
self.logger.info(f"Rolled back to checkpoint {checkpoint_id}")
```
Este decorator envuelve cada operación crítica de tu agent. Si algo falla, tienes el estado previo capturado. Si algo succeeds, validas que la transición fue legítima.
La Estrategia de Recovery por Tipo de Error
Transient Errors: Retry con Exponential Backoff
```python
class TransientRecovery:
def __init__(self, max_retries: int = 3, base_delay: float = 1.0):
self.max_retries = max_retries
self.base_delay = base_delay
def retry(self, func, args, *kwargs):
last_exception = None
for attempt in range(self.max_retries):
try:
return func(args, *kwargs)
except TransientError as e:
last_exception = e
delay = self.base_delay (2 * attempt)
jitter = random.uniform(0, delay * 0.1)
logging.info(f"Retry attempt {attempt + 1} after {delay + jitter}s")
time.sleep(delay + jitter)
raise MaxRetriesExceeded(last_exception)
```
Los errores transient son temporales. Timeout de API, rate limits, problemas de red. Retry con exponential backoff evita saturar el sistema mientras le das tiempo a恢复正常.
Logic Errors: Alternative Path Execution
```python
class LogicRecovery:
def __init__(self, agent):
self.agent = agent
def execute_alternative(self, original_task: Task, error: LogicError):
Analiza quéwent wrong
analysis = self.analyze_logic_error(error)
Genera paths alternativos
alternatives = self.agent.generate_alternative_paths(
original_task,
avoiding=analysis["problematic_patterns"]
)
Ejecuta el primer alternative viable
for alternative in alternatives:
try:
return self.agent.execute(alternative)
except LogicError:
continue
Si ningún alternative funciona, escalate
return self.escalate_to_human(original_task, error)
```
Los errores de lógica significan que tu agent encontró un caso que no sabe manejar. En lugar de fallar, genera alternativas y prueba hasta que una funcione.
Fatal Errors: Human Escalation
```python
class FatalRecovery:
def __init__(self, escalation_handler):
self.escalation_handler = escalation_handler
def escalate(self, task: Task, error: FatalError, checkpoint_state: dict):
escalation = {
"task": task.to_dict(),
"error": str(error),
"checkpoint_state": checkpoint_state,
"agent_attempts": self.get_attempt_history(task.id),
"recommended_action": self.analyze(error)
}
self.escalation_handler.send(escalation)
self.logger.critical(f"Escalated task {task.id} to human review")
return EscalationResult(
task_id=task.id,
status="awaiting_human_decision"
)
```
Los errores fatales no tienen recovery automático. Corruption de datos, security breaches, estados inconsistentes. Aquí es donde entra el human-in-the-loop framework.
El Ciclo de Aprendizaje Continuo
Lo que separa un agent decente de uno excelente es que aprende de sus errores.
```python
class ErrorLearningLoop:
def __init__(self, storage):
self.storage = storage
def log_recovery(self, error: ClassifiedError, recovery: RecoveryResult):
self.storage.insert("error_recovery_log", {
"error_type": error.error_type.value,
"error_pattern": self.extract_pattern(error),
"recovery_strategy": recovery.strategy_used,
"success": recovery.succeeded,
"context": error.context
})
def improve_classifier(self):
Analiza historical errors para mejorar patterns
historical = self.storage.query("""
SELECT error_pattern, error_type, COUNT(*) as frequency
FROM error_recovery_log
WHERE success = true
GROUP BY error_pattern, error_type
""")
Actualiza classifier con patterns descubiertos
for row in historical:
self.update_classifier_pattern(row["error_pattern"], row["error_type"])
```
Cada error recuperado es un dato. El classifier aprende de patterns exitosos. El recovery system mejora con cada iteración.
La Métrica que Demuestra el Valor
Los 40% de reducción en deuda técnica no son accidentales.
Cada error recuperado con un framework estructurado genera documentación. El siguiente error similar se resuelve más rápido. Las causas raíz se identifican sistemáticamente.
Los fixes ad-hoc crean dependencias ocultas. El agent sobrevive hoy, pero mañana tienes tres edge cases que dependen de ese fix improvisado.
Los fixes estructurados construyen conocimiento. El agent aprende a reconocer patterns, anticipar problemas, y resolverlos antes de que escalen.
Esta es la diferencia entre un agent que funciona y un agent que mejora con el tiempo.
Implementa Esto Hoy
Step 1: Instrumenta tu execution loop con checkpoints antes de cada acción crítica.
Step 2: Implementa el ErrorClassifier con las categorías (transient, logic, resource, fatal) y patterns específicos para tu dominio.
Step 3: Construye el RecoveryStrategy dispatcher que routing errores al handler correcto.
Step 4: Añade el checkpoint decorator a todas las operaciones que modifican estado.
Step 5: Configura el escalation handler para errores fatales con contexto completo.
Step 6: Activa el learning loop para que el sistema mejore con cada recuperación.
Lo Que Nadie Te Cuenta
La mayoría de tutorials de AI agents terminan en "y aquí tu agent ejecuta la tarea".
Nadie te cuenta qué pasa cuando falla.
La diferencia entre un agent en development y un agent en producción no es el prompt. No es el modelo. No es la tool selection.
Es el sistema de error recovery.
Los agents que sobreviven en producción son los que tienen un plan para cuando las cosas fallan. Los que crashean son los que asumen que las cosas no van a fallar.
Construye para el 5% de errores. El 95% de correctness se encarga solo.
El 40% de reducción en deuda técnica es consecuencia de ese enfoque.
El verdadero diferenciador entre agents mediocres y agents resilientes no es la capacidad de ejecutar bien. Es la capacidad de recuperarse cuando ejecutan mal.
Empieza con el checkpoint decorator. Añade el classifier. Implementa el retry logic. Cada capa añade resiliencia.
Tu agent no necesita ser perfecto. Necesita saber qué hacer cuando no lo es.
Lee el artículo completo en brianmenagomez.com
Más sobre mis servicios en brianmenagomez.com
Herramientas: Conversor IAE CNAE · Gestorias cerca de ti · Calculadora IRPF