Une conception de mémoire à quatre couches pour des agents d’IA avec traçabilité
Séparation de la mémoire à court et long terme, quatre niveaux (tampon, épisodique, sémantique, procédural), choix de stockage, écritures asynchrones, et mécanismes de protection contre le poisonement de la mémoire.
Imaginez un agent qui échoue à l’étape quatorze d’un long flux de travail. Ce n’est pas la faute du modèle ; l’agent ne sait tout simplement plus ce qu’il a déjà fait. L’échec suit un schéma familier : les appels aux outils sont corrects et les prompts solides, mais dès que la conversation dépasse une certaine longueur invisible, l’agent commence à faire des boucles. Il répète des questions déjà répondues, récupère des données qu’il possède déjà et se contredit entre l’étape six et l’étape douze, comme si les cinq premières étapes n’avaient jamais eu lieu.
Remplacer le modèle par un autre de type frontier ne change que rarement cela. Le goulot d’étranglement réside dans l’infrastructure entourant le modèle, plus précisément dans l’absence d’une véritable architecture de mémoire. Coller les derniers N messages dans le prompt fonctionne pour un chatbot, mais cela ne marche plus lorsque l’agent doit exécuter un flux de travail de vingt étapes et se souvenir, à l’étape dix-sept, de ce qu’il a décidé à l’étape trois.
Un modèle qui permet de résoudre ce problème est une pile mémoire à quatre couches, avec un espace de stockage adapté derrière chaque couche. Ce qui suit explique la fonction de chacune des couches, la manière dont les équipes de GitHub, Anthropic et Berkeley décrivent leurs versions respectives, ainsi que des exemples en Python pour chaque composant.
Une fenêtre de contexte n’est pas de la mémoire
Une fenêtre de contexte correspond à une mémoire de travail : de la RAM, et non d’un disque dur. Chaque token qu’elle contient a un coût. Remplir une fenêtre de 128 K de tokens avec l’historique complet d’une conversation semble sécurisé, mais cela devient problématique lorsqu’on utilise un agent nécessitant de nombreux outils. Un flux de travail de trente étapes, incluant des appels d’outils, des observations et des traces de raisonnement, consomme rapidement 100 K de tokens ; lorsque la fenêtre est pleine, le contenu le plus ancien est supprimé en premier, ce qui correspond généralement aux contraintes initiales et à la formulation de la tâche par l’utilisateur.
Un problème plus subtil est encore pire. Une étude menée par des chercheurs de Stanford et de l’UC Berkeley, publiée en 2024 dans une revue ACL sous le titre « Lost in the Middle: How Language Models Use Long Contexts », a révélé que les performances des modèles de langage suivent une courbe en forme de U au sein de la fenêtre de contexte. Ces modèles fonctionnent bien avec les informations situées au début et à la fin du contexte, mais se montrent nettement moins performants face aux informations enfouies au milieu. Cet effet a été observé sur plusieurs architectures, tailles de modèles et longueurs de contexte, avec des fenêtres testées allant de 4K à 32K tokens. Une étude de suivi publiée en 2025 par le MIT a proposé une explication mathématique, affirmant que le masquage de l’attention causale ainsi que l’encodage de position rotatif réduisent tous deux le poids accordé aux tokens éloignés des extrémités de la séquence.
Pour les agents, la conséquence est évidente : même si tout correspond, le modèle ne traitera pas les éléments de manière équitable. Une contrainte critique à hauteur de 30 000 tokens dans un contexte de 60 000 tokens a de fortes chances d’être pratiquement ignorée.
Le modèle oublie parce que le système lui fournit plus d’informations qu’il ne peut traiter de manière fiable, et rien ne contrôle ce qui entre dans son champ de perception.
La division de base : mémoire à court terme et mémoire à long terme
Au-delà de cette taxinomie en quatre niveaux, il existe une division plus simple que la plupart des ingénieurs ont en tête. La mémoire dans les systèmes d’IA se divise en mémoire à court terme et mémoire à long terme, et la prise de conscience de leurs limites chacune motive l’élaboration d’une architecture plus fine.
Mémoire à court terme est limitée à une session et n’existe que pour une seule interaction. Elle contient ce dont l’agent a besoin en ce moment : l’état de la tâche actuelle, l’étape à laquelle il se trouve, la dernière sortie d’outil et le trace de raisonnement en cours de construction. En pratique, il s’agit de la fenêtre de contexte, qui est très dynamique, constamment mise à jour, coûteuse par token et effacée à la fin de la session.
La documentation de LangChain l’exprime bien en décrivant la mémoire à court terme comme un espace de stockage pour l’historique des conversations, les sorties d’outil et l’état de raisonnement temporaire. Ce qu’elle omet de mentionner, c’est que c’est aussi le premier élément à être compromis à grande échelle. Dans un flux de travail comprenant vingt étapes, les sorties intermédiaires, les réflexions, les résultats de recherche et les réponses des outils s’accumulent et se disputent le même espace limité. Cela ressemble à essayer de garder un plan complexe en tête tandis que quelqu’un n’arrête pas d’intervenir avec de nouvelles informations.
La mémoire à court terme comporte deux composantes pratiques :
- L’état de la tâche représente le suivi en temps réel des progrès : l’étape actuelle, les sous-tâches en attente, l’état d’exécution des outils ainsi que la coordination entre les agents. Un agent de planification de voyage conserve l’destination, les limites budgétaires, les vols explorés et les hôtels sélectionnés afin de ne pas refaire le travail des étapes précédentes lorsqu’il atteint l’étape huit.
- Les sorties intermédiaires sont des résultats de raisonnement générés au fur et à mesure : les résultats des outils, les résumés produits, les sorties SQL, les extraits de recherche et les notes temporaires. L’équipe d’ingénierie d’Anthropic a identifié un schéma récurrent dans les systèmes en production, selon lequel les agents qui conservent des notes en dehors de la fenêtre de contexte et les consultent ultérieurement obtiennent de meilleurs résultats sur des tâches longues que ceux qui tentent de garder tout dans le contexte. La prise de notes constitue elle-même une stratégie de mémoire à court terme.
Mémoire à long terme : elle survit aux sessions. Elle permet à un agent de se rappeler qu’un utilisateur préfère des réponses concises, que le déploiement de la semaine dernière a échoué en raison d’un problème de migration, ou que le schéma de la base de données a changé en janvier. LangChain la définit comme une mémoire qui persiste entre les threads et les sessions ; en termes pratiques, ces données ont besoin d’une base de données, et non d’une fenêtre de contexte.
Elle comporte deux types fondamentaux :
- Préférences de l’utilisateur : ce sont des signaux de personnalisation qui s’accumulent avec le temps, incluant le style d’écriture, le ton, le niveau technique, les choix d’outils et les habitudes de mise en forme. Un assistant d’écriture qui se souvient que vous souhaitez toujours du markdown et un niveau technique élevé devient utile d’une manière que aucun système sans état ne peut égaler, quel que soit la qualité de son modèle.
De nombreuses équipes commettent les mêmes erreurs. Elles utilisent une mémoire à court terme en remplissant la fenêtre de contexte, et une mémoire à long terme en créant un stockage vectoriel. Les deux approches sont erronées, ou du moins incomplètes. La première échoue à grande échelle pour les raisons mentionnées ci-dessus. La seconde ne fonctionne pas, car la recherche de similarité est un mauvais moyen pour récupérer des journaux de session, des préférences ou des règles de comportement, et chaque type de mémoire à long terme nécessite un backend différent. La conception en quatre couches représente une version plus précise de cette séparation court/long terme, en associant chaque type de mémoire au stockage qu’il requiert.
Quatre types de mémoire, une taxinomie
Les équipes qui développent des systèmes d’IA en production sont parvenues indépendamment à la même taxonomie. Les noms diffèrent, mais la structure reste identique.
Le projet MemGPT, présenté par Charles Packer et ses collègues de l’UC Berkeley en 2023, a popularisé l’idée de traiter la fenêtre de contexte d’un LLM comme une RAM, c’est-à-dire quelque chose de limité mais précieux. Un système d’exploitation étend la mémoire grâce au pagination et à l’stockage sur disque, et les frameworks d’agents peuvent simuler de la même manière une mémoire à long terme en déplaçant des contextes anciens vers un stockage plus volumineux et en les rappelant lorsque c’est nécessaire. Cette idée est ensuite devenue la base de Letta.
L’équipe LangMem de LangChain, le framework ReMem de Google DeepMind ainsi que l’architecture de production de Mem0 décrivent toutes des variantes de quatre types :
- La mémoire de travail correspond à tout ce qui est présent dans le contexte en temps réel : les différents échanges de la conversation, les documents consultés et les résultats générés par les outils. Elle est rapide, limitée et coûteuse par token.
- La mémoire épisodique stocke des enregistrements des interactions passées, comme ce que l’agent a fait pour un utilisateur donné mardi dernier et quelles en ont été les conséquences. Elle est chronologique et liée à une session, généralement conservée dans un stockage rapide de type clé-valeur ou une base de données SQL, et est consultée en fonction de la récente utilisation ou de la similarité sémantique.
- La mémoire sémantique conserve des faits généraux, des connaissances sectorielles ainsi que des informations acquises au fil du temps. C’est la couche RAG, sollicitée grâce à la similarité des embeddings chaque fois qu’une tâche nécessite des connaissances sectorielles ou des résultats antérieurs.
Ces types correspondent à la science cognitive, ce qui explique pourquoi les équipes en viennent à les adopter. Ils correspondent aux systèmes de mémoire épisodique, sémantique et procédurale en neurosciences humaines, et la mémoire de travail en tant que fenêtre contextuelle n’est pas qu’une métaphore : c’est l’analogie fonctionnelle la plus proche qui existe.
Sélection du stockage : SQL, vecteur ou graphe
Quelle base de données doit stocker la mémoire ? Avant d’écrire du code, la réponse honnête est qu’elle varie en fonction du type de mémoire concerné.
Bases de données vectorielles (Weaviate, Pinecone, Qdrant, Chroma) conviennent à la mémoire sémantique, car elles permettent des recherches de similarité sur des embeddings de manière rapide à grande échelle, et la plupart offrent désormais une récupération hybride qui combine BM25 avec les vecteurs, surpassant clairement les recherches vectorielles pures sur du texte. Elles manquent de raisonnement temporel et de requêtes relationnelles, et deux faits contradictoires présentant des embeddings presque identiques peuvent vous induire en erreur. Gardez à l’écart d’elles les enregistrements épisodiques et les traces d’audit structurées.
Bases de données SQL telles que PostgreSQL et SQLite conviennent à la mémoire épisodique. Les requêtes sont structurées par ID d’utilisateur, session, type de tâche ou horodatage, le volume d’écriture est faible, et les garanties ACID sont importantes lorsque un agent enregistre ce qu’il a fait et quand. PostgreSQL avec l’extension pgvector offre des requêtes relationnelles et une recherche vectorielle dans un seul système. Pour moins de dizaines de millions d’enregistrements, c’est le choix pratique, avec des opérations familières, un hébergement abordable et SQL pour le débogage.
Bases de données graphiques telles que Neo4j ne conviennent qu’à un système de mémoire lorsque les données épisodiques ou sémantiques possèdent une structure relationnelle complexe nécessitant des requêtes à plusieurs étapes, comme savoir qui a rapporté quoi à qui il y a trois mois lors de la rédaction d’un document. Les simples enregistrements de tâches d’agent ne remplissent pas ces critères. De nombreuses équipes recourent à des bases graphiques avant de se rendre compte qu’elles avaient en réalité besoin de SQL.
Le modèle qui prévaut dans les déploiements en production est un stockage relationnel (SQL ou PostgreSQL) pour les épisodes et les profils d’utilisateurs, un stockage vectoriel pour les recherches sémantiques, ainsi qu’un état en mémoire pour le contexte de travail ; PostgreSQL associé à pgvector constitue la solution hybride idéale pour les équipes souhaitant réduire la charge opérationnelle. Choisissez le niveau de stockage en fonction des fonctions à assumer par le type de mémoire, et ne vectorisez pas tout.
Niveau 1 : mémoire tampon avec synthèse
La forme la plus basique consiste en une file d’attente limitée à un nombre fixe de messages :
from collections import deque
from typing import List, Dict
class BufferMemory:
def __init__(self, max_turns: int = 10):
# Each turn = one user message + one assistant message
self.history: deque = deque(maxlen=max_turns * 2)
def add_message(self, role: str, content: str):
self.history.append({"role": role, "content": content})
def get_context(self) -> List[Dict]:
return list(self.history)
def token_estimate(self) -> int:
total_chars = sum(len(m["content"]) for m in self.history)
return total_chars // 4 # rough approximation
def clear(self):
self.history.clear()
Cela fonctionne pour des échanges de courte durée. Lorsque les conversations durent plus longtemps, les messages anciens disparaissent progressivement à la fin et l’agent perd le fil. Une approche plus efficace, que LangMem qualifie de méthode "hot path", consiste à résumer les messages anciens avant de les supprimer :
import os
from collections import deque
from typing import List, Dict
from openai import OpenAI
class SummarizingBufferMemory:
def __init__(self, max_turns: int = 8, summary_batch: int = 4):
self.client = OpenAI(
base_url="https://openrouter.ai/api/v1",
api_key=os.environ["OPENROUTER_API_KEY"],
)
self.model = "deepseek/deepseek-v4-flash-0731"
self.recent: deque = deque(maxlen=max_turns * 2)
self.rolling_summary: str = ""
self.summary_batch = summary_batch
def add_message(self, role: str, content: str):
if len(self.recent) == self.recent.maxlen:
self._compress_oldest()
self.recent.append({"role": role, "content": content})
def _compress_oldest(self):
batch = [self.recent.popleft() for _ in range(min(self.summary_batch * 2, len(self.recent)))]
text = "\n".join(f"{m['role']}: {m['content']}" for m in batch)
response = self.client.chat.completions.create(
model=self.model,
max_tokens=300,
messages=[
{
"role": "system",
"content": "You compress conversation history. Reply with the summary only.",
},
{
"role": "user",
"content": (
"Summarize this conversation segment in 2-4 sentences. "
"Preserve any decisions made, constraints stated, and conclusions reached.\n\n"
f"{text}"
),
},
],
)
new_summary = response.choices[0].message.content.strip()
if self.rolling_summary:
self.rolling_summary = f"{self.rolling_summary} | {new_summary}"
else:
self.rolling_summary = new_summary
def get_context(self) -> List[Dict]:
context = []
if self.rolling_summary:
context.append({
"role": "system",
"content": f"[Prior conversation summary: {self.rolling_summary}]",
})
context.extend(list(self.recent))
return context
LangMem sépare la synthèse du chemin critique, qui s’exécute en temps réel et ajoute un délai pour l’utilisateur, de la synthèse en arrière-plan qui a lieu après la session. Pour les applications destinées aux utilisateurs, l’option en arrière-plan est presque toujours préférable : les résumés s’améliorent car toute la conversation peut être analysée, et personne n’attend. Pour les agents autonomes fonctionnant sans surveillance, le chemin critique est acceptable puisqu’il n’y a de toute façon aucun utilisateur qui attend.
Niveau 2 : mémoire épisodique qui vérifie ses propres citations
Ici, de nombreuses équipes échouent : elles mettent en place des systèmes de récupération mais n’enregistrent jamais ce que l’agent a réellement fait.
Le blog technique de GitHub Copilot décrit un modèle qui mérite d’être reproduit. Chaque entrée de mémoire contient des références, c’est-à-dire des informations précises telles que des chemins de fichiers, des URL, des noms d’outils et des numéros de ligne, qui permettent de vérifier au moment de la consultation si l’information stockée est toujours valable. GitHub qualifie cela de validation just-in-time. Lorsque l’agent charge une entrée de mémoire au début d’une tâche, il vérifie que la source citée existe toujours et qu’elle confirme encore l’affirmation. Sinon, l’entrée de mémoire est corrigée ou supprimée, permettant ainsi au pool de se régénérer sans nécessiter de mise à jour hors ligne.
GitHub a constaté un effet mesurable : l’agent de codage doté de mémoire a montré une augmentation de 7 % dans le taux de fusion des demandes de pull request, soit 90 % contre 83 %.
import json
import sqlite3
from datetime import datetime, timedelta
from dataclasses import dataclass, field
@dataclass
class Episode:
task_type: str
user_query: str
outcome: str
tools_used: list
citations: list # source references for validation
duration_seconds: float
success: bool
session_id: str = ""
created_at: str = field(default_factory=lambda: datetime.utcnow().isoformat())
class EpisodicMemory:
EXPIRY_DAYS = 28 # GitHub Copilot's production default
def __init__(self, db_path: str = "episodes.db"):
self.conn = sqlite3.connect(db_path, check_same_thread=False)
self._init_schema()
def _init_schema(self):
self.conn.executescript("""
CREATE TABLE IF NOT EXISTS episodes (
id INTEGER PRIMARY KEY AUTOINCREMENT,
task_type TEXT NOT NULL,
user_query TEXT,
outcome TEXT,
tools_used TEXT,
citations TEXT,
duration_s REAL,
success INTEGER,
session_id TEXT,
created_at TEXT
);
CREATE INDEX IF NOT EXISTS idx_task_type ON episodes (task_type, success, created_at);
""")
self.conn.commit()
def record(self, ep: Episode):
self.conn.execute(
"""
INSERT INTO episodes
(task_type, user_query, outcome, tools_used, citations,
duration_s, success, session_id, created_at)
VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?)
""",
(
ep.task_type, ep.user_query, ep.outcome,
json.dumps(ep.tools_used), json.dumps(ep.citations),
ep.duration_seconds, int(ep.success),
ep.session_id, ep.created_at,
),
)
self.conn.commit()
def recall_similar(self, task_type: str, limit: int = 3) -> list[Episode]:
cutoff = (datetime.utcnow() - timedelta(days=self.EXPIRY_DAYS)).isoformat()
cursor = self.conn.execute(
"""
SELECT task_type, user_query, outcome, tools_used, citations,
duration_s, success, session_id, created_at
FROM episodes
WHERE task_type = ? AND success = 1 AND created_at > ?
ORDER BY created_at DESC
LIMIT ?
""",
(task_type, cutoff, limit),
)
return [
Episode(
task_type=r[0], user_query=r[1], outcome=r[2],
tools_used=json.loads(r[3]), citations=json.loads(r[4]),
duration_seconds=r[5], success=bool(r[6]),
session_id=r[7], created_at=r[8],
)
for r in cursor.fetchall()
]
def validate_citations(self, episode: Episode, validator_fn) -> bool:
"""
validator_fn(citation: str) -> bool
Check if cited sources are still valid (file exists, URL responds, etc.)
Return False if any citation fails; the episode should be discarded.
"""
return all(validator_fn(c) for c in episode.citations)
La durée de validité de 28 jours suit le choix opérationnel de GitHub. Les épisodes anciens ne sont pas supprimés, mais ils deviennent inaccessibles, et une requête LIMIT sur un index filtré par date est rapide et ne nécessite aucun travail de mise à jour en arrière-plan.
Niveau 3 : mémoire sémantique utilisant la recherche hybride de Weaviate
C’est le niveau RAG. La plupart des guides traitent de l’installation et de l’arrêt, tandis que les véritables problèmes en environnement de production concernent la qualité de la récupération des informations et leur provenance.
Une erreur fréquente consiste à rendre la mémoire sémantique en lecture seule. Un agent à long terme a besoin qu’elle soit en lecture-écriture : lorsque l’agent découvre un fait au cours d’une tâche, le valide et décide qu’il vaut la peine de le conserver, il doit être enregistré dans la mémoire sémantique pour les sessions futures. La deuxième erreur est de se fier uniquement à la similarité cosinus. La recherche hybride de Weaviate combine la correspondance par mots-clés BM25 avec la similarité vectorielle, ce qui lui permet généralement de surpasser chacune de ces méthodes isolément pour la récupération dans la mémoire de l’agent.
Troisièmement, et c’est ce qui ouvre la porte au empoisonnement des données, les entrées ne disposent souvent d’aucune source fiable. Chacune nécessite une source, un score de fiabilité et un identifiant de session. En l’absence de ces éléments, il est impossible d’effectuer une recherche pondérée en fonction de la fiabilité ; les entrées obsolètes ou douteuses ne peuvent pas être déclassées, et retracer une recherche erronée devient pure spéculation.
import uuid as uuid_lib
from datetime import datetime
import weaviate
from weaviate.classes.config import Configure, Property, DataType, VectorDistances
from weaviate.classes.query import MetadataQuery, HybridFusion, Filter
def embed(text: str) -> list[float]:
"""
Swap in any embedding source: OpenAI, Cohere, sentence-transformers, etc.
Returns a normalized float vector.
"""
from sentence_transformers import SentenceTransformer
_model = SentenceTransformer("all-MiniLM-L6-v2")
return _model.encode(text, normalize_embeddings=True).tolist()
class WeaviateSemanticMemory:
COLLECTION = "AgentMemory"
def __init__(self, host: str = "localhost", port: int = 8080):
self.client = weaviate.connect_to_local(host=host, port=port)
self._ensure_collection()
def _ensure_collection(self):
if self.client.collections.exists(self.COLLECTION):
return
self.client.collections.create(
name=self.COLLECTION,
# We provide our own vectors; no built-in vectorizer needed.
# Swap to Configure.Vectorizer.text2vec_openai() if you prefer managed embedding.
vectorizer_config=Configure.Vectorizer.none(),
vector_index_config=Configure.VectorIndex.hnsw(
distance_metric=VectorDistances.COSINE
),
properties=[
Property(name="content", data_type=DataType.TEXT),
Property(name="task_type", data_type=DataType.TEXT),
Property(name="source", data_type=DataType.TEXT),
Property(name="citations", data_type=DataType.TEXT),
Property(name="session_id", data_type=DataType.TEXT),
Property(name="confidence", data_type=DataType.NUMBER),
Property(name="created_at", data_type=DataType.TEXT),
],
)
def store(
self,
content: str,
task_type: str = "",
source: str = "agent",
citations: str = "",
session_id: str = "",
confidence: float = 1.0,
) -> str:
collection = self.client.collections.get(self.COLLECTION)
doc_id = str(uuid_lib.uuid4())
collection.data.insert(
properties={
"content": content,
"task_type": task_type,
"source": source,
"citations": citations,
"session_id": session_id,
"confidence": confidence,
"created_at": datetime.utcnow().isoformat(),
},
vector=embed(content),
uuid=doc_id,
)
return doc_id
def retrieve_hybrid(
self,
query: str,
n_results: int = 5,
min_confidence: float = 0.6,
task_type: str = None,
) -> list[dict]:
collection = self.client.collections.get(self.COLLECTION)
# Filter by confidence floor and optionally by task type
confidence_filter = Filter.by_property("confidence").greater_or_equal(min_confidence)
if task_type:
active_filter = (
Filter.by_property("task_type").equal(task_type) & confidence_filter
)
else:
active_filter = confidence_filter
results = collection.query.hybrid(
query=query,
vector=embed(query),
limit=n_results,
fusion_type=HybridFusion.RELATIVE_SCORE,
filters=active_filter,
return_metadata=MetadataQuery(score=True),
)
return [
{
"content": obj.properties["content"],
"score": obj.metadata.score,
"confidence": obj.properties.get("confidence", 1.0),
"source": obj.properties.get("source", ""),
"citations": obj.properties.get("citations", ""),
"uuid": str(obj.uuid),
}
for obj in results.objects
]
def update_confidence(self, doc_id: str, new_confidence: float):
collection = self.client.collections.get(self.COLLECTION)
collection.data.update(
uuid=doc_id,
properties={"confidence": new_confidence},
)
def close(self):
self.client.close()
Deux éléments dans retrieve_hybrid manquent dans de nombreuses implémentations. Le premier concerne le fait que l’endpoint hybride de Weaviate reçoit à la fois le texte brut de la requête (utilisé pour BM25) et son embedding (utilisé pour la recherche vectorielle). Les résultats basés sur des mots-clés exacts proviennent du premier, tandis que ceux fondés sur la proximité sémantique proviennent du second ; leur combinaison offre des performances supérieures à celle de chacun pris séparément. Le second élément est le filtre min_confidence, qui exclut des mémoires peu fiables du contexte, quel que soit leur degré de proximité sémantique. En réalité, il s’agit d’un mécanisme de sécurité et non simplement d’une modification de la pertinence.
La méthode update_confidence fournit le mécanisme de retour d’information. Si le raisonnement propre de l’agent contredit une mémoire récupérée au cours d’une tâche, il convient de réduire sa confiance. Si une vérification par citation le confirme, il faut l’augmenter. Le pool doit apprendre à partir de son utilisation.
Niveau 4 : mémoire procédurale et évolution des prompts
Cette couche est la plus souvent négligée. La conception initiale consiste en un fichier JSON contenant des règles que l’agent lit, suit et étend en cas d’échec. Cette méthode fonctionne mais atteint ses limites, car les règles situées dans un fichier séparé se disputent l’espace de contexte et nécessitent une récupération explicite.
Les ingénieurs de LangMem chez LangChain ont proposé une idée intéressante à retenir : en environnement de production, la mémoire procédurale fait partie intégrante du prompt système lui-même. Elle n’est ni stockée à côté du prompt ni récupérée à partir de celui-ci ; elle en constitue le contenu. Chaque fois qu’un échec apprend à l’agent une nouvelle règle de conduite, le prompt lui-même est réécrit, pratique qu’ils désignent sous le nom d’optimisation du prompt en tant que mémoire procédurale. Les comportements appris deviennent alors des instructions en langage naturel que le modèle lit au début de chaque interaction.
La version plus simple basée sur un fichier JSON est celle par laquelle la plupart des équipes devraient commencer :
import json
import os
import re
from datetime import datetime, timedelta, timezone
from openai import OpenAI
MODEL = "deepseek/deepseek-v4-flash-0731"
client = OpenAI(
base_url="https://openrouter.ai/api/v1",
api_key=os.environ["OPENROUTER_API_KEY"],
)
class ProceduralMemory:
def __init__(self, rules_path: str = "procedural_rules.json"):
self.rules_path = rules_path
self.rules: list[dict] = self._load()
def _load(self) -> list[dict]:
try:
with open(self.rules_path) as f:
return json.load(f)
except FileNotFoundError:
return []
def add_rule(self, situation: str, action: str, reason: str, confidence: float = 1.0):
self.rules.append({
"situation": situation,
"action": action,
"reason": reason,
"confidence": float(confidence),
"added_at": datetime.now(timezone.utc).isoformat(),
"trigger_count": 0,
})
self._save()
def get_applicable_rules(self, context: str, min_confidence: float = 0.7) -> list[dict]:
relevant = []
ctx_lower = context.lower()
dirty = False
for rule in self.rules:
if rule.get("confidence", 1.0) < min_confidence:
continue
keywords = rule["situation"].lower().split()
if not keywords:
continue
hits = sum(1 for kw in keywords if kw in ctx_lower)
if hits >= max(1, len(keywords) // 3):
rule["trigger_count"] = rule.get("trigger_count", 0) + 1
relevant.append(rule)
dirty = True
if dirty:
self._save()
relevant.sort(key=lambda r: r.get("confidence", 1.0), reverse=True)
return relevant[:5]
def prune_stale(self, max_age_days: int = 60, min_triggers: int = 2):
cutoff = (datetime.now(timezone.utc) - timedelta(days=max_age_days)).isoformat()
self.rules = [
r for r in self.rules
if r.get("added_at", "") > cutoff or r.get("trigger_count", 0) >= min_triggers
]
self._save()
def _save(self):
tmp = f"{self.rules_path}.tmp"
with open(tmp, "w") as f:
json.dump(self.rules, f, indent=2)
os.replace(tmp, self.rules_path)
def _parse_json(text: str) -> dict:
text = text.strip()
fenced = re.search(r"```(?:json)?\s*(.*?)```", text, re.S)
if fenced:
text = fenced.group(1).strip()
return json.loads(text)
def extract_rule_from_failure(failure_trace: str, memory: ProceduralMemory):
response = client.chat.completions.create(
model=MODEL,
max_tokens=250,
response_format={"type": "json_object"},
messages=[
{
"role": "system",
"content": "You return only a JSON object. No prose, no code fences.",
},
{
"role": "user",
"content": (
"A task failed. Extract one behavioral rule to prevent this failure.\n"
"Return ONLY valid JSON with keys: situation, action, reason, "
"confidence (0.0-1.0)\n"
f"Failure trace:\n{failure_trace}"
),
},
],
)
try:
rule = _parse_json(response.choices[0].message.content)
except (json.JSONDecodeError, AttributeError, TypeError):
return None
if not all(k in rule for k in ("situation", "action", "reason")):
return None
memory.add_rule(
situation=str(rule["situation"]),
action=str(rule["action"]),
reason=str(rule["reason"]),
confidence=float(rule.get("confidence", 1.0)),
)
return rule
prune_stale s’attaque à l’oubli au niveau procédural. Une règle créée après une défaillance inhabituelle et qui n’est plus jamais déclenchée devrait disparaître, tandis qu’une règle qui s’active régulièrement reste en vigueur quel que soit son âge. C’est un outil rudimentaire, mais c’est précisément à ce niveau que l’oubli est le plus facile à gérer.
Lorsqu’un agent fonctionne de manière continue, l’étape suivante naturelle consiste à intégrer dans le prompt du système lui-même les règles qui sont à la fois fiables et fréquemment utilisées. LangMem automatise ce processus grâce à une boucle d’optimisation des prompts, et une sélection manuelle convient parfaitement pour des configurations plus petites.
Relier les différents niveaux
La liste suivante montre comment les quatre niveaux s’assemblent en environnement de production. Une contrainte est incontournable : l’écriture dans la mémoire ne doit en aucun cas ralentir la réponse.
import json
from concurrent.futures import ThreadPoolExecutor
from dataclasses import dataclass, field
@dataclass
class TaskResult:
task_type: str
query: str
outcome: str
tools_used: list
citations: list
duration_seconds: float
success: bool
new_facts: list[dict] = field(default_factory=list)
session_id: str = ""
class MemoryManager:
def __init__(
self,
weaviate_host: str = "localhost",
episodic_db: str = "episodes.db",
rules_path: str = "procedural_rules.json",
):
self.buffer = SummarizingBufferMemory(max_turns=8)
self.episodic = EpisodicMemory(db_path=episodic_db)
self.semantic = WeaviateSemanticMemory(host=weaviate_host)
self.procedural = ProceduralMemory(rules_path=rules_path)
self._pool = ThreadPoolExecutor(max_workers=2, thread_name_prefix="memory_write")
def build_context(self, query: str, task_type: str) -> list[dict]:
context: list[dict] = []
# Procedural rules first: they constrain behavior throughout the task
rules = self.procedural.get_applicable_rules(query)
if rules:
rules_text = "\n".join(
f"- When '{r['situation']}': {r['action']} (reason: {r['reason']})"
for r in rules
)
context.append({"role": "user", "content": f"[Behavioral rules:\n{rules_text}]"})
# Semantic facts: domain knowledge and past discoveries
facts = self.semantic.retrieve_hybrid(query, n_results=5, min_confidence=0.6)
if facts:
facts_text = "\n".join(f"- {f['content']}" for f in facts)
context.append({"role": "user", "content": f"[Relevant knowledge:\n{facts_text}]"})
# Past episodes: outcome templates for similar tasks
episodes = self.episodic.recall_similar(task_type, limit=3)
if episodes:
ep_text = "\n".join(
f"- Outcome: {e.outcome} (tools: {', '.join(e.tools_used)})"
for e in episodes
)
context.append({"role": "user", "content": f"[Past similar tasks:\n{ep_text}]"})
# Current conversation last: the model reads this most carefully
context.extend(self.buffer.get_context())
return context
def record_turn(self, role: str, content: str):
self.buffer.add_message(role, content)
def persist(self, result: TaskResult):
# Submit to thread pool and return immediately; never block the caller
self._pool.submit(self._persist_worker, result)
def _persist_worker(self, result: TaskResult):
ep = Episode(
task_type=result.task_type,
user_query=result.query,
outcome=result.outcome,
tools_used=result.tools_used,
citations=result.citations,
duration_seconds=result.duration_seconds,
success=result.success,
session_id=result.session_id,
)
self.episodic.record(ep)
for fact in result.new_facts:
self.semantic.store(**fact)
if not result.success:
extract_rule_from_failure(result.outcome, self.procedural)
def shutdown(self):
self._pool.shutdown(wait=True)
self.semantic.close()
L’ordre du contexte dans build_context est intentionnel. Les recherches sur le biais d’attention positionnelle soutiennent l’idée que les modèles accordent plus d’importance au début et à la fin de la fenêtre. Les règles comportementales se trouvent en haut, car il s’agit de contraintes qui doivent guider tout ce qui suit. La conversation en cours se trouve en bas, car c’est ce dont le modèle a le plus besoin pour répondre ensuite. Les faits sémantiques et les épisodes passés se situent entre les deux, fournissant des références sans perturber la tâche en cours.
La ligne self._pool.submit(self._persist_worker, result) représente le choix technique le plus important de la classe. Chaque appel d’écriture renvoie immédiatement les résultats, de sorte que la réponse du LLM est déjà transmise à l’utilisateur avant même que toute écriture dans la base de données ne commence.
Trois décisions qui façonnent l’architecture
Avec quatre couches en place, trois décisions architecturales influencent tout le reste, et de nombreuses équipes en prennent au moins une de mauvaise manière.
Écritures en mode hot-path ou en arrière-plan. L’équipe LangMem a consacré beaucoup d’efforts à ce sujet. Les écritures en mode hot-path ont lieu pendant la conversation, lorsque l’agent détecte un fait digne d’être enregistré et effectue une écriture directement dans le flux de réponse. Les écritures en arrière-plan ont lieu après la fin de la session, afin d’analyser l’intégralité de la conversation pour déterminer ce qui doit être conservé. Le mode hot-path augmente la latence, car chaque appel à semantic.store() dans le cycle de réponse représente une confirmation dans la base de données que l’utilisateur doit attendre. Les écritures en arrière-plan améliorent la qualité des rappels, car il est possible d’examiner les thèmes et décisions tout au long de la conversation afin d’en stocker des résumés plus complets. Pour les applications professionnelles destinées aux utilisateurs, le mode en arrière-plan est presque toujours la meilleure solution. Le pool de threads présent dans persist() permet d’obtenir la majeure partie des avantages tout en évitant un pipeline entièrement asynchrone.
Récupération ou résumé. Ces deux approches résolvent des problèmes différents, et les équipes les confondent souvent. La récupération consiste à trouver le passage pertinent dans un grand corpus. La requête est spécifique et le corpus, volumineux et relativement stable, comme dans la documentation, les catalogues de produits ou le code. Elle échoue lorsque la requête est complexe, que les informations pertinentes sont dispersées dans de nombreux passages, ou que le corpus évolue plus rapidement que l’index d’incorporation. Le résumé consiste à compresser une conversation ou un document pour l’adapter au contexte. Il fonctionne lorsque les données d’entrée sont séquentielles, comme dans l’historique d’une conversation, et que certaines pertes sont acceptables ; il échoue lorsque les détails supprimés s’avèrent importants. En 2025, l’industrie a commencé à remplacer ces deux termes par « ingénierie de contexte » : la question n’est plus celle de savoir s’il s’agit de récupération ou de résumé, mais plutôt quelle combinaison de connaissances stockées, de passages récupérés et d’historique compressé offre le meilleur résultat.
Le contexte de cette requête d’inférence particulière. La méthodebuild_context du gestionnaire de mémoire constitue une ingénierie de contexte, car elle décide quelles des quatre sources de données lire, combien en prendre de chacune, ainsi que l’ordre d’assemblage des éléments. Provenance dès le premier jour. On ne peut pas ajouter une fonction de provenance à un système de mémoire ultérieurement. Dès la première écriture, chaque entrée doit indiquer sa source, son score de fiabilité, son identifiant de session et l’heure de création. Sans ces informations, il n’y a ni récupération pondérée en fonction de la fiabilité, ni filtrage par date de création, ni moyen d’auditer une mauvaise récupération. Les équipes qui omettent la provenance dès le début finissent par passer des semaines à l’ajouter à une base de données qui contient déjà des milliers d’entrées sans elle.
Empoisonnement de la mémoire : modèles d’attaque et défenses
C’est la section que les guides de conception mémoire ont tendance à omettre, alors qu’elle devrait figurer dans tous.
Une étude publiée sur arXiv en 2025 a présenté le cadre d’attaque MINJA, l’une des analyses publiques les plus approfondies sur le poisonage de la mémoire dans les agents LLM. Ses auteurs indiquent que, grâce à des interactions utilisateur ordinaires sans privilèges accrus, les attaquants peuvent atteindre des taux de réussite de l’injection supérieurs à 95 % en utilisant un schéma en trois étapes. Une étape de liaison relie un sujet innocent et légitime au sujet visé par l’attaquant. Une étape d’indication fournit des indices subtils qui font apparaître la mémoire nocive comme fiable. Une étape de raccourcissement progressif compresse l’injection pour en faire une empreinte minimale capable de survivre à l’extraction de la mémoire.
Ce qui distingue le poisonage de la mémoire de l’injection de prompt ordinaire, c’est que les mémoires contaminées franchissent les limites des sessions et s’accumulent. Une injection réussie lors de la première session peut influencer toutes les sessions ultérieures, indéfiniment.
Un article distinct de 2025 sur les défenses par confiance bayésienne a décrit une architecture à cinq niveaux visant à couvrir la surface d’attaque :
- Moderation des entrées au moment de l’écriture. Avant chaque écriture en mémoire, un score de confiance combiné est calculé à partir de plusieurs signaux indépendants : l’origine des données, une analyse sémantique des schémas d’instructions et la détection d’anomalies par rapport à l’historique de l’utilisateur. Aucun signal seul n’est suffisant, et les injections MINJA sont conçues pour passer les vérifications basées sur un seul signal.
- Suppression des instructions avant l’écriture. Si une donnée candidate ressemble à une commande ("toujours", "jamais", " vous devez", "ne pas"), elle doit être nettoyée ou éliminée plutôt que sauvegardée. Les données factuelles authentiques n’ont jamais besoin de donner des ordres à l’agent.
min_confidence du côté Weaviate de la couche 3 en est un exemple. Les souvenirs à faible provenance restent hors contexte, quel que soit leur degré de similarité. Dans une implémentation plus complète, le champ de confiance est mis à jour en fonction des résultats de validation : ainsi, les souvenirs validés gagnent en confiance tandis que ceux qui sont contredits la perdent.Pour la plupart des équipes, l’ordre logique consiste d’abord à appliquer des étiquettes de provenance lors de chaque écriture, puis à détecter les instructions au moment de l’écriture, enfin à effectuer une récupération pondérée en fonction de la fiabilité. La surveillance comportementale vient en dernier car elle nécessite une base de référence comportementale, et sa création requiert des semaines de données issues du fonctionnement réel.
Un dernier point : des recherches menées sur des systèmes de mémoire du monde réel contenant déjà des entrées légitimes ont montré que les conditions réelles réduisent considérablement l’efficacité des attaques par rapport aux environnements de laboratoire, car les entrées contaminées concurrencent les mémoires légitimes pour obtenir des places de récupération. Cela ne justifie pas de négliger les mesures de protection ; au contraire, cela montre qu’il est essentiel de maintenir les entrées légitimes de haute qualité et à forte fiabilité, car elles constituent la première ligne de défense.
Quoi construire en premier
Tous les quatre niveaux ne sont pas nécessaires immédiatement. Pour la plupart des applications d’IA, deux d’entre eux couvrent environ 80 % des problèmes de mémoire : le Niveau 1 (tampon plus résumé) et le Niveau 3 (Weaviate avec recherche hybride, filtre de confiance et traçabilité). Ensemble, ils gèrent les deux problèmes les plus fréquents, à savoir un agent qui oublie une partie de sa tâche et un autre qui ne possède aucune connaissance de domaine. Créez-les tous les deux, comparez l’achèvement des tâches avant et après, puis jugez seulement ce qui manque.
Ajoutez une mémoire épisodique lorsque vous vous demandez pourquoi l’agent a oublié ce qu’il a accompli lors de la session précédente, car cette question est le signal. Ne l’ajoutez pas plus tôt. Ajoutez une mémoire procédurale lorsque le même type d’erreur se reproduit à travers différentes tâches et que l’ingénierie des prompts ne parvient pas à y remédier, car un schéma de comportement nécessite une mémoire comportementale.
PostgreSQL avec pgvector constitue le choix par défaut judicieux pour la plupart des équipes. Une instance de base de données avec CREATE EXTENSION IF NOT EXISTS vector offre la conformité ACID, un SQL flexible, une recherche de similarité vectorielle et un système de notation hybride ; selon certaines estimations, son hébergement coûte environ 45 dollars par mois pour 10 000 mémoires. Une base de données vectorielle indépendante n’est rentable que lorsque la quantité de mémoires sémantiques devient si importante que l’index HNSW de PostgreSQL devient un frein, ce qui n’arrive presque jamais aux équipes.
Rendez chaque écriture de mémoire asynchrone. La réponse à l’utilisateur ne devrait pas dépendre d’une confirmation dans la base de données ; confiez donc cette tâche à un pool de threads et passez à autre chose.
Évitez les stockages vectoriels en mémoire vive en production. Une implémentation sans stockage persistant perd toute connaissance accumulée dès la première redémarrage du serveur, une erreur que peu de personnes commettent à deux reprises.