Systèmes de mémoire pour agents — court terme, long terme, mem0, Letta, RGPD

TL;DR — Un agent sans mémoire est un stagiaire amnésique. Cinq types de mémoires à distinguer : court terme (conversation courante, fenêtre de contexte), long terme vectorielle (historique interactions), épisodique (qui-quoi-quand factuel), sémantique (knowledge graph d'entités), procédural (skills/playbooks appris). Outils 2026 : mem0 (le plus pragmatique), Letta/MemGPT (mémoire OS-like hiérarchique), Zep (orienté chat history + facts), Cognee (graph), Anthropic Memory API (managed), LangGraph checkpointers (state persistant). Côté FR : RGPD = droit à l'oubli implémenté concrètement (purge ciblée par user_id), DPIA si données sensibles, jamais de PII dans embeddings non chiffrés au repos.

🧠 Mental model

┌──────────────────────────────────────────────────────────────────────┐
│                    AGENT BRAIN                                       │
│                                                                      │
│   ┌─────────────────┐         ┌─────────────────────────────────┐   │
│   │ Short-term      │ ←─────→ │ Working memory (context window) │   │
│   │ (conv history)  │         └─────────────────────────────────┘   │
│   └─────────────────┘                                                │
│                                                                      │
│   ┌─────────────────┐                                                │
│   │ Long-term       │   ┌─────────┐  ┌───────────┐  ┌─────────────┐ │
│   │  (persistant)   │ → │Episodic │  │ Semantic  │  │ Procedural  │ │
│   └─────────────────┘   │"qui/quoi│  │"entités + │  │ "skills,    │ │
│                         │ /quand" │  │ relations"│  │  playbooks" │ │
│                         └─────────┘  └───────────┘  └─────────────┘ │
│                            ↓             ↓              ↓           │
│                         vector store / graph DB / postgres          │
│                                                                      │
│   ┌─────────────────┐                                                │
│   │ Forgetting      │  ← TTL, RGPD purge, decay, contradiction      │
│   │ mechanism       │                                                │
│   └─────────────────┘                                                │
└──────────────────────────────────────────────────────────────────────┘

Analogie immobilier : sans mémoire, l'agent immo demande à chaque appel "bonjour, vous cherchez quoi ?". Avec mémoire :

• Court terme : la conversation actuelle ("3 pièces, Paris 11e").
• Épisodique : "le 12 avril, M. Durand a visité le 23 rue Faidherbe, retour négatif (vis-à-vis)".
• Sémantique : M. Durand → cherche 75-90m², 600-750k€, évite RDC, refuse vis-à-vis, conjoint télétravaille.
• Procédural : pour M. Durand → toujours envoyer photos haute résolution avant de proposer visite.

🛠️ Code minimal

mem0 en 20 lignes avec Postgres + pgvector + Claude.

# pip install mem0ai anthropic psycopg[binary]
from mem0 import Memory
from anthropic import Anthropic

config = {
    "vector_store": {
        "provider": "pgvector",
        "config": {"connection_string": "postgresql://user:pwd@localhost/agentdb", "collection_name": "memories"},
    },
    # mem0 utilise ce LLM pour EXTRAIRE et DÉDUPLIQUER les faits → Sonnet (rapide, moins cher)
    "llm": {"provider": "anthropic", "config": {"model": "claude-sonnet-4-6"}},
    "embedder": {"provider": "openai", "config": {"model": "text-embedding-3-small"}},
}

memory = Memory.from_config(config)
client = Anthropic()

def chat(user_id: str, user_msg: str) -> str:
    related = memory.search(query=user_msg, user_id=user_id, limit=5)
    ctx = "\n".join([m["memory"] for m in related["results"]])
    resp = client.messages.create(
        model="claude-opus-4-8",  # le tour conversationnel face client → le bon modèle
        max_tokens=1024,
        # ⚠️ les mémoires sont volatiles : on les met dans messages (après le breakpoint),
        # PAS dans system. Sinon on casse le prompt cache à chaque tour (cf. plus bas).
        messages=[{"role": "user", "content": f"Mémoires utilisateur:\n{ctx}\n\n{user_msg}"}],
    )
    answer = resp.content[0].text
    memory.add(messages=[{"role": "user", "content": user_msg}, {"role": "assistant", "content": answer}], user_id=user_id)
    return answer

# Usage
print(chat("u_42", "Je cherche T3 Paris 11e, max 750k€"))
# Plus tard...
print(chat("u_42", "Tu te souviens de mon budget ?"))  # → "Oui, 750k€ max"

🎬 Cas d'usage concrets

Scénario 1 — Assistant RH onboarding (mémoire épisodique + procédurale)

Qui : ETI industrielle Alsace, 600 salariés, 80 onboardings/an. Problème : l'assistant RH oublie d'une session à l'autre les préférences du nouvel arrivant (régime alimentaire cantine, télétravail souhaité, formations demandées). Résultat : nouveau collab a l'impression d'être un dossier anonyme. Solution : agent mem0 avec mémoire par employee_id. Stocke prefs (épisodique), playbooks d'onboarding par métier (procédural).

memory.add("Préfère télétravail 3j/sem, mardi-mercredi-jeudi", user_id="emp_NEW042")
memory.add("Allergie gluten, cantine doit prévoir option", user_id="emp_NEW042")
memory.add("Demande formation Kubernetes prioritaire", user_id="emp_NEW042")
# 3 mois plus tard...
memory.search("planning formation", user_id="emp_NEW042")
# → "Kubernetes prioritaire"

Gains € : satisfaction onboarding 6.2 → 8.7/10. Turnover < 6 mois -45%. Sur 80 onboardings × 25% turnover évité × 18k€ coût recrutement = 324k€/an évités. Projet facturé 55k€, ROI < 3 mois.

Scénario 2 — Agent vente B2B SaaS (mémoire sémantique + épisodique)

Qui : éditeur SaaS RH lyonnais (le même que le précédent doc), 25 SDR. Problème : interlocuteur compte change tous les 18 mois, mais le compte (entreprise) a une histoire — interactions passées, deals perdus, raisons. SDR repart de zéro à chaque renouvellement contact. Solution : mémoire double niveau : account_id (entreprise) + contact_id (personne). Agent qui synthétise l'historique avant chaque RDV.

# Mémoire compte
memory.add("Renouvellement Workday prévu T2 2027", agent_id="acc_TOTAL")
memory.add("Achète RPS uniquement si certifié ISO 27701", agent_id="acc_TOTAL")
# Mémoire contact
memory.add("Préfère échanges courts, 15 min max", user_id="contact_marie.dupont")
memory.add("Sensible aux références CAC40", user_id="contact_marie.dupont")

Gains € : taux de conversion deals +18%, taux de rebond +30%. ARR additionnel +450k€/an estimé. Projet facturé 78k€.

Scénario 3 — Assistant juridique état des dossiers (mémoire épisodique + sémantique)

Qui : cabinet d'avocats Lille, droit des affaires, 250 dossiers actifs. Problème : un dossier peut durer 18 mois, passer par 4-5 collaborateurs, avec 200+ docs. Réintégrer un dossier après absence prend 2-3h. Solution : agent Letta/MemGPT (hiérarchique : main context + recall storage + archival storage). Mémoire structurée par dossier avec entités (parties, juridictions, dates clés, jurisprudences citées).

# Letta-style
from letta import create_client, ChatMemory
client = create_client()
memory = ChatMemory(human="Avocat Me Martin", persona="Assistant dossier")
agent = client.create_agent(name="dossier_2024_4521", memory=memory)
client.send_message(agent.id, "Le 12/05, audience conciliation, partie adverse a refusé l'expertise judiciaire.")
client.send_message(agent.id, "Synthèse de la situation actuelle ?")
# → l'agent agrège short + long term, rend une synthèse en 200 mots

Gains € : 2h → 10 min de réintégration dossier. 250 dossiers × 6 réintégrations/an × 1h50 × 180€/h = 495k€/an de temps avocat libéré. Cabinet a payé 95k€.

Scénario 4 — Agent immobilier proactif (mémoire long terme + decay)

Qui : agence immobilière Paris 11e (notre end-to-end ci-dessous). Problème : un acheteur peut chercher pendant 6-12 mois. L'agent doit se rappeler ses critères, ses refus, et le contacter proactivement quand un bien matche. Solution : voir end-to-end.

Scénario 5 — Assistant souveraineté FR (mémoire chiffrée on-prem)

Qui : groupe assurance mutualiste, DSI exige "rien hors UE, rien dans le cloud US". Problème : besoin d'assistant conversationnel pour le centre d'appel, qui se souvient des sociétaires (multi-appels, dossiers ouverts) sans toucher OpenAI/Anthropic Cloud. Solution : stack 100% on-prem : Mistral Large self-hosted (vLLM sur DGX H100), Qdrant chiffré at-rest (LUKS), embeddings via Mistral-Embed local, KMS HSM Thales.

# Stack souveraine
from mistralai import Mistral  # client compatible OpenAI mais self-hosted
mistral = Mistral(base_url="https://llm.internal.mutuelle.fr", api_key=os.environ["MISTRAL_TOKEN"])
# Qdrant on-prem
from qdrant_client import QdrantClient
qdrant = QdrantClient("qdrant-1.internal", port=6333, https=True, prefer_grpc=True)
# mem0 configuré custom
mem_config = {
    "vector_store": {"provider": "qdrant", "config": {"client": qdrant, "collection_name": "societaires"}},
    "llm": {"provider": "mistral", "config": {"model": "mistral-large", "base_url": mistral.base_url}},
    "embedder": {"provider": "mistral", "config": {"model": "mistral-embed"}},
}

Gains € : pas calculable directement. Permet de passer le comité sécurité, sans quoi le projet n'existerait pas. Mission facturée 145k€ (équipe 2 personnes, 3 mois).

🛠️ Exemple end-to-end

Use case : agence immobilière, agent qui mémorise les critères + refus d'un acheteur sur 6 mois et propose proactivement quand un nouveau bien matche. Stack : mem0 + LangGraph + Postgres checkpoint + webhook listing.

# pip install mem0ai langgraph langchain-anthropic psycopg[binary] fastapi
from mem0 import Memory
from langgraph.graph import StateGraph, END
from langgraph.checkpoint.postgres import PostgresSaver
from langchain_anthropic import ChatAnthropic
from typing import TypedDict
from datetime import datetime, timedelta
from fastapi import FastAPI, BackgroundTasks
import json

# ----- Setup -----
DB_URL = "postgresql://immo:pwd@localhost/immo_agent"
checkpointer = PostgresSaver.from_conn_string(DB_URL)
checkpointer.setup()

mem_config = {
    "vector_store": {"provider": "pgvector", "config": {"connection_string": DB_URL, "collection_name": "buyer_memories"}},
    "llm": {"provider": "anthropic", "config": {"model": "claude-sonnet-4-6"}},  # extraction de faits
    "embedder": {"provider": "openai", "config": {"model": "text-embedding-3-small"}},
}
memory = Memory.from_config(mem_config)
# extraction/matching/chat : Sonnet suffit et coûte ~1.7× moins que l'Opus
# (Sonnet 4.6 = 3$/15$ par M tokens in/out, Opus 4.8 = 5$/25$).
# ⚠️ temperature n'existe plus sur Opus 4.8/4.7 (400). Sur Sonnet 4.6 il est encore accepté
# mais déprécié : pilote le comportement par le prompt, pas par temperature.
llm = ChatAnthropic(model="claude-sonnet-4-6")

# ----- State -----
class State(TypedDict):
    buyer_id: str
    user_msg: str
    new_listing: dict | None
    response: str
    proactive: bool

# ----- Nodes -----
EXTRACT_PROMPT = """Extrais les critères/contraintes acheteur de ce message. JSON list de faits courts.
Catégories: budget, surface, zone, nb_pieces, etage, exposition, refus, contraintes.
Message: {msg}
Exemple: ["budget max 750k", "refuse RDC", "min 75m²"]"""

def extract_facts_node(state: State):
    if not state.get("user_msg"):
        return {}
    resp = llm.invoke(EXTRACT_PROMPT.format(msg=state["user_msg"]))
    try:
        facts = json.loads(resp.content)
    except Exception:
        facts = []
    for f in facts:
        memory.add(f, user_id=state["buyer_id"], metadata={"source": "chat", "ts": datetime.utcnow().isoformat()})
    return {}

MATCH_PROMPT = """Tu es agent immobilier. Voici le bien et les critères acheteur stockés.
Décide si proposer ce bien (score 0-10) et rédige un message court de proposition.
Réponds JSON: {{"score": 8, "should_propose": true, "message": "..."}}.

Bien: {listing}
Critères acheteur (mémoires):
{memories}
"""

def match_listing_node(state: State):
    if not state.get("new_listing"):
        return {"response": "", "proactive": False}
    mems = memory.get_all(user_id=state["buyer_id"])
    mem_text = "\n".join([m["memory"] for m in mems["results"]])
    resp = llm.invoke(MATCH_PROMPT.format(listing=json.dumps(state["new_listing"]), memories=mem_text))
    data = json.loads(resp.content)
    if data["should_propose"] and data["score"] >= 7:
        # log proactive contact
        memory.add(f"Proposé bien {state['new_listing']['ref']} le {datetime.utcnow().date()}", user_id=state["buyer_id"])
        return {"response": data["message"], "proactive": True}
    return {"response": "", "proactive": False}

CHAT_PROMPT = """Tu es agent immobilier humain. Réponds à l'acheteur en t'appuyant sur ses critères mémorisés.
Critères:
{memories}

Message acheteur: {msg}
"""

def chat_node(state: State):
    if not state.get("user_msg"):
        return {}
    mems = memory.search(query=state["user_msg"], user_id=state["buyer_id"], limit=8)
    mem_text = "\n".join([m["memory"] for m in mems["results"]])
    resp = llm.invoke(CHAT_PROMPT.format(memories=mem_text, msg=state["user_msg"]))
    return {"response": resp.content}

def route_entry(state: State):
    if state.get("new_listing"):
        return "match"
    return "extract"

# ----- Graph -----
g = StateGraph(State)
g.add_node("extract", extract_facts_node)
g.add_node("chat", chat_node)
g.add_node("match", match_listing_node)
g.set_conditional_entry_point(route_entry, {"match": "match", "extract": "extract"})
g.add_edge("extract", "chat")
g.add_edge("chat", END)
g.add_edge("match", END)
app = g.compile(checkpointer=checkpointer)

# ----- FastAPI -----
api = FastAPI()

@api.post("/chat/{buyer_id}")
def chat(buyer_id: str, msg: str):
    cfg = {"configurable": {"thread_id": buyer_id}}
    result = app.invoke({"buyer_id": buyer_id, "user_msg": msg, "new_listing": None, "response": "", "proactive": False}, config=cfg)
    return {"response": result["response"]}

@api.post("/new_listing")
def new_listing(listing: dict, bg: BackgroundTasks):
    """Webhook quand un nouveau bien arrive. On le matche contre tous les acheteurs actifs."""
    def match_all():
        # Récupère acheteurs actifs (< 6 mois inactivité)
        buyers = ["u_42", "u_67", "u_103"]  # en vrai: query Postgres
        for bid in buyers:
            cfg = {"configurable": {"thread_id": bid}}
            res = app.invoke({"buyer_id": bid, "user_msg": "", "new_listing": listing, "response": "", "proactive": False}, config=cfg)
            if res["proactive"]:
                send_email(bid, res["response"])  # SMTP/SendGrid
    bg.add_task(match_all)
    return {"status": "queued"}

# ----- RGPD: droit à l'oubli -----
@api.delete("/buyer/{buyer_id}/memory")
def forget_buyer(buyer_id: str):
    memory.delete_all(user_id=buyer_id)
    # checkpoints LangGraph aussi
    checkpointer.delete_thread(buyer_id)
    return {"status": "purged", "buyer_id": buyer_id}

def send_email(buyer_id, message):
    print(f"[email→{buyer_id}] {message}")  # stub

Flux concret :

1. Acheteur écrit "Je cherche T3 Paris 11e, max 750k, refuse RDC, exposition sud" → extract produit 4 mémoires.
2. Plusieurs sessions sur 3 mois ajoutent : "refuse Place Voltaire (bruit)", "min 75m²", "syndic propre".
3. Webhook reçoit nouveau bien rue Faidherbe, 78m², 2e étage, 720k€, sud-est → match score 8 → email proactif.
4. Acheteur peut demander purge RGPD à tout moment.

Coût mensuel pour 200 acheteurs actifs : ~12€ embeddings + 35€ Claude + 8€ Postgres = ~55€/mois. Vs un agent immo qui passerait 30 min de relance par acheteur/mois = 100h/mois × 40€ = 4 000€. ROI 70×.

🎯 Patterns courants

Pattern "summarize-then-forget"

Court terme grossit → résumer toutes les 20 turns, garder le résumé + dernières 5 turns. Évite de saturer le contexte.

if len(conv_history) > 20:
    summary = llm.invoke(f"Résume cette conversation: {conv_history[:15]}")
    conv_history = [{"role": "system", "content": f"Résumé: {summary}"}] + conv_history[-5:]

Pattern "memory router"

Avant chaque LLM call, décider quels types de mémoire interroger (épisodique seulement vs full). Économise tokens.

def router(query):
    if "rappel" in query or "souviens" in query: return ["episodic", "semantic"]
    if "comment faire" in query: return ["procedural"]
    return ["semantic"]

Pattern "memory write-on-meaningful"

Ne pas stocker chaque turn. Filtrer : "ce turn contient-il un fait actionnable ?". mem0 fait ça nativement via un LLM filter.

Pattern "decay temporel"

Mémoires anciennes pondérées plus bas. Solution simple : score = cosine * exp(-age_days/180). Évite que des préférences obsolètes pourrissent les recommandations.

Pattern "knowledge graph hybride"

Pour entités structurées (personnes, sociétés, contrats), un graph Neo4j + résumés vectoriels par noeud bat un vector store nu sur la précision relationnelle.

🔄 Versions & écosystème 2026

Outil	Version stable mai 2026	Force	Faiblesse
mem0	0.3.x	Simple, open-source, multi-backend	Recall épisodique perfectible
Letta (ex MemGPT)	0.7.x	Mémoire OS-like, archival storage natif	Lourd, courbe d'apprentissage
Zep Cloud	v2	Facts extraction + temporal graph	SaaS, données hors UE possible
Cognee	0.2.x	Knowledge graph + vector hybride	Jeune, doc inégale
Anthropic Memory API	beta 2026	Managed, intégré tool_use Claude	Vendor lock-in, opacité
LangGraph checkpointers	Postgres / Redis	State persistant, fork de conversation	Pas une "vraie" mémoire sémantique
Pinecone Inference	v3	Vector store managé, latence < 50ms	Cher (~70$/mois minimum)
Qdrant + payload filter	1.10.x	Self-hosted, RGPD-friendly, perf top	Ops à gérer

Recommandation FR 2026 :

• Mission < 100k€ : mem0 + Postgres pgvector self-hosted (Scaleway, OVH).
• Mission > 100k€ santé/finance : Cognee ou Letta + Qdrant, audit RGPD complet.
• Banque / mutuelle : self-hosted obligatoire, jamais Zep Cloud ou Anthropic Memory API en direct (sauf via Bedrock EU region avec DPA).

⚠️ Pitfalls

1. Tout stocker → vector store pollué, recall pourri. Filtrer agressivement avant memory.add().
2. Embeddings de PII en clair → un dump base = leak RGPD. Chiffrer au repos (Postgres TDE, S3 SSE-KMS).
3. Pas de user_id partitioning → cross-contamination entre clients (un acheteur reçoit les biens d'un autre). Toujours WHERE user_id = ? côté backend, jamais via prompt.
4. Vector search sans filter metadata → résultats hors contexte temporel (mémoire de 2024 dans un contexte 2026). Toujours filtrer par ts, user_id, source.
5. Memory contradictions non gérées : "budget 500k" puis "budget 750k" → l'agent moyenne ou hésite. Implémenter un step "reconcile" qui supprime les mémoires obsolètes.
6. Pas de TTL sur mémoires sensibles → durée de conservation excède RGPD (ex: bancaire = 10 ans max, prospects = 3 ans). Job de purge automatique.
7. Mémoire procédurale figée : on stocke "envoyer email après visite" mais l'agent l'applique en boucle, même si l'acheteur a explicitement dit "pas d'email". Toujours croiser procédural avec contraintes épisodiques.
8. Coût embeddings explose sur batch (200k turns à embedder = 50€ d'un coup). Batcher les embeddings (1000 par appel) et utiliser models small en eval.
9. Confusion thread_id LangGraph vs user_id mem0 : ce sont deux IDs distincts. Un user peut avoir N threads, chaque thread a son checkpoint, mais les mémoires sont sur user_id. Documenter clairement.
10. Backup/restore non testés : un crash Postgres = mémoires perdues = client furieux. Snapshots quotidiens + test restore mensuel obligatoires.

🧠 Comment un staff engineer raisonne sur la mémoire

La mémoire n'est pas une base de données, c'est un pipeline de retrieval avec budget de tokens, latence p99 et failure modes silencieux. Le piège junior est de traiter memory.add() / memory.search() comme des appels SQL. Un senior raisonne en quatre axes.

1. Le write path est un LLM call caché — donc un coût, une latence et un point de non-déterminisme

memory.add() chez mem0 n'écrit pas le message brut : il appelle un LLM pour extraire des faits, détecter les contradictions et fusionner avec la mémoire existante. Conséquences que le junior ignore :

• Chaque add coûte un appel LLM (~500-1500 tokens in/out). À 200 turns/jour × 1000 users, ça fait 200k appels d'extraction/jour. C'est souvent le premier poste de coût, devant les embeddings et le LLM conversationnel.
• Le write est non-déterministe : le même message peut produire des faits différents selon la run. Tu ne peux pas tester par égalité exacte — tu testes par recall sur un golden set.
• Le write peut bloquer la réponse user. Si tu fais add synchrone avant de répondre, tu ajoutes 1-2s de latence p99 perçue. Pattern senior : add en fire-and-forget (background task / queue), jamais sur le chemin critique de la réponse.

# ❌ junior : l'extraction bloque la réponse
answer = generate(user_msg, ctx)
memory.add(messages=[...], user_id=uid)   # +1.5s avant de rendre la main
return answer

# ✅ senior : la réponse part, l'écriture mémoire est asynchrone
answer = generate(user_msg, ctx)
background_tasks.add_task(memory.add, messages=[...], user_id=uid)
return answer   # latence dominée par le seul call conversationnel

2. Le read path est un problème de budget, pas de "trouver l'info"

Le retrieval n'est jamais "renvoie tout ce qui matche". C'est "renvoie le sous-ensemble qui maximise l'utilité par token injecté", sous contrainte de fenêtre et de coût. Trois leviers, dans l'ordre où un senior les active :

Levier	Effet	Quand
limit=k (top-k)	Borne le nombre de mémoires	Toujours. Démarre à k=5-8, mesure recall@k.
Filtre metadata (user_id, ts, type)	Élimine le hors-scope avant le scoring sémantique	Toujours. Non négociable pour le multi-tenant (cf. pitfall 3).
Re-ranking / decay	Reordonne par pertinence × fraîcheur	Quand le top-k brut injecte des faits obsolètes.

Le piège : augmenter k pour "ne rien rater" dégrade la qualité (le LLM se noie dans le bruit) et le coût. Le bon réflexe est l'inverse : k petit + filtre agressif + decay, puis monter k seulement si recall@k le justifie sur le golden set.

3. Les failure modes sont silencieux — il faut les instrumenter

Un agent à mémoire ne plante pas, il dérive. Pas d'exception, juste des recommandations qui se dégradent. Métriques à logger sur chaque search (pas en batch après coup) :

• cache_read_input_tokens / usage du call conversationnel → coût réel par tour (cf. prompt caching ci-dessous).
• n_memories_returned, top_score, mean_score → si top_score chute, le retrieval est en train de pourrir.
• memory_hit_used → est-ce que le LLM a réellement utilisé une mémoire injectée, ou l'a-t-il ignorée ? (heuristique : citation/overlap entre réponse et mémoires).
• write_latency, extraction_tokens → surveiller la dérive du coût d'écriture.

Sans ça, tu découvres la dérive quand le client se plaint, pas avant.

4. Prompt caching : le contexte mémoire stable doit précéder la requête volatile

Les mémoires injectées dans le system changent à chaque tour → si tu les mets avant un préambule stable, tu casses le cache de tout ce qui suit. Render order = tools → system → messages : mets le prompt système figé en premier (avec cache_control), et les mémoires + la question utilisateur après le dernier breakpoint. Sinon cache_read_input_tokens reste à 0 et tu paies le plein tarif à chaque tour.

resp = client.messages.create(
    model="claude-opus-4-8",
    max_tokens=1024,
    # Opus 4.8 : thinking adaptatif + effort (PAS de budget_tokens → 400 sur 4.7/4.8).
    # Sur un tour conversationnel court, "low" suffit ; "high" pour du raisonnement multi-étapes.
    thinking={"type": "adaptive"},
    output_config={"effort": "low"},
    system=[
        # figé entre tous les tours → mis en cache (préfixe stable : persona + schéma d'outils)
        {"type": "text", "text": AGENT_PERSONA, "cache_control": {"type": "ephemeral"}},
    ],
    messages=[
        # volatile (mémoires + question) → APRÈS le breakpoint, jamais avant
        {"role": "user", "content": f"Mémoires:\n{ctx}\n\nQuestion: {user_msg}"},
    ],
)
log.info("cost", cache_read=resp.usage.cache_read_input_tokens, in_=resp.usage.input_tokens)

Vérification senior : si cache_read_input_tokens reste à 0 sur des tours successifs avec le même persona, un invalidateur silencieux est à l'œuvre — un datetime.now() ou un user_id interpolé dans system, un json.dumps non trié, ou un set d'outils qui change. Le minimum cachable sur Opus 4.8 est 4096 tokens : un persona plus court ne sera jamais mis en cache (pas d'erreur, juste cache_creation_input_tokens: 0). Coût réel d'un tour = input_tokens + cache_creation_input_tokens + cache_read_input_tokens — ne lis jamais input_tokens seul pour juger le coût.

Règle de design : tout ce qui est stable (persona, instructions, schéma d'outils) vit dans le préfixe caché ; tout ce qui varie par tour (mémoires retrouvées, message user) vit après. Un datetime.now() ou un user_id interpolé dans le system suffit à tout invalider.

5. Le write mémoire est un appel réseau — il faut le rendre résilient

L'extraction memory.add() part en fire-and-forget (axe 1), mais "asynchrone" ne veut pas dire "best-effort silencieux". Un serveur de prod attend du write path qu'il soit typé, retryé et observé, sinon la dérive (axe 3) devient une perte de données invisible.

• AsyncAnthropic côté serveur, pas le client sync. Un agent FastAPI qui bloque un worker sur chaque extraction LLM s'effondre sous la charge. L'extraction part dans une tâche async (ou une queue durable type Redis/SQS si tu veux survivre à un crash process — une BackgroundTask FastAPI meurt avec le process).
• max_retries + exceptions typées. Le SDK retry 429/5xx en exponential backoff (max_retries=2 par défaut). Attrape RateLimitError, OverloadedError, APITimeoutError explicitement — jamais de if "429" in str(e). Un OverloadedError sur le write n'est pas fatal (on re-extrait au prochain tour), mais doit être loggé, pas avalé.
• Timeout par appel. Une extraction qui pend (réseau, surcharge) ne doit pas accumuler des tâches zombies. Mets un timeout= explicite sur le client d'extraction ; au-delà, on drop le write de ce tour et on logge extraction_dropped.
• Parallélise les writes indépendants avec asyncio.gather. Quand un webhook matche un bien contre N acheteurs (cf. end-to-end), les N extractions/logs sont indépendants : gather au lieu d'une boucle séquentielle divise la latence du batch par N.

from anthropic import AsyncAnthropic, RateLimitError, OverloadedError, APITimeoutError

aclient = AsyncAnthropic(max_retries=3, timeout=20.0)

async def safe_add(messages, user_id):
    try:
        await memory.add(messages=messages, user_id=user_id)  # mem0 async backend
    except (RateLimitError, OverloadedError, APITimeoutError) as e:
        log.warning("extraction_dropped", user_id=user_id, err=type(e).__name__)
        # éventuelle cohérence : on re-extraira au prochain tour ; jamais on ne bloque la réponse

💰 Pricing / ROI client

Décomposition pricing typique mission "agent à mémoire prod" :

Phase	Jours	TJM	Total
Cadrage + DPIA RGPD	3 j	1 200€	3 600€
Stack memory (mem0 + pgvector + checkpoint)	8 j	1 300€	10 400€
Pipeline d'extraction faits + write filter	6 j	1 300€	7 800€
Reconcile + decay + TTL	4 j	1 300€	5 200€
Module RGPD (purge, export, audit)	3 j	1 300€	3 900€
Eval + dashboard recall@k	4 j	1 350€	5 400€
Total POC + V1	28 j		36 300€

Argumentaire valeur : un agent qui se souvient = +30% NPS client + 2-5× rétention. Un agent qui oublie = abandon dès la 3e session.

🧪 Testing / Eval

Métriques clés mémoire :

• Recall@k : sur 100 questions dorées, l'info pertinente est-elle dans top-k mémoires retournées ?
• Precision@k : combien des k mémoires retournées sont effectivement utiles ?
• Memory hit rate : % de turns où une mémoire long-terme a été utilisée.
• Contradiction rate : % de mémoires en contradiction avec les nouveaux faits.
• Decay coherence : un fait de 2 ans est-il pondéré moins qu'un fait d'hier ?

# Test recall@5 sur dataset doré
GOLDEN = [
    {"user_id": "test_u1", "query": "Mon budget ?", "expected_keywords": ["750", "750k"]},
    {"user_id": "test_u1", "query": "Étage ?", "expected_keywords": ["refuse RDC"]},
]

def eval_recall_at_k(k=5):
    hits = 0
    for case in GOLDEN:
        results = memory.search(case["query"], user_id=case["user_id"], limit=k)
        text = " ".join([r["memory"] for r in results["results"]])
        if any(kw in text for kw in case["expected_keywords"]):
            hits += 1
    return hits / len(GOLDEN)

print(f"Recall@5: {eval_recall_at_k(5):.2%}")

Test RGPD purge : ajouter 100 faits pour un user_id, appeler delete_all, vérifier que search retourne 0 résultat + que le row Postgres est physically deleted (pas soft delete).

🧱 Architecture mémoire stratifiée (référence)

Pour les missions sérieuses, séparer physiquement les 4 couches mémoire :

┌─────────────────────────────────────────────────────────────────┐
│ Couche 1 — Working memory (in-context)                          │
│ Stockage : RAM, scoped au turn                                  │
│ TTL : 1 turn                                                    │
│ Format : messages JSON                                          │
├─────────────────────────────────────────────────────────────────┤
│ Couche 2 — Conversation buffer + summary                        │
│ Stockage : Redis (TTL 24h) puis Postgres                        │
│ TTL : 30 jours                                                  │
│ Format : turns + résumés roulants                               │
├─────────────────────────────────────────────────────────────────┤
│ Couche 3 — Long-term semantic                                   │
│ Stockage : pgvector ou Qdrant                                   │
│ TTL : 3 ans (prospects) / 10 ans (financier)                    │
│ Format : embeddings + metadata + texte                          │
├─────────────────────────────────────────────────────────────────┤
│ Couche 4 — Knowledge graph (optionnel)                          │
│ Stockage : Neo4j, ArangoDB                                      │
│ TTL : indéfini (entités stables)                                │
│ Format : nodes + edges typés                                    │
└─────────────────────────────────────────────────────────────────┘

Routing query → couche :

def route_query(query, user_id):
    # Question explicite sur passé long → couche 3
    if any(k in query.lower() for k in ["rappel", "souviens", "dernière fois"]):
        return search_long_term(query, user_id)
    # Question relationnelle → couche 4
    if any(k in query.lower() for k in ["lien avec", "relié à", "associé"]):
        return search_graph(query, user_id)
    # Tout court terme → couche 1+2 suffit
    return get_conversation_context(user_id)

🧾 Conformité RGPD : matrice d'opérations

Opération RGPD	Implémentation technique
Droit d'accès (art.15)	API /user/{id}/memories retournant tous les faits stockés
Droit à l'oubli (art.17)	API /user/{id} DELETE + purge vector store + checkpointer LangGraph
Droit de rectification (art.16)	API /user/{id}/memories/{mid} PATCH
Portabilité (art.20)	Export JSON structuré + embeddings (optionnel) signed-URL
Minimisation (art.5)	LLM filter "ce fait est-il vraiment nécessaire ?" avant memory.add()
Limitation durée	Job nightly DELETE WHERE created_at < NOW() - INTERVAL '3 years'
Sécurité (art.32)	Chiffrement at-rest (TDE), in-transit (TLS), KMS pour clés
Logs des accès	Audit table memory_access avec qui/quand/quel user
DPIA	À produire si données sensibles (santé, opinions, syndic, judiciaire)

# Implémentation droit à l'oubli atomique
@api.delete("/user/{user_id}")
def gdpr_forget(user_id: str):
    with db.transaction():
        # 1. Vector store
        memory.delete_all(user_id=user_id)
        # 2. Checkpointer LangGraph
        for thread_id in get_user_threads(user_id):
            checkpointer.delete_thread(thread_id)
        # 3. Conversation buffer Redis
        redis.delete(f"conv:{user_id}:*")
        # 4. Audit log: enregistrer la demande
        audit_log.insert(user_id=user_id, action="gdpr_delete", ts=datetime.utcnow())
    return {"status": "purged", "user_id": user_id, "verified": verify_purged(user_id)}

🔀 Mémoire collaborative multi-agent

Quand plusieurs agents partagent un contexte (équipe vente, supervisor multi-workers), trois patterns :

1. Shared episodic store : tous les agents écrivent dans un même org_id namespace, lisent les uns les autres. Risque : pollution croisée.
2. Agent-scoped + supervisor digest : chaque agent a sa mémoire, le supervisor lit toutes les mémoires et écrit un digest commun.
3. Event sourcing : agents publient des "événements mémoire" dans Kafka, chacun reconstruit son état. Cher mais audit-friendly.

# Pattern 2 : agent-scoped + supervisor digest
memory.add("Lead Total qualifié niveau A", agent_id="agent_sdr_alice")
memory.add("RDV planifié 23/05 avec DRH", agent_id="agent_sdr_alice")
# Supervisor agrège tous les soirs
for sdr in team:
    facts = memory.get_all(agent_id=sdr)
    summary = llm.invoke(f"Résume l'activité de {sdr}: {facts}")
    memory.add(f"{sdr}: {summary}", agent_id="supervisor")

🔁 Quand utiliser / éviter

Utiliser de la mémoire long terme quand :

• Utilisateur récurrent (B2B, SaaS, agent perso) sur plusieurs sessions.
• Personnalisation = valeur business (immo, vente, RH, juridique).
• Conformité exige un audit trail (santé, finance, droit).

Éviter quand :

• One-shot chatbot (FAQ produit, helpdesk transactionnel) — surcoût inutile.
• Données ultra-sensibles sans budget RGPD/sécu (médical sans DPO) — risque > valeur.
• Volume très faible (< 50 utilisateurs uniques) — un fichier JSON suffit.

Outil par cas :

• B2B SaaS standard → mem0 + pgvector.
• Banque / santé / juridique → Letta self-hosted + Qdrant chiffré.
• Multi-modal (audio/vidéo aussi) → Anthropic Memory API.
• Graph relationnel critique (KYC, anti-fraude) → Cognee ou Neo4j custom.

❓ FAQ freelance

"Pourquoi pas juste stocker tout dans Postgres avec un LIKE ?" Pour 50 utilisateurs et un domaine étroit, ça marche. Au-delà, les requêtes en langage naturel ("rappel mon budget") ne mappent pas à du SQL trivial — il faut soit comprendre la requête (LLM + texte structuré), soit faire de la similarité sémantique (embeddings). Embeddings + filter metadata = compromis pragmatique.

"mem0 vs Letta vs Zep, lequel choisir vraiment ?"

• mem0 si tu veux du simple, fonctionnel, open-source, et que tu pilotes ton infra.
• Letta si tu veux un système "OS-like" avec mémoire archivale et tu as le temps d'investir.
• Zep si tu veux du managed et que la conformité UE n'est pas critique. Pour les missions FR 90% du temps : mem0 + pgvector self-hosted.

"Comment je gère le droit à l'oubli sur Anthropic Memory API ?" L'API expose des endpoints de suppression par user_id. Documenter dans le DPA que Anthropic agit en sous-traitant, et tester réellement l'effacement (audit avant chaque release). Pour clients très conservateurs (mutuelles, banques), souvent l'Anthropic Memory API ne passe pas et il faut self-host.

"Combien de mémoires par utilisateur stocker ?" Empiriquement : 50-300 mémoires utiles par utilisateur après filtrage. Au-delà, soit tu over-stockes (mauvais filtre), soit l'utilisateur est ultra-actif (compte VIP). 1000+ mémoires actives = revoir l'extraction.

"Faut-il un knowledge graph ou les embeddings suffisent ?" Embeddings suffisent dans 80% des cas. KG nécessaire si tes données ont une structure relationnelle riche que tu dois requêter par chemins ("qui a parlé à qui sur quel dossier ?"). Pour KYC, anti-fraude, généalogie d'entreprises : KG. Pour assistant conversationnel : embeddings.

✅ Checklist mise en prod mémoire

• [ ] Schéma vector store défini, indexes créés (pgvector HNSW, Qdrant payload)
• [ ] user_id (ou tenant_id) partitioning testé sur cross-tenant queries
• [ ] Extraction filter LLM en place (pas tout stocker)
• [ ] Reconciliation contradictions implémentée
• [ ] TTL / decay configuré selon réglementation
• [ ] Endpoint RGPD /user/{id} DELETE testé end-to-end
• [ ] Export RGPD /user/{id}/memories GET testé
• [ ] Encryption at-rest activée (TDE Postgres, KMS S3)
• [ ] TLS partout (LLM, vector store, DB)
• [ ] Backup quotidien testé (snapshot + restore réel)
• [ ] Dashboard recall@k et coût mensuel
• [ ] Documentation DPIA disponible (si données sensibles)
• [ ] Plan de purge "panique" si breach (script purge_all.py testé)

💡 Pattern avancé : "active forgetting" basé sur contradictions

Mémoires deviennent obsolètes au fil du temps. Plutôt qu'un decay aveugle, on active un job nightly qui détecte contradictions et purge :

def active_forget(user_id: str):
    all_mems = memory.get_all(user_id=user_id)["results"]
    # Cluster par sujet via embeddings
    clusters = cluster_by_similarity(all_mems, threshold=0.78)
    for cluster in clusters:
        if len(cluster) < 2:
            continue
        # LLM détecte contradictions intra-cluster
        prompt = f"Voici des faits sur le même sujet, ordonnés par date:\n{cluster}\nIdentifie quels faits sont obsolètes (contredits par un fait plus récent). JSON: {{'obsolete_ids': [...]}}"
        resp = llm.invoke(prompt)
        obsolete = json.loads(resp.content)["obsolete_ids"]
        for mid in obsolete:
            memory.delete(memory_id=mid, user_id=user_id)
            audit_log.insert(user_id=user_id, action="active_forget", memory_id=mid)

Tourne 1×/semaine. Économise tokens, améliore qualité des recall, démontre compliance RGPD ("minimisation des données").

🧭 Pattern : mémoire procédurale (skills appris)

Distinct des autres : on stocke "comment l'agent doit faire" plutôt que "ce que le user veut". Exemple agent vente :

# Skill stocké après succès
memory.add(
    "Procédure démo prospect 'startup tech' : commencer par démo API, puis pricing, jamais la roadmap (trop spéculative)",
    agent_id="agent_sales",
    metadata={"type": "procedural", "skill": "demo_startup_tech", "success_count": 14}
)
# Retrieval avant chaque démo
skills = memory.search("démo prospect startup", agent_id="agent_sales", filters={"type": "procedural"})

C'est une forme de "self-improvement" : l'agent apprend de ses succès passés. Attention : à un moment, il faut un humain pour valider quels patterns deviennent des skills permanents (sinon on accumule du bruit).

🏋️ Exercices

Progressifs : du "fais-le marcher" au "défends le chiffre / casse-le puis répare-le". Fais-les dans l'ordre.

Exercice 1 — Mémoire async sans bloquer la réponse

Objectif : transformer un agent où memory.add() est sur le chemin critique en un agent où l'écriture est fire-and-forget, et prouver le gain de latence p99.

Indice/Solution : déplace add dans une BackgroundTasks FastAPI (ou une queue). Benchmark : 50 tours en mesurant time.perf_counter() autour de la réponse, version sync vs async. Tu dois voir la p99 chuter d'environ la durée d'un appel d'extraction (~1-1.5s). Piège à gérer : si la réponse N+1 arrive avant que l'écriture N soit committée, la mémoire du tour N manque → assume cette fenêtre de cohérence éventuelle ou sérialise par user_id.

Exercice 2 — Golden set + recall@k, puis tuning de k

Objectif : construire un dataset doré de 30 questions/réponses sur 5 users fictifs, mesurer recall@k pour k ∈ {3, 5, 8, 15}, et choisir k par la donnée, pas au feeling.

Indice/Solution : reprends eval_recall_at_k. Trace recall@k ET coût (tokens injectés ≈ k × taille moyenne mémoire) sur le même graphe. Tu dois observer un genou : au-delà d'un certain k, recall plafonne mais le coût et le bruit montent. Le bon k est juste avant le plafond. Bonus : montre qu'ajouter un filtre ts (decay) permet de baisser k à recall constant.

Exercice 3 — Reconcile des contradictions (write-time)

Objectif : implémenter un step reconcile qui, à l'écriture d'un fait contredisant un fait existant ("budget 500k" → "budget 750k"), supprime l'ancien au lieu de stocker les deux.

Indice/Solution : avant add, search les mémoires du même cluster sémantique (seuil cosine > 0.8), demande au LLM {"obsolete_ids": [...]} en sortie structurée (output_config.format + schéma, pas du XML à la main), purge, puis écris. Test : injecte 3 budgets successifs, vérifie qu'il ne reste qu'une mémoire budget et que c'est la dernière. Failure mode à exhiber : deux écritures concurrentes du même user → race → décris la mitigation (lock par user_id ou idempotence sur le contenu).

Exercice 4 — RGPD : droit à l'oubli atomique et vérifié

Objectif : implémenter DELETE /user/{id} qui purge vector store + checkpoints LangGraph + buffer Redis dans une transaction, et renvoie une preuve que search retourne 0 ET que la ligne Postgres est physiquement supprimée (pas soft-delete).

Indice/Solution : enchaîne memory.delete_all, checkpointer.delete_thread pour chaque thread, redis.delete, puis audit_log.insert. La preuve : re-search après purge (doit être vide) + requête SQL SELECT count(*) WHERE user_id = ? (doit être 0). Piège qui fait échouer l'audit : un index ou un cache embeddings qui garde une copie → liste tous les endroits où la PII peut survivre (replicas, backups, logs) et documente lesquels sont dans le scope du DELETE.

Exercice 5 — Casse-le : empoisonnement cross-tenant, puis répare

Objectif : écrire un test qui prouve une fuite cross-tenant si le user_id partitioning est mal fait, puis corriger pour que la fuite soit impossible.

Indice/Solution : crée user A et user B, fais croire à l'agent (via prompt injection dans un message de A) qu'il doit "se rappeler des biens de tous les clients", puis search côté B. Si le filtre user_id est appliqué dans le prompt plutôt qu'au niveau du backend (clause WHERE/payload filter), B voit les données de A → fuite. Répare en imposant le filtre au niveau du vector store (payload filter Qdrant / WHERE user_id pgvector), jamais via instruction LLM. Le test doit rester vert même avec une injection adverse.

Exercice 6 — Défends le chiffre : coût mensuel d'un agent à 200 users

Objectif : produire et justifier une estimation de coût mensuel défendable devant un CTO, décomposée : embeddings, extraction (write LLM), retrieval+chat (read LLM), stockage. Puis montrer le levier qui la divise par 2.

Indice/Solution : compte les appels réels — N tours/user/mois × (1 embedding + 1 extraction + 1 chat). Le poste dominant est souvent l'extraction (1 LLM call par add), pas le chat. Leviers à chiffrer : (a) filtre write-on-meaningful pour ne pas écrire chaque tour (−40-70 % d'extractions), (b) prompt caching sur le persona stable (−~80 % sur l'input du chat), (c) Sonnet au lieu d'Opus pour l'extraction. Livrable : un tableau avant/après avec le usage.cache_read_input_tokens mesuré, pas estimé.

Exercice 7 — Casse-le sous charge : le write path async qui perd des mémoires

Objectif : monter un agent où l'extraction part en BackgroundTasks, le mettre sous charge (provider qui répond 529 Overloaded 1 fois sur 5), prouver que des mémoires sont silencieusement perdues, puis rendre le write path résilient sans jamais bloquer la réponse user.

Indice/Solution : simule le provider avec un mock qui lève OverloadedError/APITimeoutError de façon aléatoire. Tire 200 tours concurrents (asyncio.gather) et compte les faits réellement persistés vs attendus → tu verras un trou (les add qui ont échoué sont avalés par la BackgroundTask qui meurt sans bruit). Répare en couches : (1) AsyncAnthropic(max_retries=3, timeout=20) pour absorber les 529 transitoires, (2) attrape les exceptions typées et logge extraction_dropped au lieu d'avaler, (3) remplace la BackgroundTask (qui meurt avec le process) par une queue durable (Redis/SQS) si la perte d'un write est inacceptable. Failure mode à exhiber et défendre : même avec retry, un crash worker entre la réponse et le commit du write laisse une fenêtre de cohérence éventuelle — décris-la explicitement (au pire on re-extrait au tour suivant) plutôt que de prétendre une garantie forte que tu n'as pas.

🎤 En entretien

« Pourquoi un vector store plutôt que juste stocker l'historique de conversation dans le contexte ? » Parce que le contexte est borné et coûteux : injecter 6 mois d'historique brut sature la fenêtre et fait payer le plein tarif à chaque tour. La mémoire vectorielle ne ramène que les k faits pertinents (retrieval), et l'extraction de faits compresse le bruit conversationnel en signal réutilisable.

« mem0 appelle un LLM à l'écriture. Quelle est l'implication architecturale ? » Le write path est coûteux, non-déterministe et lent — donc (1) il ne doit jamais bloquer la réponse user (fire-and-forget), (2) c'est souvent le premier poste de coût (un LLM call par add), et (3) on ne le teste pas par égalité exacte mais par recall sur un golden set.

« Comment garantis-tu l'isolation entre deux clients dans un agent multi-tenant à mémoire ? » Le filtre user_id/tenant_id s'applique au niveau du backend (payload filter Qdrant, clause WHERE pgvector), jamais via une instruction dans le prompt — un prompt est contournable par injection, un filtre backend ne l'est pas. Toute requête sans ce filtre est un bug de sécurité, pas de pertinence.

« Le droit à l'oubli RGPD sur un système à embeddings : quels endroits oublie-t-on de purger ? » Au-delà du vector store : les checkpoints d'état (LangGraph), le buffer conversationnel (Redis), les replicas de lecture, les backups/snapshots, et les logs contenant de la PII. Un delete_all qui ne couvre que le vector store laisse fuir — il faut une purge atomique multi-store et une vérification post-purge (re-search vide + count SQL à 0), plus une politique de TTL sur les backups.

« Ton agent à mémoire injecte du contexte qui change à chaque tour. Comment tu protèges le prompt cache ? » Render order = tools → system → messages. Le persona/schéma d'outils (stable) va dans system avec un cache_control, les mémoires retrouvées + la question (volatiles) vont après le dernier breakpoint, dans messages. Mettre les mémoires dans system casse le cache de tout ce qui suit → cache_read_input_tokens reste à 0 et tu paies le plein tarif à chaque tour. Sur Opus 4.8 le préfixe doit faire ≥ 4096 tokens pour être cachable, sinon ça ne cache pas silencieusement.

« Pourquoi tester un système mémoire par recall sur golden set plutôt que par assertion exacte ? » Parce que le write path (memory.add) est un appel LLM non-déterministe : le même message peut produire des faits formulés différemment d'une run à l'autre. Une assertion assert facts == [...] est fragile par construction. On teste le comportement observable — recall@k et precision@k sur un dataset doré — et on traite la sortie structurée du reconcile via output_config.format + schéma plutôt qu'un parsing XML fait main, qui dérive.

🔗 Liens

• mem0 — docs.mem0.ai
• Letta (ex MemGPT) — letta.com — papier arxiv.org/abs/2310.08560
• Zep — getzep.com
• Cognee — github.com/topoteretes/cognee
• LangGraph persistence — langchain-ai.github.io/langgraph/concepts/persistence
• Anthropic Memory API — docs.anthropic.com
• CNIL guide IA générative 2025 — cnil.fr/fr/intelligence-artificielle
• Paper "Memory in LLM Agents survey" (2024) — arxiv.org/abs/2404.13501