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.

Edge AI Agents: Cómo reemplazar tu CMS tradicional con un agente autónomo (sin quemar tu presupuesto ni tu SEO)

¿Cuánto pagas al mes por mantener un CMS tradicional que 90% del tiempo solo sirve HTML estático? Los Edge AI Agents prometen eliminar esa infraestructura pesada, generando contenido personalizado directamente en el edge, cerca del usuario y sin servidores que administrar. Pero antes de migrar tu blog corporativo o tu sitio de documentación, necesitas saber dónde está el truco: esta arquitectura no es gratis ilimitado, no es determinista, y definitivamente no sirve para todo.

La arquitectura real: cómo funciona sin mentirte sobre la latencia

Un Edge AI Agent en Cloudflare (la opción más madura actualmente) combina tres piezas: Workers (cómputo edge), Vectorize (base de datos vectorial) y Workers AI (inferencia de modelos locales). La idea es simple: en lugar de consultar una API de CMS, tu Worker recibe la petición, recupera contexto relevante de Vectorize mediante búsqueda semántica, y genera la respuesta HTML usando un modelo como Llama 3.2 3B directamente en la red de Cloudflare.

Pero hablemos claro sobre los números. Generar una respuesta completa con retrieval incluido no te dará un TTFB (Time to First Byte) de menos de 100ms. Estás hablando de 200ms a 800ms dependiendo de la complejidad del prompt y el tamaño del contexto. Esto es aceptable para contenido dinámico personalizado, pero mortal si pretendes reemplazar la home de un e-commerce con esto.

Otro error común: no puedes usar SQLite-vec dentro de un Worker. Los Workers de Cloudflare corren en V8 sin filesystem persistente. Para el vector search debes usar Vectorize nativo (que sí persiste índices en la red edge) o hacer llamadas a una API externa. Si ves ejemplos con Python y SQLite local, eso es solo para prototipar en tu laptop, no para producción edge.

Cálculo de costos: el plan gratuito tiene techo (y bajo)

La documentación de Cloudflare suena generosa, pero los límites del plan gratuito son estrictos:

  • Workers AI: 100,000 tokens de LLM por día (no 1 millón). Un artículo promedio de 800 palabras consume ~1,200 tokens entre prompt y completación. Haces 80 publicaciones al día y se acabó.
  • Vectorize: 1 millón de consultas vectoriales al día en el tier gratuito (suficiente para MVP, pero no para tráfico viral).
  • KV: Operaciones de lectura ilimitadas, pero escrituras con consistencia eventual (hasta 60 segundos de propagación global). No sirve para contenido que «cambia cada minuto» ni para contadores en tiempo real.
  • En producción real, los números cambian:

  • Workers AI (pago): ~$0.011 por 1,000 tokens para Llama 3.2 3B.
  • Un sitio con 10,000 visitas diarias donde cada página genera 1,500 tokens = 15M tokens/día = ~$165 mensuales solo en inferencia.
  • Esto sin contar el costo oculto: desarrollo, mantenimiento de prompts, debugging de alucinaciones a las 3 AM, y la curva de aprendizaje de una arquitectura eventualmente consistente.

    El problema del SEO y las alucinaciones

    Aquí está el riesgo que pocos mencionan: el contenido generado por LLM en cada request es no-determinista. Google puede indexar tu página sobre «mejores frameworks JavaScript» hoy y mañana, para la misma URL, el agente genera contenido diferente o peor: información contradictoria. Esto crea riesgo de contenido duplicado, cannibalización de keywords y penalizaciones por «thin content» cambiante.

    Estrategias de mitigación obligatorias:

  • Temperatura 0: Fuerza al modelo a elegir siempre el token más probable, reduciendo variabilidad.
  • Few-shot prompting: Incluye ejemplos exactos del formato de salida deseado en el system prompt.
  • Caché agresiva: Usa Cloudflare Cache API para almacenar el HTML generado durante horas (o días) si el contenido no es personalizado por usuario.
  • SSG híbrido: Genera artículos base estáticamente (usando el agente en build time) y reserva el edge AI solo para personalización de componentes (recomendaciones, resúmenes contextuales).
  • // Ejemplo: Worker con mitigación de alucinaciones y caché
    export default {
      async fetch(request, env) {
        const cache = caches.default;
        const cached = await cache.match(request);
        if (cached) return cached;
        // Retrieval desde Vectorize (NO sqlite-vec)
        const query = new URL(request.url).searchParams.get('q');
        const vectors = await env.VECTORIZE_INDEX.query(
          await env.AI.run('@cf/baai/bge-base-en-v1.5', { text: query }),
          { topK: 3 }
        );
        // Generación con temperatura 0 y few-shot
        const response = await env.AI.run('@cf/meta/llama-3.2-3b-instruct', {
          messages: [
            { role: 'system', content: 'Eres un editor técnico. Responde SOLO con HTML válido. Ejemplo: <article><h1>...</h1></article>' },
            { role: 'user', content: `Contexto: ${JSON.stringify(vectors)}. Pregunta: ${query}` }
          ],
          temperature: 0.0, // Determinismo máximo
          max_tokens: 1000
        });
        const html = response.response;
        // Cachear por 1 hora para evitar regeneración costosa
        const resp = new Response(html, { headers: { 'Content-Type': 'text/html', 'Cache-Control': 'public, max-age=3600' } });
        cache.put(request, resp.clone());
        return resp;
      }
    };
    

    Limitaciones críticas: cuando NO usar Edge AI Agents

    No todo se soluciona con magia edge. Evita esta arquitectura si:

  • E-commerce crítico: Un error de alucinación en el precio o disponibilidad de stock te expone a demandas. Los CMS tradicionales tienen auditoría de contenido; un LLM generativo no garantiza consistencia legal.
  • Alto volumen de escritura concurrente: KV no soporta transacciones ACID. Si dos usuarios editan «el último post» simultáneamente, perderás datos.
  • Requisitos de auditoría: ¿Necesitas saber exactamente qué texto mostraste a un usuario específico el 15 de marzo a las 3 PM? La generación dinámica dificulta el versionado forense.
  • SEO puro: Si tu negocio depende del ranking orgánico, el riesgo de variabilidad semántica no vale la pena frente a HTML estático pre-renderizado.

Conclusión: herramienta, no religión

Los Edge AI Agents son extraordinarios para personalización de contenido existente (adaptar un artículo técnico al nivel del lector, generar resúmenes contextuales, responder preguntas sobre documentación). Son terribles como reemplazo 1:1 de un CMS editorial tradicional donde el control, la auditoría y el SEO determinista son requisitos.

Mi recomendación práctica: usa esta arquitectura para el 10% de tu sitio que necesita verdadera personalización inteligente, manteniendo el 90% como contenido estático generado en build time. Así pagas centavos por la magia, no miles de dólares por alucinaciones en producción.

¿Listo para experimentar? Empieza con un proyecto paralelo de bajo riesgo (un FAQ interno o un recomendador de contenido), mide tus costos reales durante 30 días, y valida que Google indexa correctamente antes de tocar tu sitio principal. La infraestructura edge es poderosa, pero solo si conoces exactamente dónde están sus límites.

¿Has intentado migrar un CMS tradicional a IA generativa? Cuéntame en los comentarios qué rompió primero: el presupuesto o el SEO.

Model Context Protocol (MCP): Estandarización de herramientas para agentes de IA en 2025

El principal cuello de botella en arquitecturas de agentes de IA no es la capacidad del modelo, sino la integración con sistemas externos. Cada nueva herramienta—desde una base de datos hasta una API interna—requiere código específico que acopla al agente con la implementación concreta. Model Context Protocol (MCP), propuesto por Anthropic en noviembre de 2024 y actualmente en evolución constante, busca eliminar esa fricción mediante un protocolo abierto que estandariza cómo los agentes descubren y ejecutan herramientas externas. No se trata de un estándar formal ratificado por organismos como IETF o W3C, sino de un proyecto open source (licencia MIT) que ofrece una capa de abstracción práctica y está ganando tracción rápida en el ecosistema de desarrollo de agentes.

El problema de la integración ad-hoc

Antes de MCP, integrar un agente con herramientas externas implicaba escribir código específico por cada servicio. Si tu agente necesitaba consultar PostgreSQL, interactuar con Slack y leer archivos de S3, debías implementar tres adaptadores distintos, manejar autenticaciones particulares y mapear manualmente los esquemas de entrada/salida. Este enfoque genera:

  • Acoplamiento excesivo: Cambios en la API de la herramienta obligan a modificar la lógica del agente.
  • Duplicación de esfuerzos: Cada equipo reimplementa conectores similares.
  • Dificultad para escalar: Añadir una nueva capacidad requiere tocar el código core del agente.
  • MCP aborda esto mediante un modelo cliente-servidor donde las herramientas se exponen como servicios independientes que cualquier agente con capacidad MCP puede consumir, siempre y cuando implemente la capa cliente correspondiente.

    Arquitectura y componentes clave

    El protocolo define tres elementos fundamentales que operan sobre JSON-RPC 2.0:

    1. Servidores MCP

    Son procesos independientes que exponen recursos (datos contextuales) y herramientas (funciones ejecutables). Un servidor PostgreSQL, por ejemplo, expone operaciones de consulta sin revelar detalles de conexión al agente.

    2. Clientes MCP

    Implementaciones integradas en el agente o en el framework que orquesta la ejecución (como LangChain o la SDK de Anthropic). El cliente gestiona el descubrimiento de herramientas, el paso de parámetros y la serialización de resultados.

    3. Capa de transporte

    Actualmente soporta:

  • stdio: Comunicación via streams estándar (ideal para servidores locales).
  • SSE (Server-Sent Events): Para servidores remotos sobre HTTP.
  • Esta separación permite que un mismo servidor MCP sirva a múltiples agentes, y que un agente consuma múltiples servidores sin código específico por integración.

    Implementación práctica: un servidor PostgreSQL

    A continuación, un servidor MCP funcional que expone consultas a PostgreSQL. Nota importante: utilizamos asyncpg para mantener la compatibilidad asíncrona con el loop de eventos de MCP, evitando el bloqueo que causaría psycopg2.

    # server.py
    import asyncpg
    import asyncio
    from mcp.server import Server
    from mcp.types import Tool, TextContent
    
    app = Server("postgres-server")
    
    @app.list_tools()
    async def list_tools() -> list[Tool]:
        return [
            Tool(
                name="query_database",
                description="Ejecuta consultas SELECT seguras",
                inputSchema={
                    "type": "object",
                    "properties": {
                        "sql": {"type": "string", "description": "Consulta SQL"}
                    },
                    "required": ["sql"]
                }
            )
        ]
    
    @app.call_tool()
    async def call_tool(name: str, arguments: dict) -> list[TextContent]:
        if name != "query_database":
            raise ValueError(f"Herramienta desconocida: {name}")
        
        # Uso correcto de librería asíncrona
        conn = await asyncpg.connect(
            "postgresql://user:pass@localhost/db",
            min_size=1, max_size=10
        )
        try:
            rows = await conn.fetch(arguments["sql"])
            result = "n".join([str(dict(row)) for row in rows])
            return [TextContent(type="text", text=result)]
        finally:
            await conn.close()
    
    async def main():
        from mcp.server.stdio import stdio_server
        async with stdio_server() as (read_stream, write_stream):
            await app.run(
                read_stream, 
                write_stream,
                app.create_initialization_options()
            )
    
    if __name__ == "__main__":
        asyncio.run(main())
    

    Para simplificar el desarrollo en Python, existe FastMCP, un SDK de alto nivel sobre la implementación oficial que reduce la verbosidad:

    from fastmcp import FastMCP
    import asyncpg
    
    mcp = FastMCP("postgres-server")
    
    @mcp.tool()
    async def query_database(sql: str) -> str:
        """Ejecuta consultas SELECT seguras"""
        conn = await asyncpg.connect("postgresql://user:pass@localhost/db")
        try:
            rows = await conn.fetch(sql)
            return "n".join([str(dict(row)) for row in rows])
        finally:
            await conn.close()
    
    if __name__ == "__main__":
        mcp.run()
    

    Limitaciones y consideraciones actuales (Enero 2025)

    Si bien MCP representa un avance significativo, es crucial entender su estado y restricciones:

  • No es plug-and-play universal: Para que un agente consuma un servidor MCP, debe implementar explícitamente un cliente MCP. No funciona automáticamente con agentes existentes que no tengan esta capa.
  • Especificación en evolución: La versión actual (2024-11-05) cambia rápidamente. Lo que funciona hoy puede requerir ajustes menores en meses venideros.
  • Gestión de estado: El protocolo es stateless por diseño. Si necesitas sesiones persistentes o gestión compleja de estado entre llamadas, deberás implementarla a nivel de aplicación.
  • Seguridad: El servidor MCP tiene acceso directo a tus sistemas. Debes implementar validación de inputs, rate limiting y autenticación en la capa de transporte, ya que el protocolo base no prescribe mecanismos de seguridad específicos.

Conclusión

Model Context Protocol ofrece una ruta concreta para desacoplar agentes de IA de sus herramientas, promoviendo una arquitectura modular donde los desarrolladores pueden añadir capacidades sin modificar el core del sistema. Sin embargo, su adopción requiere una inversión inicial: implementar la capa cliente en tu agente y adaptar tus herramientas al formato de servidor MCP.

Aprendizajes clave:

  • MCP reduce la duplicación de código de integración, pero introduce una dependencia arquitectónica (el cliente MCP).
  • Usa siempre librerías asíncronas compatibles (asyncpg, aiohttp) en servidores MCP para no bloquear el event loop.
  • Evalúa FastMCP si buscas prototipar rápidamente en Python, pero migra a la SDK oficial si necesitas control granular del protocolo.
  • Llamada a la acción: Identifica una herramienta interna que actualmente integres con código específico en tus agentes. Intenta encapsularla en un servidor MCP durante esta semana. La fricción inicial es real, pero el desacoplamiento posterior justifica el esfuerzo cuando tu stack de herramientas comience a crecer exponencialmente.

    Arquitectura de memoria para Agentes de IA: Estrategias para mantener contexto persistente sin costos exponenciales

    El problema no es que tu agente olvide lo que dijo hace cinco minutos, es que recordarlo todo le cuesta 40 dólares por hora a tu empresa. La arquitectura de memoria para Agentes de IA se ha convertido en el diferenciador técnico entre un prototipo de juguete y un sistema escalable en producción. Mientras los modelos multiplican sus ventanas de contexto hasta 128k o incluso 200k tokens, enviar la conversación completa en cada llamada es económicamente suicida. Aquí no vamos a reinventar LangChain: te explico cómo funciona la memoria bajo el capó para que optimices lo que ya usas o construyas tu propia solución sin caer en trampas de diseño que duplican tus costos de inferencia.

    El problema real: obesidad del contexto y «lost in the middle»

    Los grandes modelos de lenguaje sufren de dos dolencias crónicas cuando les envías demasiado texto. La primera es económica: los precios de API escalan linealmente con los tokens de entrada. Si mantienes un historial de 20 mensajes de 500 tokens cada uno, estás pagando por 10k tokens de «contexto muerto» antes de que el modelo escriba una sola palabra.

    La segunda es de atención: el fenómeno «lost in the middle» demuestra que la precisión del modelo decae dramáticamente cuando la información relevante está en el medio de un contexto masivo, incluso con ventanas de 128k tokens. Más contexto no es mejor contexto; es solo contexto más caro y más ruidoso.

    Jerarquías de memoria: diseño en capas

    Un agente productivo no tiene una sola «memoria», sino tres sistemas especializados que operan a diferentes escalas temporales y económicas.

    Working Memory: ventana deslizante por tokens, no por mensajes

    El error clásico es usar deque(maxlen=10) y luego recortar por tokens. Es contradictorio: si limitas físicamente a 10 mensajes, ¿por qué validar tokens? Si cada mensaje tiene 2k tokens (aproximadamente 1,500 palabras en español), esos «solo 10 mensajes» explotan tu presupuesto.

    La solución es un límite exclusivo por tokens, sin restricción de cantidad de mensajes. En español, un token aproxima a una palabra (no uses la regla inglesa de 0.75 palabras/token ni la obsoleta de 4 caracteres/token). Para precisión quirúrgica, usa tiktoken o el tokenizer específico de tu modelo.

    import tiktoken
    from typing import List, Dict
    
    class WorkingMemory:
        def __init__(self, max_tokens: int = 4000, model: str = "gpt-4"):
            self.messages: List[Dict] = []  # Lista simple, sin maxlen arbitrario
            self.max_tokens = max_tokens
            self.encoding = tiktoken.encoding_for_model(model)
        
        def add_message(self, role: str, content: str):
            self.messages.append({"role": role, "content": content})
            self._enforce_token_limit()
        
        def _enforce_token_limit(self):
            """Recorta desde el inicio manteniendo los mensajes más recientes."""
            total = 0
            for i, msg in enumerate(reversed(self.messages)):
                # Conteo real de tokens, no aproximación por caracteres
                tokens = len(self.encoding.encode(msg["content"]))
                total += tokens
                if total > self.max_tokens:
                    cutoff = len(self.messages) - i - 1
                    self.messages = self.messages[cutoff:]
                    break
        
        def get_context(self) -> List[Dict]:
            return self.messages
    

    Este diseño garantiza que nunca excedas los 4k tokens de ventana reciente, sin importar si fueron 3 mensajes largos o 50 mensajes cortos.

    Memoria Semántica: retrieval cuando el pasado importa

    No todo lo antiguo merece estar en el prompt activo. Para conocimiento factual persistente (preferencias del usuario, documentos subidos, historial de compras), usa una base vectorial. Convierte interacciones pasadas a embeddings y recupera solo los top-k relevantes vía búsqueda por similitud cuando el agente lo solicite explícitamente.

    Esta estrategia transfiere el costo de O(n) tokens a O(1) tokens + costo fijo de embedding (barato) y retrieval (casi gratuito).

    Memoria Episódica: compresión progresiva

    Cuando la conversación es larga pero no puedes descartarla (ej. negociaciones complejas), implementa summarization condensado. Cada N interacciones, genera un resumen de los puntos clave y descarta el raw text. Tu Working Memory mantiene solo el resumen + los últimos mensajes crudos.

    class EpisodicMemory:
        def __init__(self):
            self.summaries: List[str] = []
            self.recent_raw: WorkingMemory = WorkingMemory(max_tokens=2000)
        
        def compress_if_needed(self, llm_client):
            if self.recent_raw.token_count > 3000:
                summary = llm_client.summarize(self.recent_raw.get_context())
                self.summaries.append(summary)
                self.recent_raw = WorkingMemory(max_tokens=2000)
        
        def get_full_context(self):
            return {
                "historical_summaries": self.summaries,
                "recent_messages": self.recent_raw.get_context()
            }
    

    Estimación de tokens: precisión vs velocidad

    Nunca uses len(text) // 4 para estimar tokens en español. Un texto técnico en español genera aproximadamente 1 token por palabra (ligeramente más si hay muchos signos de puntuación o código). Para producción, integra siempre el tokenizer oficial:

    • OpenAI: tiktoken
    • Anthropic: API de conteo propia o heurística de 1.2 tokens/palabra
    • Open source (Llama, Mistral): transformers.AutoTokenizer
    • Un error común es asumir que «40k tokens por interacción» es normal. En escenarios conversacionales reales, un mensaje de usuario rara vez supera los 200-500 tokens. Si estás procesando documentos masivos, hazlo por chunks, no cargues PDFs completos en el contexto de chat.

      En producción: ¿Construir o adoptar?

      No reinventes la rueda a menos que tengas restricciones de latencia extremas o lógica de negocio muy específica.

    • LangChain ofrece ConversationBufferWindowMemory (limita por mensajes) y ConversationSummaryMemory (limita por tokens con resumen). Úsalos como punto de partida.
    • LlamaIndex es superior para arquitecturas RAG complejas donde la memoria semántica es el componente dominante.
    • Custom (como el código anterior) solo cuando necesites control granular sobre el recorte (ej. priorizar mensajes del sistema sobre mensajes del usuario) o cuando estés optimizando costos en escala masiva donde cada token cuenta.

    Conclusión

    La arquitectura de memoria eficiente se reduce a tres principios: recorta por tokens reales, no por cantidad de mensajes; mueve el conocimiento histórico a sistemas de retrieval, no al prompt; y comprime progresivamente lo que no cabe en tu ventana de trabajo. Los modelos con 200k tokens de contexto son una trampa seductora: te permiten ser perezoso, pero te cobran por esa pereza en cada inferencia.

    Antes de optimizar código, mide tu consumo real de tokens por sesión de usuario. Si estás por encima de los 8k tokens de entrada promedio, tienes un problema de arquitectura, no de modelo.

    ¿Dónde está el cuello de botella en tu agente? Revisa tu última factura de API y dime cuántos tokens estás enviando de contexto histórico vs. cuántos generas de respuesta. Si la proporción está desbalanceada, es hora de rediseñar tu memoria.

    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