Self-Healing en Agentes de IA: Recuperación Automática Cuando tus Scrapers Fallan en Producción

Tu agente funcionó perfecto en local. Los tests pasaron verde. Pero a las 3 AM del domingo, un deploy cambió una clase de Tailwind de bg-blue-500 a bg-blue-600 y tu automatización web murió silenciosamente, perdiendo datos críticos. La recuperación automática en agentes de IA no es un lujo: es una necesidad cuando operas scraping o RPA a escala. Vamos a construir un sistema que no solo detecte fallos, sino que los corrija sin despertarte.

El problema real: Clases dinámicas, no data-testid

El error común es culpar a los equipos de frontend por no usar data-testid. La realidad: incluso con buenas prácticas, enfrentas:

  • Frameworks CSS-in-JS: Styled Components, Emotion o Tailwind generan clases hash como sc-bdfBQB o bg-[#1da1f2] que cambian entre builds.
  • Rehydration de React/Vue: El DOM inicial difiere del renderizado final, causando «element detached».
  • A/B testing en producción: El mismo flujo puede mostrar variantes UI diferentes según el bucket del usuario.
  • No se trata de «selectores frágiles», sino de entornos web inherentemente dinámicos.

    Arquitectura de recuperación: Más que un try-catch

    Un sistema de self-healing robusto sigue cuatro fases:

  • Detección granular: Capturar excepciones específicas (TimeoutError, ElementHandleError, StrictModeViolation) en lugar de hacer parsing de strings de error.
  • Diagnóstico contextual: Tomar snapshot del DOM visible y el selector que falló (no genéricos).
  • Curación asistida por LLM: Generar selectores alternativos basados en atributos semánticos (texto contenido, role ARIA) en lugar de clases CSS.
  • Validación segura: Verificar que el nuevo selector no contenga payloads maliciosos antes de ejecutar.
  • Evita usar MD5 del HTML completo para detectar «estabilidad DOM»: es computacionalmente caro y falla con relojes, animaciones CSS o contadores de notificaciones. Mejor esperar a que el innerText de un contenedor padre sea consistente por 300ms.

    Implementación práctica con Playwright

    Este código corrige los errores comunes de tutoriales básicos: maneja excepciones específicas, valida seguridad y cachea respuestas LLM para no quebrarte financieramente.

    import time
    import hashlib
    from playwright.sync_api import sync_playwright, TimeoutError as PlaywrightTimeout
    from openai import OpenAI
    
    class ResilientAgent:
        def __init__(self):
            self.client = OpenAI()
            self.healing_cache = {}  # Reduce costos LLM en 80%
            
        def is_safe_selector(self, selector):
            """Bloquea inyección de código en selectores generados"""
            dangerous = ['javascript:', 'onerror=', 'onload=', '<script', 'eval(']
            return not any(d in selector.lower() for d in dangerous)
        
        def wait_for_stability(self, page, checks=3):
            """Espera estabilidad sin MD5: compara innerText ligero"""
            last = ""
            stable = 0
            for _ in range(50):  # Max 5 segundos
                current = page.evaluate("document.body?.innerText?.substring(0,500)")
                if current == last:
                    stable += 1
                    if stable >= checks:
                        return True
                else:
                    stable = 0
                last = current
                time.sleep(0.1)
            return False
        
        def heal_selector(self, failed_selector, error_type, page):
            """Invoca LLM solo con contexto relevante (primeros 2KB del body)"""
            cache_key = hashlib.md5(f"{failed_selector}:{error_type}".encode()).hexdigest()
            if cache_key in self.healing_cache:
                return self.healing_cache[cache_key]
                
            context = page.evaluate("""
                () => document.body.innerHTML.substring(0,2000)
                    .replace(/<scriptb[^<]*(?:(?!</script>)<[^<]*)*</script>/gi, '')
            """)
            
            prompt = f"""Selector fallido: '{failed_selector}'
    Error: {error_type}
    DOM (limpio): ...{context}...
    Genera UN selector CSS alternativo robusto basado en atributos estables (id, name, aria-label, texto exacto). 
    Solo el selector, sin comillas ni explicaciones."""
            
            # Usar modelo pequeño para latencia <800ms
            resp = self.client.chat.completions.create(
                model="gpt-4o-mini",
                messages=[{"role": "user", "content": prompt}],
                max_tokens=60
            )
            
            new_selector = resp.choices[0].message.content.strip()
            
            if not self.is_safe_selector(new_selector):
                raise SecurityError(f"Selector rechazado por seguridad: {new_selector}")
                
            self.healing_cache[cache_key] = new_selector
            return new_selector
        
        def rebuild_action(self, page, selector, action, value=None):
            """Reconstruye la acción con reintentos explícitos"""
            loc = page.locator(selector).first  # .first evita errores strict mode
            
            if action == "click":
                loc.click(timeout=8000)
            elif action == "fill":
                loc.fill(value, timeout=8000)
            elif action == "select":
                loc.select_option(value, timeout=8000)
        
        def execute(self, page, selector, action="click", value=None):
            """Entry point con manejo de excepciones específicas de Playwright"""
            current_selector = selector
            
            try:
                self.rebuild_action(page, current_selector, action, value)
                return {"status": "success", "selector": current_selector, "healed": False}
                
            except PlaywrightTimeout:
                # DOM lento o selector obsoleto
                self.wait_for_stability(page)
                current_selector = self.heal_selector(selector, "TimeoutError", page)
                self.rebuild_action(page, current_selector, action, value)
                return {"status": "success", "selector": current_selector, "healed": True}
                
            except Exception as e:
                # Elemento desconectado del DOM (rehydration)
                error_str = str(e).lower()
                if "detached" in error_str or "strict" in error_str:
                    current_selector = self.heal_selector(selector, "DetachedError", page)
                    self.rebuild_action(page, current_selector, action, value)
                    return {"status": "success", "selector": current_selector, "healed": True}
                raise
    

    Costos, latencia y seguridad: Lo que duele

    Invocar GPT-4 en cada fallo es económicamente suicida. Un agente medio puede fallar 200 veces al día. A $0.03 por invocación, hablamos de $180/mes solo en «curación». Estrategias de mitigación:

  • Caching agresivo: Los selectores de headers, footers y menús son estables. Usa un diccionario LRU para no pagar dos veces por el mismo error.
  • Modelos locales: Qwen2.5-Coder (7B) o Llama 3.2 (3B) corren en CPU y generan selectores válidos en 300ms sin costo por token.
  • Rate limiting: Si fallan más de 5 selectores distintos en 1 minuto, pausa el agente. Es probable que el sitio haya cambiado radicalmente (rediseño) y estarás quemando dinero en correcciones que no funcionarán.
  • Riesgo de inyección: Un atacante que controle el HTML podría inyectar javascript:alert(1) como «selector sugerido». Siempre sanitiza con whitelist (^[a-zA-Z0-9#[]._-='"s]+$) antes de pasar el selector a Playwright.

    Alternativas maduras: No reinventes la rueda

    Antes de montar tu propio sistema, evalúa:

  • Helium (open-source): Wrapper sobre Selenium/Playwright con self-healing básico basado en heurísticas (texto visible, posición).
  • Scrapy-Playwright: Ideal para scraping masivo, maneja reintentos a nivel de middleware, aunque sin LLM.
  • Testim/Mabl (comerciales): Líderes en testing auto-mantenible. Usan computer vision + ML para identificar elementos incluso si cambia el DOM por completo. Precio alto ($300-500/mes) pero viable para empresas.
  • Checklist para producción

  • [ ] Implementa excepciones específicas (PlaywrightTimeout, ElementHandleError) antes de invocar al LLM.
  • [ ] Nunca uses MD5 de HTML completo; prioriza estabilidad de texto visible o MutationObserver.
  • [ ] Valida selectores generados contra patrones peligrosos (javascript:, onerror).
  • [ ] Cachea correcciones por al menos 24 horas para reducir costos OpenAI.
  • [ ] Considera modelos locales (Ollama + llama3.2) para latencia <500ms en vez de APIs remotas.
  • [ ] Monitorea el ratio «healing events / total actions». Si supera el 5%, el sitio objetivo es demasiado volátil; necesitas cambiar estrategia (ej: computer vision).

Conclusión

La recuperación automática en agentes de IA no es magia: es ingeniería defensiva. Un sistema bien diseñado no solo te ahorra pagos de guardia a las 3 AM, sino que hace tus automatizaciones web resilientes ante el caos natural del frontend moderno. Pero recuerda: cada invocación a un LLM es un dólar que sale de tu bolsillo. Diseña para fallar barato, curar rápido y nunca confiar ciegamente en lo que un modelo genera sin validación.

¿Ya implementaste self-healing en tus agentes? Comparte tu approach en los comentarios o únete al newsletter donde semanalmente deconstruimos arquitecturas de IA para producción real.

Cómo evitar que tu agente LLM olvide conversaciones críticas: guía de memoria a largo plazo en producción

Tu agente recuerda perfectamente lo que le dijiste hace cinco minutos, pero ignora instrucciones clave del onboarding que hiciste ayer. Si te suena familiar, estás enfrentando el problema de contexto limitado inherente a los transformers. Cuando implementamos patrones de memoria a largo plazo en producción, no estamos solo guardando historiales de chat: estamos diseñando sistemas que evitan la degradación del rendimiento por recuperación de documentos irrelevantes y el fenómeno documentado como lost in the middle (donde los LLMs ignoran información ubicada en el centro de contextos extensos).

El problema real: más allá del límite de tokens

Los modelos actuales soportan ventanas de 128k o 200k tokens, pero la capacidad efectiva de razonamiento sobre todo ese contexto no escala linealmente. Investigaciones de Stanford y Anthropic demuestran que el accuracy cae significativamente cuando la información relevante se encuentra en medio de un contexto extenso, especialmente al superar los 20k tokens.

Esto se agrava cuando tu sistema RAG recupera documentos semánticamente similares pero conceptualmente irrelevantes. Un retrieval naive puede saturar el contexto con ruido, haciendo que el modelo «olvide» restricciones críticas de seguridad o preferencias del usuario establecidas en interacciones previas.

Arquitectura de tres capas: más allá del simple historial

Basándonos en taxonomías formalizadas por proyectos como [MemGPT](https://memgpt.ai/) (Packer et al., 2023) y los módulos de memoria de LangChain, una arquitectura robusta de producción separa la memoria en tres capas distintas:

Memoria semántica (conocimiento externo)

Vector stores como Pinecone, Weaviate o pgvector almacenan embeddings de conversaciones pasadas, documentos y hechos persistentes. Sin embargo, el simple cosine similarity no es suficiente: necesitas estrategias de reranking (como Cohere Rerank o cross-encoders) para filtrar falsos positivos semánticos antes de inyectar al prompt.

Memoria procedural (aprendizajes codificados)

Son reglas explícitas derivadas del comportamiento del usuario que modifican la ejecución del agente. A diferencia de la semántica, aquí almacenamos código ejecutable o configuraciones estructuradas:

def actualizar_estilo_usuario(user_id: str, feedback: dict) -> None:
    """
    Actualiza preferencias de comunicación basadas en interacciones previas.
    """
    preferencias = {
        "tono": feedback.get("formalidad", "neutral"),
        "longitud_respuesta": feedback.get("verbosity", "conciso"),
        "tecnologias_frecuentes": feedback.get("stack", [])
    }
    
    prompt_sistema = f"""
    Eres un asistente técnico. Reglas estrictas para este usuario:
    - Tono: {preferencias['tono']}
    - Longitud: {preferencias['longitud_respuesta']}
    - Contexto técnico priorizado: {', '.join(preferencias['tecnologias_frecuentes'])}
    - NUNCA sugieras soluciones fuera de su stack sin consultar primero
    """
    
    guardar_en_redis(user_id, prompt_sistema, ttl=86400)

Memoria episódica (historial reciente)

El buffer de ventana deslizante de las últimas N interacciones. Crítico para mantener coherencia en la conversación actual, pero volátil por diseño.

Estrategias de recuperación inteligente

Filtrado estricto por usuario (context contamination)

Un error común en producción es no aislar los vectores por user_id o session_id. Recuperar memorias de usuario A cuando conversas con usuario B genera context contamination, filtraciones de privacidad y alucinaciones cruzadas. Implementa metadata filtering obligatorio en tus queries vectoriales:

# Query segura con namespace isolation
results = index.query(
    vector=embedding,
    filter={"user_id": {"$eq": current_user_id}},
    top_k=5,
    namespace=f"user_{current_user_id}"
)

Reranking antes del prompt

Incluso con buenos embeddings, recuperar 10 documentos donde solo 2 son relevantes satura el contexto. Utiliza modelos de reranking para reordenar por relevancia real:

import cohere

co = cohere.Client(api_key)

def rerank_documents(query, documents, top_n=3):
    results = co.rerank(
        model="rerank-english-v2.0",
        query=query,
        documents=documents,
        top_n=top_n
    )
    return [documents[r.index] for r in results.results]

Esto mitiga el lost in the middle al reducir la cantidad de ruido que llega al modelo.

Implementación práctica y costos

Para la capa semántica, Pinecone en modo serverless es una opción popular, aunque debes verificar precios actualizados según tu región AWS/GCP y volumen de consultas. Nota: Los precios varían significativamente entre modo serverless (por GB almacenado y consultas) versus pods dedicados (por hora). La cifra orientativa de ~$0.10 por millón de vectores puede fluctuar según dimensiones y región; valida siempre la calculadora oficial antes de escalar.*

Patrón de sincronización híbrida

Combina las tres capas en tu pipeline:

  • Pre-procesamiento: Recupera preferencias procedurales de Redis/PostgreSQL (sub-10ms)
  • Retrieval semántico: Query vectorial con metadata filtering estricto
  • Reranking: Filtra a máximo 3-5 documentos altamente relevantes
  • Construcción de prompt: Inyecta memoria procedural primero (reglas duras), luego contexto episódico reciente, finalmente documentos rerankeados
  • async def build_context(user_id: str, current_message: str):
        # 1. Memoria procedural (alta prioridad)
        system_rules = await redis.get(f"user:{user_id}:preferences")
        
        # 2. Memoria episódica (últimos 3 turnos)
        recent_chat = await get_recent_history(user_id, limit=3)
        
        # 3. Memoria semántica con reranking
        raw_docs = await vector_search(current_message, user_id=user_id, top_k=10)
        relevant_docs = rerank_documents(current_message, raw_docs, top_n=3)
        
        return {
            "system": system_rules,
            "context": recent_chat + relevant_docs
        }
    

    Conclusión

    Implementar memoria a largo plazo no es solo «guardar todo en una base vectorial». Requiere arquitectura disciplinada: aislamiento estricto de usuarios para evitar contaminación, reranking para combatir el ruido del retrieval, y separación clara entre conocimiento factual (semántico), reglas de ejecución (procedural) y contexto conversacional (episódico).

    Aprendizaje clave: Un agente con 100k tokens de contexto pero sin filtrado inteligente performa peor que uno con 8k tokens y recuperación curada. La calidad del contexto siempre supera a la cantidad.

    ¿Ya implementaste algún patrón de memoria avanzada? Comparte tu arquitectura en los comentarios o sígueme para el próximo post sobre evaluación automatizada de retrieval en RAG pipelines.

    LinkedIn
    Share
    Instagram
    WhatsApp