SamuelChien/coleccion-mega-habilidades

Sistemas de Memoria para Agentes

La memoria es la piedra angular de los agentes inteligentes. Sin ella, cada interacción comienza desde cero. Esta habilidad cubre la arquitectura de la memoria del agente: a corto plazo (ventana de contexto), a largo plazo (almacenes vectoriales) y las arquitecturas cognitivas que las organizan.

Idea clave: La memoria no es solo almacenamiento, es recuperación. Un millón de datos almacenados no significan nada si no puedes encontrar el correcto. Las estrategias de fragmentación, incrustación y recuperación determinan si tu agente recuerda u olvida.

El campo está fragmentado con terminología inconsistente. Utilizamos el marco de arquitectura cognitiva CoALA: memoria semántica (hechos), memoria episódica (experiencias) y memoria procedimental (conocimiento procedimental).

Principios

• Calidad de la memoria = calidad de la recuperación, no cantidad de almacenamiento
• Fragmentar para la recuperación, no para el almacenamiento
• El aislamiento del contexto es enemigo de la memoria
• Tipo de memoria adecuado para la información adecuada
• Hacer decaer las memorias antiguas: no todo debe ser para siempre
• Probar la precisión de la recuperación antes de producción
• La formación de memoria en segundo plano supera a la de tiempo real

Capacidades

• memoria-agente
• memoria-largo-plazo
• memoria-corto-plazo
• memoria-de-trabajo
• memoria-episódica
• memoria-semántica
• memoria-procedimental
• recuperacion-memoria
• formacion-memoria
• decaimiento-memoria

Alcance

• operaciones-base-de-datos-vectorial → ingeniero-de-datos
• arquitectura-pipeline-rag → arquitecto-llm
• seleccion-modelo-incrustacion → ingeniero-ml
• diseno-grafo-conocimiento → ingeniero-conocimiento

Herramientas

Marcos_de_memoria

• LangMem (LangChain) - Cuándo: Agentes LangGraph con memoria persistente Nota: Tipos de memoria semántica, episódica y procedimental
• MemGPT / Letta - Cuándo: Gestión de contexto virtual, memoria tipo SO Nota: Niveles de memoria jerárquicos, paginación automática
• Mem0 - Cuándo: Capa de memoria de usuario para personalización Nota: Diseñado para preferencias e historial del usuario

Almacenes_vectoriales

• Pinecone - Cuándo: Gestionado, escala empresarial (miles de millones de vectores) Nota: Mejor rendimiento de consultas, mayor costo
• Qdrant - Cuándo: Filtrado de metadatos complejo, código abierto Nota: Basado en Rust, excelente filtrado
• Weaviate - Cuándo: Búsqueda híbrida, funciones de grafo de conocimiento Nota: Interfaz GraphQL, bueno para relaciones
• ChromaDB - Cuándo: Prototipado, aplicaciones pequeñas/medianas Nota: Amigable para desarrolladores, ~20ms p50 con 100K vectores
• pgvector - Cuándo: Ya usas PostgreSQL, configuración más simple Nota: Bueno para <1M vectores, herramientas conocidas

Modelos_de_incrustacion

• OpenAI text-embedding-3-large - Cuándo: Mejor calidad, 3072 dimensiones Nota: $0.13/1M tokens
• OpenAI text-embedding-3-small - Cuándo: Buen equilibrio, 1536 dimensiones Nota: $0.02/1M tokens, 5 veces más barato
• nomic-embed-text-v1.5 - Cuándo: Código abierto, despliegue local Nota: 768 dimensiones, buena calidad
• all-MiniLM-L6-v2 - Cuándo: Ligero, incrustación local rápida Nota: 384 dimensiones, latencia más baja

Patrones

Arquitectura de Tipo de Memoria

Elegir el tipo de memoria adecuado para diferentes informaciones

Cuándo usarlo: Al diseñar el sistema de memoria del agente

ARQUITECTURA DE TIPO DE MEMORIA (Marco CoALA):

""" Tres tipos de memoria para diferentes propósitos:

1. Memoria Semántica: Hechos y conocimiento

   • Lo que sabes sobre el mundo
   • Preferencias del usuario, conocimiento del dominio
   • Almacenada en perfiles (estructurados) o colecciones (no estructurados)

2. Memoria Episódica: Experiencias y eventos

   • Lo que sucedió (eventos con marca de tiempo)
   • Conversaciones pasadas, resultados de tareas
   • Usada para aprender de la experiencia

3. Memoria Procedimental: Cómo hacer cosas

   • Reglas, habilidades, flujos de trabajo
   • A menudo implementada como ejemplos de pocos disparos
   • "¿Cómo resolví esto antes?" """

Implementación LangMem

""" from langmem import MemoryStore from langgraph.graph import StateGraph

Inicializar almacén de memoria

memory = MemoryStore( connection_string=os.environ["POSTGRES_URL"] )

Memoria semántica: perfil de usuario

await memory.semantic.upsert( namespace="perfil_usuario", key=user_id, content={ "name": "Alicia", "preferences": ["modo oscuro", "respuestas concisas"], "expertise_level": "desarrollador", } )

Memoria episódica: interacción pasada

await memory.episodic.add( namespace="conversaciones", content={ "timestamp": datetime.now(), "summary": "Ayudé a depurar un problema de autenticación", "outcome": "resuelto", "key_insights": ["La causa raíz fue la expiración del token"], }, metadata={"user_id": user_id, "topic": "depuración"} )

Memoria procedimental: patrón aprendido

await memory.procedural.add( namespace="habilidades", content={ "task_type": "depurar_auth", "steps": ["Verificar expiración del token", "Verificar flujo de actualización"], "example_interaction": few_shot_example, } ) """

Recuperación de Memoria en Tiempo de Ejecución

""" async def prepare_context(user_id, query): # Obtener perfil de usuario (semántica) profile = await memory.semantic.get( namespace="perfil_usuario", key=user_id )

# Encontrar experiencias pasadas relevantes (episódica)
similar_experiences = await memory.episodic.search(
    namespace="conversaciones",
    query=query,
    filter={"user_id": user_id},
    limit=3
)

# Encontrar habilidades relevantes (procedimental)
relevant_skills = await memory.procedural.search(
    namespace="habilidades",
    query=query,
    limit=2
)

return {
    "profile": profile,
    "past_experiences": similar_experiences,
    "relevant_skills": relevant_skills,
}

"""

Patrón de Selección de Almacén Vectorial

Elegir la base de datos vectorial adecuada para tu caso de uso

Cuándo usarlo: Al configurar el almacenamiento de memoria persistente

SELECCIÓN DE ALMACÉN VECTORIAL:

""" Matriz de decisión:

	Pinecone	Qdrant	Weaviate	ChromaDB	pgvector
Escala	Miles de millones	100M+	100M+	1M	1M
Gestionado	Sí	Ambos	Ambos	Propio	Propio
Filtrado	Básico	Mejor	Bueno	Básico	SQL
Híbrido	No	Sí	Mejor	No	Sí
Costo	Alto	Medio	Medio	Gratis	Gratis
Latencia	5ms	7ms	10ms	20ms	15ms
"""

Pinecone (Escala Empresarial)

""" from pinecone import Pinecone

pc = Pinecone(api_key=os.environ["PINECONE_API_KEY"]) index = pc.Index("memoria-agente")

Insertar/actualizar con metadatos

index.upsert( vectors=[ { "id": f"memoria-{uuid4()}", "values": embedding, "metadata": { "user_id": user_id, "timestamp": datetime.now().isoformat(), "type": "episódica", "content": texto_memoria, } } ], namespace=namespace )

Consulta con filtro

results = index.query( vector=query_embedding, filter={"user_id": user_id, "type": "episódica"}, top_k=5, include_metadata=True ) """

Qdrant (Filtrado Complejo)

""" from qdrant_client import QdrantClient from qdrant_client.models import PointStruct, Filter, FieldCondition

client = QdrantClient(url="http://localhost:6333")

Filtrado complejo con Qdrant

results = client.search( collection_name="memoria_agente", query_vector=query_embedding, query_filter=Filter( must=[ FieldCondition(key="user_id", match={"value": user_id}), FieldCondition(key="type", match={"value": "semántica"}), ], should=[ FieldCondition(key="topic", match={"any": ["auth", "seguridad"]}), ] ), limit=5 ) """

ChromaDB (Prototipado)

""" import chromadb

client = chromadb.PersistentClient(path="./memoria_db") collection = client.create_collection("memoria_agente") if "memoria_agente" not in client.list_collections() else client.get_collection("memoria_agente")

Simple y rápido para prototipos

collection.add( ids=[str(uuid4())], embeddings=[embedding], documents=[texto_memoria], metadatas=[{"user_id": user_id, "type": "episódica"}] )

results = collection.query( query_embeddings=[query_embedding], n_results=5, where={"user_id": user_id} ) """

Patrón de Estrategia de Fragmentación

Dividir documentos en fragmentos recuperables

Cuándo usarlo: Al procesar documentos para el almacenamiento de memoria

ESTRATEGIAS DE FRAGMENTACIÓN:

""" El dilema de la fragmentación:

• Demasiado grande: El vector pierde especificidad
• Demasiado pequeño: Pierde contexto

El tamaño óptimo del fragmento depende de:

• Tipo de documento (código vs prosa vs datos)
• Patrones de consulta (fáctico vs exploratorio)
• Modelo de incrustación (cada uno tiene su punto óptimo)

Guía general: 256-512 tokens para la mayoría de los casos de uso """

Fragmentación de Tamaño Fijo (Base)

""" from langchain.text_splitter import RecursiveCharacterTextSplitter

splitter = RecursiveCharacterTextSplitter( chunk_size=500, # Caracteres chunk_overlap=50, # La superposición evita cortar oraciones separators=["\n\n", "\n", ". ", " ", ""] # Orden de prioridad )

chunks = splitter.split_text(documento) """

Fragmentación Semántica (Mejor Calidad)

""" from langchain_experimental.text_splitter import SemanticChunker from langchain_openai import OpenAIEmbeddings

Divide basado en similitud semántica

splitter = SemanticChunker( embeddings=OpenAIEmbeddings(), breakpoint_threshold_type="percentile", breakpoint_threshold_amount=95 )

chunks = splitter.split_text(documento) """

Fragmentación Consciente de la Estructura (Documentos con Jerarquía)

""" from langchain.text_splitter import MarkdownHeaderTextSplitter

Respeta la estructura del documento

splitter = MarkdownHeaderTextSplitter( headers_to_split_on=[ ("#", "Encabezado 1"), ("##", "Encabezado 2"), ("###", "Encabezado 3"), ] )

chunks = splitter.split_text(doc_markdown)

Cada fragmento tiene metadatos de encabezado para contexto

"""

Fragmentación Contextual (Enfoque de Anthropic)

"""

Añadir contexto a cada fragmento antes de incrustar

Reduce las fallas de recuperación en un 35%

def añadir_contexto_al_fragmento(fragmento, resumen_documento): prompt_contexto = f''' Resumen del documento: {resumen_documento}

El siguiente es un fragmento de este documento:
{fragmento}
'''
return prompt_contexto

Incrustar el fragmento contextualizado, no el fragmento crudo

for fragmento in fragmentos: contextualizado = añadir_contexto_al_fragmento(fragmento, resumen) embedding = incrustar(contextualizado) almacenar(fragmento, embedding) # Almacena original, incrusta contextualizado """

Fragmentación Específica para Código

""" from langchain.text_splitter import Language, RecursiveCharacterTextSplitter

División consciente del lenguaje

python_splitter = RecursiveCharacterTextSplitter.from_language( language=Language.PYTHON, chunk_size=1000, chunk_overlap=200 )

Respeta los límites de funciones/clases

chunks = python_splitter.split_text(codigo_python) """

Formación de Memoria en Segundo Plano

Procesar recuerdos de forma asíncrona para mejor calidad

Cuándo usarlo: Deseas una mayor tasa de recuperación sin ralentizar las interacciones

FORMACIÓN DE MEMORIA EN SEGUNDO PLANO:

""" La extracción de memoria en tiempo real ralentiza las conversaciones y añade complejidad a las llamadas de herramientas del agente. El procesamiento en segundo plano después de las conversaciones produce recuerdos de mayor calidad.

Patrón: Formación de memoria subconsciente """

Procesamiento en Segundo Plano con LangGraph

""" from langgraph.graph import StateGraph from langgraph.checkpoint.postgres import PostgresSaver

async def procesador_memoria_segundo_plano(thread_id: str): # Ejecutar después de que la conversación termine o esté inactiva conversacion = await cargar_conversacion(thread_id)

# Extraer ideas sin presión de tiempo
insights = await llm.invoke('''
    Analiza esta conversación y extrae:
    1. Hechos clave aprendidos sobre el usuario
    2. Preferencias del usuario reveladas
    3. Tareas completadas o pendientes
    4. Patrones en el comportamiento del usuario

    Sé minucioso - esto se ejecuta en segundo plano.

    Conversación:
    {conversacion}
''')

# Almacenar en memoria a largo plazo
for insight in insights:
    await memory.semantic.upsert(
        namespace="ideas_usuario",
        key=generar_clave(insight),
        content=insight,
        metadata={"hilo_origen": thread_id}
    )

Activar al final de la conversación o tiempo de inactividad

@al_inactividad_conversacion(tiempo_limite_minutos=5) async def procesar_conversacion(thread_id): await procesador_memoria_segundo_plano(thread_id) """

Consolidación de Memoria (Como el Sueño)

"""

Consolidar y deduplicar recuerdos periódicamente

async def consolidar_recuerdos(user_id: str): # Obtener todos los recuerdos del usuario recuerdos = await memory.semantic.list( namespace="ideas_usuario", filter={"user_id": user_id} )

# Encontrar recuerdos similares (posibles duplicados)
clusters = agrupar_por_similitud(recuerdos, umbral=0.9)

# Fusionar recuerdos similares
for cluster in clusters:
    if len(cluster) > 1:
        fusionado = await llm.invoke(f'''
            Consolida estos recuerdos relacionados en uno:
            {cluster}

            Conserva toda la información importante.
        ''')
        await memory.semantic.upsert(
            namespace="ideas_usuario",
            key=generar_clave(fusionado),
            content=fusionado
        )
        # Eliminar originales
        for antiguo in cluster:
            await memory.semantic.delete(antiguo.id)

"""

Patrón de Decaimiento de Memoria

Olvidar recuerdos antiguos e irrelevantes

Cuándo usarlo: La memoria crece mucho, la recuperación se ralentiza

DECAIMIENTO DE MEMORIA:

""" No todos los recuerdos deben vivir para siempre:

• Las preferencias antiguas pueden estar obsoletas
• Los detalles de las tareas pierden relevancia
• Los recuerdos contradictorios confunden la recuperación

Implementar un decaimiento inteligente basado en:

• Actualidad (¿cuándo se creó/accedió?)
• Frecuencia (¿con qué frecuencia se recupera?)
• Importancia (¿es un hecho central o un detalle?) """

Decaimiento Basado en Tiempo

""" from datetime import datetime, timedelta

async def decaer_recuerdos_antiguos(espacio_nombres: str, max_dias_edad: int): corte = datetime.now() - timedelta(days=max_dias_edad)

recuerdos_antiguos = await memory.episodic.list(
    namespace=espacio_nombres,
    filter={"ultimo_acceso": {"$lt": corte.isoformat()}}
)

for mem in recuerdos_antiguos:
    # Eliminación suave (marcar como archivado)
    await memory.episodic.update(
        id=mem.id,
        metadata={"archivado": True, "archivado_en": datetime.now().isoformat()}
    )
    # Opcional: eliminar después de un período de gracia

"""