Que fait Systèmes de Mémoire pour Agents ?
La mémoire est la pierre angulaire des agents intelligents. Sans elle, chaque interaction commence de zéro. Ce skill couvre l'architecture de la mémoire des agents : mémoire à court terme (fenêtre de contexte), mémoire à long terme (bases vectorielles) et les architectures cognitives qui les organisent.
Point clé : la mémoire n'est pas seulement du stockage - c'est de la récupération. Un million de faits stockés ne signifient rien si vous ne pouvez pas trouver le bon. Les stratégies de découpage, d'embedding et de récupération déterminent si votre agent se souvient ou oublie.
Le domaine est fragmenté, avec une terminologie inconsistante. Nous utilisons le cadre d'architecture cognitive CoALA : mémoire sémantique (faits), mémoire épisodique (expériences) et mémoire procédurale (savoir-faire).
Principes
- Qualité de la mémoire = qualité de la récupération, pas quantité de stockage
- Découper pour la récupération, pas pour le stockage
- L'isolation du contexte est l'ennemie de la mémoire
- Le bon type de mémoire pour la bonne information
- Faire dégrader les vieux souvenirs - tout ne doit pas être éternel
- Tester la précision de récupération avant la production
- La formation de mémoire en arrière-plan bat celle en temps réel
Capacités
- agent-memory
- long-term-memory
- short-term-memory
- working-memory
- episodic-memory
- semantic-memory
- procedural-memory
- memory-retrieval
- memory-formation
- memory-decay
Périmètre
- vector-database-operations → data-engineer
- rag-pipeline-architecture → llm-architect
- embedding-model-selection → ml-engineer
- knowledge-graph-design → knowledge-engineer
Outillage
Memory_frameworks
- LangMem (LangChain) - Quand : Agents LangGraph avec mémoire persistante. Note : Types de mémoire sémantique, épisodique, procédurale.
- MemGPT / Letta - Quand : Gestion de contexte virtuelle, mémoire de type OS. Note : Niveaux de mémoire hiérarchiques, pagination automatique.
- Mem0 - Quand : Couche mémoire utilisateur pour la personnalisation. Note : Conçu pour les préférences et l'historique utilisateur.
Vector_stores
- Pinecone - Quand : Géré, échelle entreprise (milliards de vecteurs). Note : Meilleure performance de requête, coût le plus élevé.
- Qdrant - Quand : Filtrage de métadonnées complexe, open-source. Note : Basé sur Rust, excellent filtrage.
- Weaviate - Quand : Recherche hybride, fonctionnalités de graphe de connaissances. Note : Interface GraphQL, bon pour les relations.
- ChromaDB - Quand : Prototypage, applications petites/moyennes. Note : Convivial pour les développeurs, ~20ms p50 à 100K vecteurs.
- pgvector - Quand : Utilise déjà PostgreSQL, configuration plus simple. Note : Bon pour <1M vecteurs, outillage familier.
Embedding_models
- OpenAI text-embedding-3-large - Quand : Meilleure qualité, 3072 dimensions. Note : 0,13 $/1M tokens.
- OpenAI text-embedding-3-small - Quand : Bon équilibre, 1536 dimensions. Note : 0,02 $/1M tokens, 5x moins cher.
- nomic-embed-text-v1.5 - Quand : Open-source, déploiement local. Note : 768 dimensions, bonne qualité.
- all-MiniLM-L6-v2 - Quand : Léger, embedding local rapide. Note : 384 dimensions, latence la plus faible.
Modèles
Architecture des types de mémoire
Choisir le bon type de mémoire pour différentes informations
Quand l'utiliser : Concevoir un système de mémoire pour agent
ARCHITECTURE DES TYPES DE MÉMOIRE (Cadre CoALA) :
""" Trois types de mémoire à des fins différentes :
-
Mémoire sémantique : Faits et connaissances
- Ce que vous savez sur le monde
- Préférences utilisateur, connaissances du domaine
- Stockée dans des profils (structurés) ou des collections (non structurés)
-
Mémoire épisodique : Expériences et événements
- Ce qui s'est passé (événements horodatés)
- Conversations passées, résultats de tâches
- Utilisée pour apprendre de l'expérience
-
Mémoire procédurale : Comment faire les choses
- Règles, compétences, workflows
- Souvent implémentée comme des exemples few-shot
- "Comment ai-je résolu cela auparavant ?" """
Implémentation LangMem
""" from langmem import MemoryStore from langgraph.graph import StateGraph
Initialiser le magasin de mémoire
memory = MemoryStore( connection_string=os.environ["POSTGRES_URL"] )
Mémoire sémantique : profil utilisateur
await memory.semantic.upsert( namespace="user_profile", key=user_id, content={ "name": "Alice", "preferences": ["mode sombre", "réponses concises"], "expertise_level": "developeur", } )
Mémoire épisodique : interaction passée
await memory.episodic.add( namespace="conversations", content={ "timestamp": datetime.now(), "summary": "A aidé à déboguer un problème d'authentification", "outcome": "résolu", "key_insights": ["L'expiration du jeton était la cause racine"], }, metadata={"user_id": user_id, "topic": "debugging"} )
Mémoire procédurale : modèle appris
await memory.procedural.add( namespace="skills", content={ "task_type": "debug_auth", "steps": ["Vérifier l'expiration du jeton", "Vérifier le flux d'actualisation"], "example_interaction": few_shot_example, } ) """
Récupération de mémoire à l'exécution
""" async def prepare_context(user_id, query): # Obtenir le profil utilisateur (sémantique) profile = await memory.semantic.get( namespace="user_profile", key=user_id )
# Trouver les expériences passées pertinentes (épisodique)
similar_experiences = await memory.episodic.search(
namespace="conversations",
query=query,
filter={"user_id": user_id},
limit=3
)
# Trouver les compétences pertinentes (procédurale)
relevant_skills = await memory.procedural.search(
namespace="skills",
query=query,
limit=2
)
return {
"profile": profile,
"past_experiences": similar_experiences,
"relevant_skills": relevant_skills,
}
"""
Modèle de sélection de base vectorielle
Choisir la bonne base de données vectorielle pour votre cas d'utilisation
Quand l'utiliser : Configurer un stockage mémoire persistant
SÉLECTION DE BASE VECTORIELLE :
""" Matrice de décision :
| Pinecone | Qdrant | Weaviate | ChromaDB | pgvector | |
|---|---|---|---|---|---|
| Échelle | Milliards | 100M+ | 100M+ | 1M | 1M |
| Géré | Oui | Les deux | Les deux | Soi-même | Soi-même |
| Filtrage | Basique | Meilleur | Bon | Basique | SQL |
| Hybride | Non | Oui | Meilleur | Non | Oui |
| Coût | Élevé | Moyen | Moyen | Gratuit | Gratuit |
| Latence | 5ms | 7ms | 10ms | 20ms | 15ms |
| """ |
Pinecone (Échelle Entreprise)
""" from pinecone import Pinecone
pc = Pinecone(api_key=os.environ["PINECONE_API_KEY"]) index = pc.Index("agent-memory")
Upsert avec métadonnées
index.upsert( vectors=[ { "id": f"memory-{uuid4()}", "values": embedding, "metadata": { "user_id": user_id, "timestamp": datetime.now().isoformat(), "type": "episodic", "content": memory_text, } } ], namespace=namespace )
Requête avec filtre
results = index.query( vector=query_embedding, filter={"user_id": user_id, "type": "episodic"}, top_k=5, include_metadata=True ) """
Qdrant (Filtrage Complexe)
""" from qdrant_client import QdrantClient from qdrant_client.models import PointStruct, Filter, FieldCondition
client = QdrantClient(url="http://localhost:6333")
Filtrage complexe avec Qdrant
results = client.search( collection_name="agent_memory", query_vector=query_embedding, query_filter=Filter( must=[ FieldCondition(key="user_id", match={"value": user_id}), FieldCondition(key="type", match={"value": "semantic"}), ], should=[ FieldCondition(key="topic", match={"any": ["auth", "security"]}), ] ), limit=5 ) """
ChromaDB (Prototypage)
""" import chromadb
client = chromadb.PersistentClient(path="./memory_db") collection = client.get_or_create_collection("agent_memory")
Simple et rapide pour les prototypes
collection.add( ids=[str(uuid4())], embeddings=[embedding], documents=[memory_text], metadatas=[{"user_id": user_id, "type": "episodic"}] )
results = collection.query( query_embeddings=[query_embedding], n_results=5, where={"user_id": user_id} ) """
Modèle de stratégie de découpage (Chunking)
Fractionner les documents en morceaux récupérables
Quand l'utiliser : Traiter des documents pour le stockage mémoire
STRATÉGIES DE DÉCOUPAGE :
""" Le dilemme du découpage :
- Trop grand : Le vecteur perd en spécificité
- Trop petit : Perte de contexte
La taille de chunk optimale dépend de :
- Le type de document (code vs prose vs données)
- Les modèles de requête (factuelle vs exploratoire)
- Le modèle d'embedding (chacun a un point idéal)
Guide général : 256-512 tokens pour la plupart des cas d'utilisation """
Découpage de Taille Fixe (Référence)
""" from langchain.text_splitter import RecursiveCharacterTextSplitter
splitter = RecursiveCharacterTextSplitter( chunk_size=500, # Caractères chunk_overlap=50, # Le chevauchement évite de couper les phrases separators=["\n\n", "\n", ". ", " ", ""] # Ordre de priorité )
chunks = splitter.split_text(document) """
Découpage Sémantique (Meilleure Qualité)
""" from langchain_experimental.text_splitter import SemanticChunker from langchain_openai import OpenAIEmbeddings
Fractionne en fonction de la similarité sémantique
splitter = SemanticChunker( embeddings=OpenAIEmbeddings(), breakpoint_threshold_type="percentile", breakpoint_threshold_amount=95 )
chunks = splitter.split_text(document) """
Découpage Sensible à la Structure (Documents Hiérarchiques)
""" from langchain.text_splitter import MarkdownHeaderTextSplitter
Respecte la structure du document
splitter = MarkdownHeaderTextSplitter( headers_to_split_on=[ ("#", "Header 1"), ("##", "Header 2"), ("###", "Header 3"), ] )
chunks = splitter.split_text(markdown_doc)
Chaque chunk a des métadonnées d'en-tête pour le contexte
"""
Découpage Contextuel (Approche d'Anthropic)
"""
Ajouter du contexte à chaque chunk avant l'embedding
Réduit les échecs de récupération de 35%
def add_context_to_chunk(chunk, document_summary): context_prompt = f''' Résumé du document : {document_summary}
Ce qui suit est un chunk de ce document :
{chunk}
'''
return context_prompt
Embeddre le chunk contextualisé, pas le chunk brut
for chunk in chunks: contextualized = add_context_to_chunk(chunk, summary) embedding = embed(contextualized) store(chunk, embedding) # Stocker l'original, embeddre le contextualisé """
Découpage Spécifique au Code
""" from langchain.text_splitter import Language, RecursiveCharacterTextSplitter
Fractionnement tenant compte du langage
python_splitter = RecursiveCharacterTextSplitter.from_language( language=Language.PYTHON, chunk_size=1000, chunk_overlap=200 )
Respecte les limites des fonctions/classes
chunks = python_splitter.split_text(python_code) """
Formation de Mémoire en Arrière-Plan
Traiter les mémoires de manière asynchrone pour une meilleure qualité
Quand l'utiliser : Vous souhaitez un meilleur rappel sans ralentir les interactions
FORMATION DE MÉMOIRE EN ARRIÈRE-PLAN :
""" L'extraction de mémoire en temps réel ralentit les conversations et ajoute de la complexité aux appels d'outils de l'agent. Le traitement en arrière-plan après les conversations produit des mémoires de meilleure qualité.
Modèle : Formation de mémoire subconsciente """
Traitement en Arrière-Plan avec LangGraph
""" from langgraph.graph import StateGraph from langgraph.checkpoint.postgres import PostgresSaver
async def background_memory_processor(thread_id: str): # S'exécute après la fin ou l'inactivité de la conversation conversation = await load_conversation(thread_id)
# Extraire les insights sans pression de temps
insights = await llm.invoke('''
Analysez cette conversation et extrayez :
1. Les faits clés appris sur l'utilisateur
2. Les préférences utilisateur révélées
3. Les tâches terminées ou en attente
4. Les modèles de comportement de l'utilisateur
Soyez minutieux - cela s'exécute en arrière-plan.
Conversation :
{conversation}
''')
# Stocker en mémoire à long terme
for insight in insights:
await memory.semantic.upsert(
namespace="user_insights",
key=generate_key(insight),
content=insight,
metadata={"source_thread": thread_id}
)
Déclencher à la fin de la conversation ou après un délai d'inactivité
@on_conversation_idle(timeout_minutes=5) async def process_conversation(thread_id): await background_memory_processor(thread_id) """
Consolidation de la Mémoire (Comme le Sommeil)
"""
Consolider et dédupliquer périodiquement les mémoires
async def consolidate_memories(user_id: str): # Obtenir toutes les mémoires pour l'utilisateur memories = await memory.semantic.list( namespace="user_insights", filter={"user_id": user_id} )
# Trouver les mémoires similaires (doublons potentiels)
clusters = cluster_by_similarity(memories, threshold=0.9)
# Fusionner les mémoires similaires
for cluster in clusters:
if len(cluster) > 1:
merged = await llm.invoke(f'''
Consolidez ces mémoires connexes en une seule :
{cluster}
Préservez toutes les informations importantes.
''')
await memory.semantic.upsert(
namespace="user_insights",
key=generate_key(merged),
content=merged
)
# Supprimer les originales
for old in cluster:
await memory.semantic.delete(old.id)
"""
Modèle de Dégradation de la Mémoire
Oublier les mémoires anciennes et non pertinentes
Quand l'utiliser : La mémoire devient volumineuse, la récupération ralentit
DÉGRADATION DE LA MÉMOIRE :
""" Toutes les mémoires ne devraient pas vivre éternellement :
- Les anciennes préférences peuvent être obsolètes
- Les détails des tâches perdent leur pertinence
- Les mémoires conflictuelles perturbent la récupération
Implémenter une dégradation intelligente basée sur :
- La récence (quand a-t-elle été créée/consultée ?)
- La fréquence (à quelle fréquence est-elle récupérée ?)
- L'importance (est-ce un fait fondamental ou un détail ?)"""
Dégradation Basée sur le Temps
""" from datetime import datetime, timedelta
async def decay_old_memories(namespace: str, max_age_days: int): cutoff = datetime.now() - timedelta(days=max_age_days)
old_memories = await memory.episodic.list(
namespace=namespace,
filter={"last_accessed": {"$lt": cutoff.isoformat()}}
)
for mem in old_memories:
# Suppression douce (marquer comme archivé)
await memory.episodic.update(
id=mem.id,
metadata={"archived": True, "archived_at": datetime.now().isoformat()}
)
return len(old_memories)
"""
Dégradation Basée sur la Fréquence
""" async def decay_low_frequency_memories(namespace, min_access_count): memories = await memory.semantic.list(namespace=namespace)
for mem in memories:
if mem.metadata.get("access_count", 0) < min_access_count:
# Supprimer ou archiver
await memory.semantic.delete(mem.id)
"""
Dégradation par Consolidation
"""
Au lieu de simplement supprimer, consolider les informations importantes avant de supprimer
async def consolidate_and_decay(user_id, namespace, threshold_days): # Obtenir les vieilles mémoires old_memories = await get_old_memories(user_id, namespace, threshold_days)
# Demander au LLM de résumer ce qui est important
important_info = await llm.invoke(f'''
Les mémoires suivantes sont anciennes. Résumez
les informations importantes qui devraient être préservées :
{old_memories}
Renvoyez :
1. Informations à conserver
2. Informations à supprimer
''')
# Stocker le résumé, supprimer les détails
await memory.semantic.upsert(
namespace=namespace,
key=generate_key(user_id),
content=important_info["preserve"]
)
for mem in important_info["delete"]:
await memory.semantic.delete(mem["id"])
"""