Conception de la mémoire d’agent à quatre niveaux avec LangGraph et Amazon Bedrock
Apprenez à doter les agents LLM de mémoires épisodique, sémantique et procédurale fonctionnelles sur Bedrock et LangGraph, ainsi qu’à les protéger contre le poisonage, les fuites de PII et les fuites d’informations entre locataires.
Sources: Agentic Architectures — Article 7: Agent Memory Architectures
Tôt ou tard, un agent de production commence à effectuer deux fois le même travail. Rien n’est cassé et la demande est correcte ; l’agent ne se souvient tout simplement pas avoir résolu un problème presque identique la semaine précédente, il suit donc les mêmes étapes, répète les mêmes erreurs intermédiaires, arrive à la même réponse, et vous devez payer à nouveau pour chaque token. Ce guide montre comment y remédier en utilisant une véritable infrastructure de mémoire plutôt qu’une fenêtre de contexte plus grande : quatre niveaux de mémoire distincts construits avec LangGraph sur Amazon Bedrock, ainsi que des mesures de sécurité en production qui empêchent la mémoire de devenir un fardeau.
Imaginez un agent d’analyse de contrats doté d’un système solide et d’une configuration multi-agents fonctionnelle, qui consacre néanmoins ses premières appels d’outils pour chaque contrat similaire à la redécouverte de schémas structurels qu’il avait déjà identifiés auparavant. Ce dont il manque, c’est de la mémoire.
Une fenêtre de contexte plus grande n’est pas de la mémoire
Lorsque la fenêtre de contexte est pleine, la réaction habituelle consiste à augmenter sa taille ou à recourir à une synthèse plus agressive. Ces deux approches sont utiles au cours d’une même séquence d’actions ; aucune ne remplace la mémoire. Une analogie utile : la fenêtre de contexte est un tableau blanc. Elle est pratique pendant une réunion et est effacée à sa fin. La mémoire, quant à elle, correspond au carnet que l’on emporte d’une réunion à l’autre.
La gestion du contexte présente trois limites spécifiques :
- Elle ne permet pas de relier les différentes séquences d’actions. Lorsque l’agent termine son travail et que son processus s’arrête, le contexte disparaît. Tout ce qui a été appris est perdu à moins de le conserver explicitement.
- Elle ne franchit pas les frontières entre agents. Dans un système multi-agents, chaque agent dispose de son propre contexte. Ce que découvre un agent chargé de l’analyse de sécurité reste invisible pour un agent chargé de l’analyse du code, à moins de le transmettre délibérément vers lui.
La mémoire répond à un besoin différent : un stock de connaissances persistant et consultable, qui survit aux exécutions individuelles et peut être partagé quel que soit le type de topologie d’agent utilisée.
Les quatre niveaux de mémoire des agents
Chaque niveau implique une infrastructure, des modes de récupération et des modes de défaillance différents :
+---------------------+--------------------------------------------------+----------------------------+
| Memory Tier | What It Stores | Survives Run End? |
+---------------------+--------------------------------------------------+----------------------------+
| Working Memory | Active context during current run | No |
| | Tool outputs, reasoning steps, partial results | |
+---------------------+--------------------------------------------------+----------------------------+
| Episodic Memory | Summaries of past runs | Yes |
| | What happened, what worked, what failed | |
+---------------------+--------------------------------------------------+----------------------------+
| Semantic Memory | Structured knowledge about the domain | Yes |
| | Facts, relationships, learned patterns | |
+---------------------+--------------------------------------------------+----------------------------+
| Procedural Memory | Successful approaches to recurring task types | Yes |
| | How to do things, not just what was done | |
+---------------------+--------------------------------------------------+----------------------------+
Ces niveaux interagissent en cycle. La mémoire de travail conserve l’exécution en cours. Lorsque l’exécution se termine, une étape de distillation écrit les données dans les trois niveaux persistants. Lorsque la prochaine exécution commence, une étape de récupération extrait à nouveau les éléments pertinents :
Run starts
|
v
+------------------+
| Working Memory | <-- active context, tool outputs, LangGraph state
| (current run) |
+------------------+
|
| Run ends: distillation step
v
+------------------+ +------------------+ +------------------+
| Episodic Memory | | Semantic Memory | | Procedural Memory|
| "what happened" | | "what I know" | | "how I do this" |
| DynamoDB | | Bedrock KB | | DynamoDB + KB |
+------------------+ +------------------+ +------------------+
| | |
+------------------------+------------------------+
|
Next run starts: retrieval step
|
v
Agent bootstrapped with
relevant prior knowledge
La mémoire épisodique indique à la prochaine exécution ce qui s’est passé, la mémoire sémantique ce qui est connu sur le domaine, et la mémoire procédurale comment aborder la tâche. Ces trois types de mémoire sont récupérés au démarrage et placés dans la mémoire de travail avant que l’agent ne fasse son premier pas de raisonnement.
Niveau 1 : gérer la mémoire de travail au cours d’une exécution
La mémoire de travail comprend tout ce qui existe pendant une seule exécution, et vous la gérez déjà que vous l’appeliez ainsi ou non. Dans LangGraph, l’état du graphe constitue la mémoire de travail : chaque message, chaque résultat d’outil et chaque résultat intermédiaire y sont stockés.
Le problème, c’est que cela continue de croître. La compression du contexte entre les étapes d’un pipeline, une technique courante utilisée dans les systèmes multi-agents, n’est en réalité qu’une gestion de la mémoire de travail sous un autre nom. La classe ci-dessous transforme cette idée en un composant réutilisable. maybe_compress s’exécute avant chaque étape de l’agent ; une fois que la liste des messages dépasse un seuil (30 par défaut), il conserve tels quels les messages système ainsi que les 15 messages non systèmes les plus récents, et remplace tout le reste par un seul résumé généré par un modèle petit et peu coûteux. Une deuxième méthode, extract_run_artifacts, s’exécute à la fin de l’exécution et demande au modèle de produire un enregistrement JSON structuré décrivant la tâche, le résultat, les constats, l’approche, les outils et les modèles réutilisables, ce qui devient ensuite l’entrée pour la mémoire épisodique. Ce fragment est écrit en Python, et non en JavaScript :
# harness/memory/working_memory.py
from typing import TypedDict, Annotated, List, Optional
from langgraph.graph.message import add_messages
from langchain_core.messages import BaseMessage, SystemMessage, HumanMessage, AIMessage
from langchain_aws import ChatBedrock
import boto3
class WorkingMemoryManager:
"""
Manages the active context within a single agent run.
Applies a sliding window with summarization to prevent context overflow.
"""
def __init__(
self,
max_messages: int = 40,
summarize_after: int = 30,
region: str = "us-east-1",
):
self.max_messages = max_messages
self.summarize_after = summarize_after
bedrock = boto3.client("bedrock-runtime", region_name=region)
# Haiku for summarization: cheap and fast for mechanical compression
self.summarizer = ChatBedrock(
client=bedrock,
model_id="anthropic.claude-haiku-4-5",
model_kwargs={"temperature": 0, "max_tokens": 2000},
)
def maybe_compress(self, messages: List[BaseMessage]) -> List[BaseMessage]:
"""
Called before each agent step. If the message list is getting long,
compress the oldest messages into a summary and keep the recent ones.
"""
if len(messages) < self.summarize_after:
return messages
# Keep the system message and the most recent messages
system_messages = [m for m in messages if isinstance(m, SystemMessage)]
non_system = [m for m in messages if not isinstance(m, SystemMessage)]
# Split: old messages get summarized, recent ones stay verbatim
cutoff = len(non_system) - 15 # always keep last 15 exchanges verbatim
to_summarize = non_system[:cutoff]
to_keep = non_system[cutoff:]
if not to_summarize:
return messages
summary_text = self._summarize(to_summarize)
summary_message = SystemMessage(
content=f"[WORKING MEMORY SUMMARY - earlier in this run]\n{summary_text}"
)
return system_messages + [summary_message] + to_keep
def _summarize(self, messages: List[BaseMessage]) -> str:
messages_text = "\n".join([
f"{type(m).__name__}: {m.content[:500]}"
for m in messages
])
response = self.summarizer.invoke([
SystemMessage(content="""
Summarize these agent messages into a compact working memory record.
Include: decisions made, tools called and their results, key findings,
and anything the agent will need to continue its work.
Discard: failed attempts, repetitive reasoning, tool errors that were resolved.
"""),
HumanMessage(content=messages_text)
])
return response.content
def extract_run_artifacts(self, messages: List[BaseMessage]) -> dict:
"""
At end of run, extracts key artifacts for episodic memory storage.
Separates what happened from how it happened.
"""
response = self.summarizer.invoke([
SystemMessage(content="""
Extract the following from these agent messages and return as JSON:
{
"task_summary": "one sentence description of what was asked",
"outcome": "COMPLETED | PARTIAL | FAILED",
"key_findings": ["list of important discoveries"],
"approach_used": "brief description of the approach that worked",
"tools_used": ["list of tool names"],
"time_taken_steps": 0,
"reusable_patterns": ["any patterns that could help on similar future tasks"]
}
Return only valid JSON.
"""),
HumanMessage(content="\n".join([
f"{type(m).__name__}: {m.content[:300]}"
for m in messages[-30:] # last 30 messages are enough
]))
])
import json
try:
return json.loads(response.content)
except json.JSONDecodeError:
return {"outcome": "UNKNOWN", "key_findings": [], "reusable_patterns": []}
L’instruction de résumé élimine les tentatives infructueuses ainsi que les erreurs des outils résolues, et l’analyse JSON recourt à un résultat UNKNOWN avec des listes vides, de sorte qu’une réponse mal formatée ne provoque jamais l’arrêt du processus. Les identifiants des modèles, tels que l’ID Haiku présent ici, changent avec le temps et peuvent nécessiter un profil d’inférence spécifique à une région ; vérifiez donc la liste actuelle des modèles Bedrock avant de les copier.
Pour le développement local, vous pouvez remplacer le résumateur Bedrock par un modèle fourni par Ollama :
# Local dev alternative
# ollama pull llama3.1:8b
from langchain_ollama import ChatOllama
summarizer = ChatOllama(model="llama3.1:8b", temperature=0, format="json")
Niveau 2 : mémoire épisodique des exécutions passées
La mémoire épisodique répond à la question « Que s’est-il passé avant ? ». Au début d’une exécution, l’agent récupère des résumés d’exécutions antérieures similaires à la tâche en cours et les utilise pour éviter le travail redondant ainsi que les erreurs connues.
La conception stocke des résumés condensés plutôt que des journaux d’exécution bruts. Les journaux complets sont coûteux à gérer et lents à rechercher ; les résumés compacts, quant à eux, sont peu coûteux et rapides. Le stockage est assuré par DynamoDB, et la récupération utilise des embeddings Bedrock pour évaluer la similarité sémantique.
Le tableau, défini dans Terraform, utilise agent_id comme clé de partition et run_id comme clé de tri. Un index secondaire global sur agent_id et created_at permet de lancer des requêtes telles que « les épisodes les plus récents pour cet agent », tandis que le TTL de DynamoDB sur expires_at supprime automatiquement les anciens enregistrements :
# infrastructure/memory_tables.tf
resource "aws_dynamodb_table" "episodic_memory" {
name = "agent-episodic-memory"
billing_mode = "PAY_PER_REQUEST"
hash_key = "agent_id"
range_key = "run_id"
attribute {
name = "agent_id"
type = "S"
}
attribute {
name = "run_id"
type = "S"
}
attribute {
name = "created_at"
type = "N"
}
global_secondary_index {
name = "agent-time-index"
hash_key = "agent_id"
range_key = "created_at"
projection_type = "ALL"
}
ttl {
attribute_name = "expires_at"
enabled = true
}
tags = {
Component = "agent-memory"
}
}
Le magasin intègre lui-même le résumé de la tâche ainsi que les principales conclusions, enregistre l’information avec une date d’expiration de 90 jours, et lors de la récupération compare l’intégration de la tâche actuelle avec les épisodes récents à l’aide de la similarité cosinus. Seuls les épisodes présentant une similarité supérieure à 0,75 sont conservés ; en cas d’égalité, la priorité est donnée aux résultats COMPLETED ; et format_for_injection affiche les gagnants sous forme de bloc étiqueté pour le prompt du système :
# harness/memory/episodic.py
import boto3
import time
import json
import uuid
from typing import List, Optional
from dataclasses import dataclass
from langchain_aws import BedrockEmbeddings
import numpy as np
@dataclass
class EpisodeRecord:
run_id: str
agent_id: str
task_summary: str
outcome: str
key_findings: List[str]
approach_used: str
tools_used: List[str]
reusable_patterns: List[str]
embedding: Optional[List[float]]
created_at: float
expires_at: float
class EpisodicMemoryStore:
"""
Stores and retrieves summaries of past agent runs.
Uses embeddings for semantic retrieval so the agent can find
relevant episodes even when the wording is different.
"""
RETENTION_DAYS = 90
def __init__(
self,
agent_id: str,
table_name: str = "agent-episodic-memory",
region: str = "us-east-1",
):
self.agent_id = agent_id
self.region = region
dynamodb = boto3.resource("dynamodb", region_name=region)
self.table = dynamodb.Table(table_name)
bedrock = boto3.client("bedrock-runtime", region_name=region)
self.embeddings = BedrockEmbeddings(
client=bedrock,
model_id="amazon.titan-embed-text-v2:0",
)
def store_episode(self, artifacts: dict) -> str:
"""
Called at the end of each run to store what happened.
Returns the run_id of the stored episode.
"""
run_id = str(uuid.uuid4())
now = time.time()
# Embed the task summary for semantic retrieval later
text_to_embed = artifacts.get("task_summary", "") + " " + \
" ".join(artifacts.get("key_findings", []))
embedding = self.embeddings.embed_query(text_to_embed)
record = {
"run_id": run_id,
"agent_id": self.agent_id,
"task_summary": artifacts.get("task_summary", ""),
"outcome": artifacts.get("outcome", "UNKNOWN"),
"key_findings": artifacts.get("key_findings", []),
"approach_used": artifacts.get("approach_used", ""),
"tools_used": artifacts.get("tools_used", []),
"reusable_patterns": artifacts.get("reusable_patterns", []),
"embedding": embedding,
"created_at": int(now),
"expires_at": int(now + self.RETENTION_DAYS * 86400),
}
self.table.put_item(Item=record)
return run_id
def retrieve_relevant_episodes(
self,
current_task: str,
max_results: int = 5,
min_similarity: float = 0.75,
) -> List[EpisodeRecord]:
"""
Finds past episodes semantically similar to the current task.
Returns them ranked by relevance, filtered by outcome quality.
"""
query_embedding = self.embeddings.embed_query(current_task)
# Fetch recent episodes for this agent
# In production, you'd paginate this and use a vector DB for scale
response = self.table.query(
IndexName="agent-time-index",
KeyConditionExpression="agent_id = :aid",
ExpressionAttributeValues={":aid": self.agent_id},
ScanIndexForward=False, # newest first
Limit=100,
)
episodes = []
for item in response.get("Items", []):
stored_embedding = item.get("embedding")
if not stored_embedding:
continue
similarity = self._cosine_similarity(query_embedding, stored_embedding)
if similarity >= min_similarity:
episodes.append((similarity, item))
# Sort by similarity, take top results, prefer COMPLETED outcomes
episodes.sort(key=lambda x: (x[0], x[1].get("outcome") == "COMPLETED"), reverse=True)
return [
EpisodeRecord(**{k: v for k, v in item.items() if k != "embedding"}, embedding=None)
for _, item in episodes[:max_results]
]
def format_for_injection(self, episodes: List[EpisodeRecord]) -> str:
"""
Formats retrieved episodes into a system prompt block
for injection at the start of a new run.
"""
if not episodes:
return ""
lines = ["[EPISODIC MEMORY: relevant past runs]\n"]
for ep in episodes:
lines.append(f"Task: {ep.task_summary}")
lines.append(f"Outcome: {ep.outcome}")
if ep.key_findings:
lines.append("Key findings:")
for f in ep.key_findings:
lines.append(f" - {f}")
if ep.reusable_patterns:
lines.append("Patterns that worked:")
for p in ep.reusable_patterns:
lines.append(f" - {p}")
lines.append("")
return "\n".join(lines)
def _cosine_similarity(self, a: List[float], b: List[float]) -> float:
a, b = np.array(a), np.array(b)
return float(np.dot(a, b) / (np.linalg.norm(a) * np.linalg.norm(b) + 1e-9))
La requête récupère au maximum 100 épisodes récents et calcule la similarité dans le code de l’application ; comme le note le commentaire, les déploiements plus importants devraient utiliser la pagination et une base de données vectorielle. Notez également que le resource boto3 DynamoDB rejette les valeurs float en Python, il faut donc convertir la liste d’embeddings en Decimal (ou la serialiser, par exemple sous forme JSON ou binaire) avant que put_item ne puisse la prendre en charge. Le modèle d’embeddings utilisé ici est décrit dans la documentation Amazon Titan Embeddings.
Niveau 3 : mémoire sémantique avec les bases de connaissances Bedrock
La mémoire épisodique enregistre les événements ; la mémoire sémantique enregistre les connaissances. La différence est subtile mais importante. Un épisode pourrait indiquer : « un contrat avec un fournisseur analysé mardi dernier contenait une clause 7.3 non standard ». Le fait sémantique correspondant stipule : « la clause 7.3 dans les contrats avec des fournisseurs couvre généralement les plafonds de responsabilité ; la fourchette normale est égale à deux fois la valeur du contrat, et tout montant inférieur à cette valeur constitue un signal d’alerte ».
Les faits sémantiques ont une durée de vie plus longue que les épisodes. Ils ne expirent pas après 90 jours ; ils sont affinés au fil du temps et servent pour de nombreux types de tâches, et non seulement pour celles qui ressemblent à des exécutions antérieures.
Bedrock Knowledge Bases constitue un cadre idéal pour ce niveau, car il gère le chunking, l’embedding, le stockage et la récupération. Ce que la plupart des équipes n’ont pas encore mis en place, c’est le mécanisme d’écriture, puisque les bases de connaissances sont généralement considérées comme uniquement lues. Le système présenté ci-dessous prend en charge les deux directions :
retrieveeffectue une recherche vectorielle sémantique, éventuellement filtrée par une clé de métadonnéesdomain, et élimine les résultats ayant un score inférieur à 0,6.store_factenregistre le fait sous forme de document JSON dans S3 et démarre une tâche d’ingestion afin que la Base de connaissances le prenne en compte.extract_facts_from_episodedemande à un modèle d’extraire des faits généralisables à partir d’une exécution terminée, en omettant les détails spécifiques à une tâche qui relèvent de la mémoire épisodique, et en ne conservant que les faits présentant une confiance d’au moins 0,7.
# harness/memory/semantic.py
import boto3
import json
import uuid
import time
from typing import List, Optional
from dataclasses import dataclass
@dataclass
class SemanticFact:
fact_id: str
content: str
domain: str # "contract_law", "security_patterns", etc.
confidence: float # 0.0 to 1.0
source_run_id: str
created_at: float
last_confirmed_at: float
class SemanticMemoryStore:
"""
Manages domain knowledge that persists across runs.
Reads from Bedrock Knowledge Base.
Writes new facts to S3 and triggers KB sync.
"""
def __init__(
self,
knowledge_base_id: str,
s3_bucket: str,
s3_prefix: str = "semantic-memory/",
region: str = "us-east-1",
):
self.kb_id = knowledge_base_id
self.s3_bucket = s3_bucket
self.s3_prefix = s3_prefix
self.region = region
self.bedrock_agent = boto3.client("bedrock-agent-runtime", region_name=region)
self.bedrock_agent_mgmt = boto3.client("bedrock-agent", region_name=region)
self.s3 = boto3.client("s3", region_name=region)
def retrieve(
self,
query: str,
domain: Optional[str] = None,
max_results: int = 5,
min_score: float = 0.6,
) -> List[str]:
"""
Retrieves relevant semantic facts for the current task.
Called at agent startup to bootstrap domain knowledge.
"""
retrieval_config = {
"vectorSearchConfiguration": {
"numberOfResults": max_results,
"overrideSearchType": "SEMANTIC",
}
}
if domain:
retrieval_config["vectorSearchConfiguration"]["filter"] = {
"equals": {"key": "domain", "value": domain}
}
response = self.bedrock_agent.retrieve(
knowledgeBaseId=self.kb_id,
retrievalQuery={"text": query},
retrievalConfiguration=retrieval_config,
)
return [
r["content"]["text"]
for r in response.get("retrievalResults", [])
if r.get("score", 0) >= min_score
]
def store_fact(self, fact: SemanticFact) -> str:
"""
Writes a new fact to S3 and triggers a KB ingestion job.
The fact will be available for retrieval after the sync completes
(typically 1-5 minutes).
For time-sensitive use cases, consider a separate fast-path
retrieval from DynamoDB alongside the KB retrieval.
"""
document = {
"fact_id": fact.fact_id,
"content": fact.content,
"domain": fact.domain,
"confidence": fact.confidence,
"source_run_id": fact.source_run_id,
"created_at": fact.created_at,
"metadata": {
"domain": fact.domain,
"confidence": str(fact.confidence),
}
}
key = f"{self.s3_prefix}{fact.domain}/{fact.fact_id}.json"
self.s3.put_object(
Bucket=self.s3_bucket,
Key=key,
Body=json.dumps(document),
ContentType="application/json",
)
# Trigger sync so new fact becomes retrievable
self.bedrock_agent_mgmt.start_ingestion_job(
knowledgeBaseId=self.kb_id,
dataSourceId=self._get_data_source_id(),
)
return fact.fact_id
def extract_facts_from_episode(
self,
episode_artifacts: dict,
extractor_model,
domain: str,
) -> List[SemanticFact]:
"""
At end of a successful run, extracts generalizable facts
from the episode and prepares them for semantic memory storage.
Only extracts from COMPLETED runs with confidence.
"""
if episode_artifacts.get("outcome") != "COMPLETED":
return []
from langchain_core.messages import SystemMessage, HumanMessage
response = extractor_model.invoke([
SystemMessage(content=f"""
Extract generalizable domain facts from this completed agent run.
Domain: {domain}
Rules:
- Only extract facts that would be useful across many different tasks
- Do not extract task-specific details (those belong in episodic memory)
- Rate confidence 0.0 to 1.0 based on how certain the agent was
- Skip anything that might be specific to this one instance
Return JSON array:
[
{{"content": "the fact", "confidence": 0.9}},
...
]
Return empty array if no generalizable facts found.
"""),
HumanMessage(content=json.dumps(episode_artifacts))
])
try:
facts_data = json.loads(response.content)
except json.JSONDecodeError:
return []
now = time.time()
return [
SemanticFact(
fact_id=str(uuid.uuid4()),
content=f["content"],
domain=domain,
confidence=f.get("confidence", 0.7),
source_run_id=episode_artifacts.get("run_id", "unknown"),
created_at=now,
last_confirmed_at=now,
)
for f in facts_data
if f.get("confidence", 0) >= 0.7 # only store high-confidence facts
]
def _get_data_source_id(self) -> str:
response = self.bedrock_agent_mgmt.list_data_sources(
knowledgeBaseId=self.kb_id,
maxResults=1,
)
return response["dataSourceSummaries"][0]["dataSourceId"]
Les nouveaux faits ne peuvent pas être récupérés immédiatement. Le processus d’ingestion prend du temps, généralement quelques minutes selon le commentaire du code ; par conséquent, les charges de travail sensibles au temps nécessitent une recherche rapide (par exemple via DynamoDB) en complément de la Base de connaissances. Deux autres points méritent d’être vérifiés contre la documentation sur l’ingestion dans la Base de connaissances avant d’utiliser ce code. Premièrement, le filtrage des métadonnées dans les Bases de connaissances s’appuie sur des attributs de métadonnées fournis dans un fichier distinct à côté de chaque document S3, et non sur des champs intégrés dans le corps du document ; ainsi, le filtre domain pourrait ne rien trouver tel qu’écrit. Deuxièmement, lancer une tâche d’ingestion pour chaque fait individuel peut entrer en conflit avec une tâche déjà en cours sur la même source de données ; regrouper les écritures et effectuer la synchronisation selon un calendrier est généralement plus adapté.
Niveau 4 : mémoire procédurale, avec une étape de validation
La mémoire procédurale est à la fois le niveau le plus puissant et le plus dangereux. Elle stocke comment faire quelque chose, et non seulement ce qui a été fait : une approche efficace pour un type de tâche récurrente est transformée en une liste de étapes reutilisables et intégrée dans les futures demandes du système.
Le danger réside dans le fait qu’une procédure défectueuse, une fois enregistrée, soit suivie sans question pour chaque tâche ultérieure. C’est pourquoi ce niveau nécessite un mécanisme de validation sur le chemin d’écriture. Dans l’entrepôt ci-dessous, propose_procedure ne s’exécute que pour les exécutions terminées et enregistre de nouvelles procédures avec validated réglé sur false, ce qui signifie qu’elles sont en attente. Un autre agent ou un humain doit approuver une procédure avant que get_procedure ne la restitue, et ce dernier exige en outre au moins trois applications antérieures. record_application_result suit la fréquence à laquelle une procédure a été appliquée ainsi que le taux de réussite, tandis que format_for_injection présente les étapes accompagnées du taux de réussite et d’un rappel indiquant qu’il s’agit d’un point de départ et non d’un script :
# harness/memory/procedural.py
import boto3
import json
import time
from typing import List, Optional
from dataclasses import dataclass
@dataclass
class Procedure:
procedure_id: str
task_type: str # "contract_review", "security_audit", etc.
title: str
steps: List[str]
success_rate: float # updated as procedure is applied to new runs
times_applied: int
times_succeeded: int
created_at: float
last_updated_at: float
validated: bool # human-validated procedures get priority
class ProceduralMemoryStore:
"""
Stores and retrieves learned procedures for recurring task types.
Write path has a validation gate: procedures are stored as PENDING
and must be validated (manually or by a second agent) before use.
This prevents a single bad run from poisoning future behavior.
"""
def __init__(
self,
table_name: str = "agent-procedural-memory",
region: str = "us-east-1",
):
dynamodb = boto3.resource("dynamodb", region_name=region)
self.table = dynamodb.Table(table_name)
def get_procedure(self, task_type: str) -> Optional[Procedure]:
"""
Returns the best validated procedure for a task type, if one exists.
Prefers human-validated procedures, then sorts by success rate.
"""
response = self.table.query(
KeyConditionExpression="task_type = :tt",
FilterExpression="validated = :v AND times_applied >= :min",
ExpressionAttributeValues={
":tt": task_type,
":v": True,
":min": 3, # must have been used at least 3 times
}
)
items = response.get("Items", [])
if not items:
return None
# Sort by success rate, prefer human-validated
best = sorted(
items,
key=lambda x: (x.get("validated", False), x.get("success_rate", 0)),
reverse=True
)[0]
return Procedure(**best)
def propose_procedure(
self,
task_type: str,
episode_artifacts: dict,
extractor_model,
) -> Optional[str]:
"""
After a successful run, proposes a new procedure based on
what worked. Stored as PENDING until validated.
Returns procedure_id if a procedure was proposed, None otherwise.
"""
if episode_artifacts.get("outcome") != "COMPLETED":
return None
from langchain_core.messages import SystemMessage, HumanMessage
response = extractor_model.invoke([
SystemMessage(content=f"""
Based on this successful agent run, extract a reusable procedure for
task type: {task_type}
The procedure should be:
- A numbered list of clear, actionable steps
- Generalizable to similar future tasks, not specific to this one
- Based only on what actually worked, not what was attempted and failed
Return JSON:
{{
"title": "short title for this procedure",
"steps": ["step 1", "step 2", ...]
}}
If the run does not contain a clearly reusable procedure, return null.
"""),
HumanMessage(content=json.dumps(episode_artifacts))
])
try:
procedure_data = json.loads(response.content)
except (json.JSONDecodeError, TypeError):
return None
if not procedure_data or not procedure_data.get("steps"):
return None
import uuid
now = time.time()
procedure_id = str(uuid.uuid4())
self.table.put_item(Item={
"procedure_id": procedure_id,
"task_type": task_type,
"title": procedure_data["title"],
"steps": procedure_data["steps"],
"success_rate": 1.0,
"times_applied": 0,
"times_succeeded": 0,
"created_at": now,
"last_updated_at": now,
"validated": False, # starts as PENDING
"source_run_id": episode_artifacts.get("run_id", "unknown"),
})
return procedure_id
def record_application_result(
self,
procedure_id: str,
succeeded: bool,
):
"""
Updates the success rate after a procedure is applied.
If success rate drops below threshold, flag for review.
"""
self.table.update_item(
Key={"procedure_id": procedure_id},
UpdateExpression="""
SET times_applied = times_applied + :one,
times_succeeded = times_succeeded + :inc,
last_updated_at = :now,
success_rate = (times_succeeded + :inc) / (times_applied + :one)
""",
ExpressionAttributeValues={
":one": 1,
":inc": 1 if succeeded else 0,
":now": time.time(),
}
)
def format_for_injection(self, procedure: Procedure) -> str:
"""Formats a procedure for injection into agent system prompt."""
lines = [
f"[PROCEDURAL MEMORY: recommended approach for {procedure.task_type}]",
f"{procedure.title}",
f"(Success rate: {procedure.success_rate:.0%} over {procedure.times_applied} uses)",
"",
]
for i, step in enumerate(procedure.steps, 1):
lines.append(f"{i}. {step}")
lines.append(
"\nThis procedure is a starting point. Adapt it if the current task requires it."
)
return "\n".join(lines)
Considérez ce fragment comme un esquisse de conception et non comme du code prêt à l’emploi, car quelques détails nécessitent des ajustements. Les expressions de mise à jour de DynamoDB ne prennent en charge que l’addition et la soustraction, ce qui oblige le calcul de success_rate à être effectué dans le code de l’application (ou au moment de la lecture) plutôt que directement à l’intérieur de UpdateExpression. La requête utilise task_type comme clé, tandis que les mises à jour utilisent procedure_id ; par conséquent, la table doit disposer d’un schéma de clé ou d’un index secondaire capable de gérer ces deux modes d’accès. Les timestamps en format flottant ainsi que success_rate nécessitent la même conversion en Decimal mentionnée précédemment. Enfin, combiner le critère de « validation » avec celui de « application au moins trois fois » peut créer une situation où l’on ne sait pas comment une procédure fraîchement validée peut obtenir ses premières applications, par exemple via un mode d’essai ou une exception approuvée manuellement.
Conserver chaque souvenir à jamais entraîne deux problèmes. Le stockage de centaines de milliers de résumés d’exécutions dans DynamoDB et S3 a un coût réel. De manière plus subtile, les souvenirs obsolètes induisent des erreurs : un schéma qui fonctionnait il y a six mois peut être erroné aujourd’hui car le domaine a évolué. Un mécanisme de oubli bien conçu est une fonctionnalité, pas un défaut.
Le gestionnaire de dégradation applique une politique différente selon le niveau. Pour les épisodes plus anciens que la période de conservation minimale (sept jours par défaut), il supprime les enregistrements dont le résultat était FAILED. Pour les procédures, il met en surbrillance plutôt que de supprimer : tout ce qui n’a pas été mis à jour au cours des 60 derniers jours, ou tout ce qui a été utilisé au moins trois fois avec un taux de succès inférieur à 0,6, reçoit une étiquette needs_review afin d’être examiné par un humain. Ce code est en Python, bien qu’il soit étiqueté comme SQL :
# harness/memory/decay.py
import boto3
import time
from decimal import Decimal
from typing import List
class MemoryDecayManager:
"""
Manages the lifecycle of stored memories.
Applies different decay strategies per memory tier.
"""
def __init__(self, region: str = "us-east-1"):
self.dynamodb = boto3.resource("dynamodb", region_name=region)
self.region = region
def apply_episodic_decay(
self,
agent_id: str,
table_name: str = "agent-episodic-memory",
max_episodes: int = 500,
min_retention_days: int = 7,
):
"""
Prunes old episodic records when the store gets large.
Keeps: recent episodes, COMPLETED outcomes, high-similarity clusters.
Removes: old FAILED runs, duplicates of similar episodes.
"""
table = self.dynamodb.Table(table_name)
now = time.time()
min_age = now - (min_retention_days * 86400)
response = table.query(
IndexName="agent-time-index",
KeyConditionExpression="agent_id = :aid AND created_at < :cutoff",
ExpressionAttributeValues={
":aid": agent_id,
":cutoff": Decimal(str(min_age)),
},
ScanIndexForward=True, # oldest first
Limit=max_episodes,
)
candidates_for_removal = [
item for item in response.get("Items", [])
if item.get("outcome") == "FAILED"
]
with table.batch_writer() as batch:
for item in candidates_for_removal:
batch.delete_item(Key={
"agent_id": item["agent_id"],
"run_id": item["run_id"],
})
def flag_stale_procedures(
self,
table_name: str = "agent-procedural-memory",
stale_after_days: int = 60,
min_success_rate: float = 0.6,
) -> List[str]:
"""
Identifies procedures that should be reviewed:
- Not used in a long time (might be outdated)
- Success rate has dropped below acceptable threshold
Does NOT delete automatically. Flags for human review.
Returns list of procedure_ids that need attention.
"""
table = self.dynamodb.Table(table_name)
stale_cutoff = time.time() - (stale_after_days * 86400)
response = table.scan(
FilterExpression=(
"last_updated_at < :cutoff OR "
"(times_applied >= :min AND success_rate < :threshold)"
),
ExpressionAttributeValues={
":cutoff": Decimal(str(stale_cutoff)),
":min": 3,
":threshold": Decimal(str(min_success_rate)),
}
)
flagged_ids = []
for item in response.get("Items", []):
table.update_item(
Key={"procedure_id": item["procedure_id"]},
UpdateExpression="SET needs_review = :v",
ExpressionAttributeValues={":v": True},
)
flagged_ids.append(item["procedure_id"])
return flagged_ids
Perdre un épisode entraîne la perte d’un peu de contexte, tandis que la suppression silencieuse d’une procédure peut éliminer une approche soigneusement développée et validée par des humains, d’où cette asymétrie. flag_stale_procedures utilise un balayage de tableau, ce qui convient à des tâches batch planifiées (comme Lambda) mais pas aux chemins fréquemment empruntés.
Intégration au début et à la fin de l’exécution
Tous les niveaux se rencontrent au début de chaque exécution. La classe bootstrap interroge le stockage des épisodes pour trouver jusqu’à trois épisodes similaires, la base de connaissances pour obtenir jusqu’à cinq faits liés au domaine, et le stockage des procédures pour identifier une procédure validée correspondant au type de tâche. Elle enrobe ensuite tout ce qui est trouvé dans des marqueurs clairement délimités afin que le modèle puisse distinguer la mémoire récupérée de la tâche elle-même :
# harness/memory/bootstrap.py
from typing import Optional
from langchain_core.messages import SystemMessage
from harness.memory.episodic import EpisodicMemoryStore
from harness.memory.semantic import SemanticMemoryStore
from harness.memory.procedural import ProceduralMemoryStore
class MemoryBootstrap:
"""
Assembles the memory context for a new agent run.
Called once at startup, before the agent sees the task.
"""
def __init__(
self,
agent_id: str,
knowledge_base_id: str,
s3_bucket: str,
region: str = "us-east-1",
):
self.episodic = EpisodicMemoryStore(agent_id, region=region)
self.semantic = SemanticMemoryStore(knowledge_base_id, s3_bucket, region=region)
self.procedural = ProceduralMemoryStore(region=region)
def build_memory_context(
self,
task: str,
task_type: Optional[str] = None,
domain: Optional[str] = None,
) -> str:
"""
Retrieves relevant memories from all tiers and assembles
them into a single context block for system prompt injection.
"""
sections = []
# Episodic: what happened on similar tasks before
episodes = self.episodic.retrieve_relevant_episodes(task, max_results=3)
if episodes:
sections.append(self.episodic.format_for_injection(episodes))
# Semantic: what is known about this domain
if domain:
facts = self.semantic.retrieve(task, domain=domain, max_results=5)
if facts:
sections.append(
"[SEMANTIC MEMORY: domain knowledge]\n" +
"\n".join(f"- {f}" for f in facts)
)
# Procedural: how to approach this task type
if task_type:
procedure = self.procedural.get_procedure(task_type)
if procedure:
sections.append(self.procedural.format_for_injection(procedure))
if not sections:
return ""
return (
"\n\n[MEMORY CONTEXT - retrieved from prior runs and domain knowledge]\n\n" +
"\n\n".join(sections) +
"\n\n[END MEMORY CONTEXT]\n"
)
Le chemin d’écriture est délibérément séparé du chemin de lecture. À la fin d’une exécution, le consolideur extrait les artefacts de la mémoire de travail, enregistre l’épisode, n’extrait et ne stocke les faits sémantiques que lorsque l’exécution est terminée, et propose une procédure qui attendra la validation. Il utilise un modèle plus volumineux que celui utilisé pour la compression en cours d’exécution, car cette étape influence le comportement futur ; assurez-vous que l’ID du modèle est encore disponible dans votre région. Le fragment utilise également Optional sans l’importer depuis typing, il convient donc d’ajouter cette importation si vous le réutilisez :
# harness/memory/consolidator.py
from harness.memory.working_memory import WorkingMemoryManager
from harness.memory.episodic import EpisodicMemoryStore
from harness.memory.semantic import SemanticMemoryStore
from harness.memory.procedural import ProceduralMemoryStore
from langchain_aws import ChatBedrock
import boto3
class MemoryConsolidator:
"""
Runs at the end of each agent run to distill and store memories.
The write path is separate from the read path by design.
"""
def __init__(
self,
agent_id: str,
knowledge_base_id: str,
s3_bucket: str,
region: str = "us-east-1",
):
bedrock = boto3.client("bedrock-runtime", region_name=region)
self.extractor = ChatBedrock(
client=bedrock,
model_id="anthropic.claude-3-5-sonnet-20241022-v2:0",
model_kwargs={"temperature": 0, "max_tokens": 4000},
)
self.working = WorkingMemoryManager(region=region)
self.episodic = EpisodicMemoryStore(agent_id, region=region)
self.semantic = SemanticMemoryStore(knowledge_base_id, s3_bucket, region=region)
self.procedural = ProceduralMemoryStore(region=region)
def consolidate(
self,
final_state: dict,
task_type: Optional[str] = None,
domain: Optional[str] = None,
):
"""
End-of-run consolidation. Extracts and stores memories from all tiers.
"""
messages = final_state.get("messages", [])
# Step 1: Extract run artifacts from working memory
artifacts = self.working.extract_run_artifacts(messages)
# Step 2: Store as episodic memory
run_id = self.episodic.store_episode(artifacts)
artifacts["run_id"] = run_id
# Step 3: Extract and store semantic facts (only from successful runs)
if domain and artifacts.get("outcome") == "COMPLETED":
facts = self.semantic.extract_facts_from_episode(
artifacts, self.extractor, domain
)
for fact in facts:
self.semantic.store_fact(fact)
# Step 4: Propose procedural memory update
if task_type:
self.procedural.propose_procedure(task_type, artifacts, self.extractor)
Renforcement de la production
Protection contre le poisonage de la mémoire
Une série d’essais défaillants peut inscrire une information erronée dans la mémoire sémantique ou proposer une procédure défectueuse, et à grande échelle l’erreur s’amplifie, car chaque essai ultérieur qui récupère cette information ou suit cette procédure l’hérite. Les mécanismes de protection sont stratifiés :
- Écrire uniquement à partir d’essais
COMPLETEDdont les résultats présentent un degré élevé de fiabilité. - Exiger des observations répétées et indépendantes avant de valider une information : un seul essai ne constitue qu’une suggestion, tandis que l’accord entre trois essais représente une preuve réelle.
- Faire passer un contrôle humain avant que toute procédure n’atteigne le statut de validation.
Une manière simple d’implémenter le deuxième niveau de protection consiste à utiliser un compteur de confirmation basé sur l’hash du contenu de l’information. Chaque observation incrémente atomiquement ce compteur avec if_not_exists, et l’information n’est validée que lorsque le compteur atteint le seuil fixé :
# Simple confirmation counter before promoting a fact
def promote_fact_if_confirmed(
fact_content: str,
confirmation_threshold: int = 3,
table_name: str = "fact-confirmation-log",
region: str = "us-east-1",
):
"""
Tracks how many independent runs have observed the same fact.
Only writes to semantic memory after threshold is reached.
"""
table = boto3.resource("dynamodb", region_name=region).Table(table_name)
import hashlib
fact_hash = hashlib.sha256(fact_content.encode()).hexdigest()[:16]
response = table.update_item(
Key={"fact_hash": fact_hash},
UpdateExpression="SET confirmation_count = if_not_exists(confirmation_count, :z) + :inc, content = :c",
ExpressionAttributeValues={":z": 0, ":inc": 1, ":c": fact_content},
ReturnValues="UPDATED_NEW",
)
return int(response["Attributes"]["confirmation_count"]) >= confirmation_threshold
Comme le texte exact est haché, les versions formulées différemment d’un même fait sont comptées séparément ; normalisez ou regroupez d’abord les faits si les promotions deviennent trop rares.
Éviter de conserver des données PII en mémoire
La mémoire de l’agent collectera des informations personnellement identifiables à moins que vous ne l’en empêchiez activement. Un agent d’analyse de contrats stocke des informations sur ces derniers, et les contrats mentionnent des personnes, des salaires ainsi que d’autres détails personnels. La méthode la plus propre consiste à nettoyer le texte avant d’écrire quoi que ce soit dans les niveaux épisodiques ou sémantiques. Amazon Comprehend propose une API de détection des données PII qui fonctionne bien sur le texte en anglais. La fonction ci-dessous remplace chaque entité détectée par son étiquette de type, en itérant depuis la fin de la chaîne afin que les positions précédentes restent valides :
import boto3
def scrub_pii_before_storage(text: str, region: str = "us-east-1") -> str:
comprehend = boto3.client("comprehend", region_name=region)
response = comprehend.detect_pii_entities(Text=text[:5000], LanguageCode="en")
scrubbed = text
# Replace detected PII with type labels, working backwards to preserve offsets
for entity in sorted(response["Entities"], key=lambda x: x["BeginOffset"], reverse=True):
scrubbed = (
scrubbed[:entity["BeginOffset"]] +
f"[{entity['Type']}]" +
scrubbed[entity["EndOffset"]:]
)
return scrubbed
Veillez à savoir que l’appel ne prend en compte que les 5 000 premiers caractères, de sorte que tout ce qui dépasse cette limite est stocké sans avoir été nettoyé. Les textes plus longs doivent être divisés en blocs, chacun étant nettoyé séparément. Les types d’entités pris en charge ainsi que leurs limites sont listés dans la documentation Comprehend PII.
Isolement des locataires
Si les agents servent plusieurs clients sur une infrastructure partagée, leurs mémoires doivent être complètement isolées : un traitement pour un client ne doit en aucun cas récupérer les mémoires d’un autre client. Le schéma le plus simple consiste à préfixer chaque clé de mémoire par un identifiant de locataire et à imposer cette séparation au niveau du stockage, par exemple à l’aide d’expressions de condition DynamoDB, plutôt que de compter uniquement sur le filtrage au niveau de l’application :
# Always include tenant_id in all memory operations
def store_episode_isolated(tenant_id: str, agent_id: str, artifacts: dict):
namespaced_agent_id = f"{tenant_id}#{agent_id}"
# All subsequent operations use namespaced_agent_id, never bare agent_id
Puisque agent_id est la clé de partition épisodique, le préfixe permet de garder les requêtes normales au sein d’un même locataire. Appliquez le même schéma de nomenclature à la base de connaissances, au tableau procédural et au journal de confirmation.
À quoi s’attendre une fois la mémoire activée
La mémoire améliore de manière mesurable les performances des agents avec le temps, mais elle introduit également des modes de défaillance plus difficiles à diagnostiquer que ceux générés par un système sans état :
- Des mémoires défectueuses sont pires que l’absence totale de mémoire. Un agent sans état confus gâche une seule exécution. Un agent disposant d’une information fausse gâche toutes les exécutions qui l’utilisent, éventuellement pendant des mois. Investissez dans des mécanismes de validation avant d’élargir la couverture mémoire.
Architecture de référence
Le cycle de vie complet d’une exécution, du démarrage à la consolidation dans les trois bases de données persistantes, se présente comme suit :
New Run Starts
|
v
+-------------------------------+
| Memory Bootstrap |
| 1. Query Episodic Store |
| 2. Query Bedrock KB |
| 3. Query Procedural Store |
| 4. Assemble context block |
+-------------------------------+
|
v
+-------------------------------+
| Agent Run (LangGraph) |
| Working Memory Manager |
| Compression at threshold |
| Tool calls with auth/retry |
+-------------------------------+
|
v
+-------------------------------+
| Memory Consolidator |
| 1. Extract run artifacts |
| 2. Scrub PII |
| 3. Write episodic record |
| 4. Propose semantic facts |
| (if confirmed: store) |
| 5. Propose procedure |
| (pending validation) |
+-------------------------------+
|
v
+------------------+ +------------------+ +------------------+
| DynamoDB | | Bedrock KB | | DynamoDB |
| Episodic Memory | | Semantic Memory | | Procedural Mem. |
| TTL: 90 days | | S3 backed | | Validated flag |
+------------------+ +------------------+ +------------------+
Et voici l’ensemble de la pile d’infrastructure, y compris l’alternative de développement local basée sur Ollama, ChromaDB et Docker Compose, qui vous permet de tester la pile mémoire sans frais AWS :
+-----------------------------+---------------------+------------------------------+
| Component | Technology | Role |
+-----------------------------+---------------------+------------------------------+
| Working Memory | LangGraph State + | Active context compression, |
| | MemorySaver | sliding window management |
+-----------------------------+---------------------+------------------------------+
| Episodic Memory Store | DynamoDB + TTL | Past run summaries, semantic |
| | | similarity retrieval |
+-----------------------------+---------------------+------------------------------+
| Episodic Embeddings | Titan Embed v2 | Semantic similarity search |
| | (Bedrock) | over episode store |
+-----------------------------+---------------------+------------------------------+
| Semantic Memory | Bedrock Knowledge | Domain facts, structured |
| | Bases + S3 | knowledge, write via ingest |
+-----------------------------+---------------------+------------------------------+
| Procedural Memory | DynamoDB | Learned task procedures, |
| | | validated flag, success rate |
+-----------------------------+---------------------+------------------------------+
| PII Scrubbing | AWS Comprehend | Before any memory write, |
| | | English PII detection |
+-----------------------------+---------------------+------------------------------+
| Fact Confirmation | DynamoDB | Multi-run confirmation gate |
| | | before semantic write |
+-----------------------------+---------------------+------------------------------+
| Memory Decay | DynamoDB TTL + | Episode pruning, stale |
| | scheduled Lambda | procedure flagging |
+-----------------------------+---------------------+------------------------------+
| Context Compression | Claude Haiku 4.5 | Working memory summarization |
+-----------------------------+---------------------+------------------------------+
| Artifact Extraction | Claude 3.5 Sonnet | End-of-run distillation |
+-----------------------------+---------------------+------------------------------+
| Local Dev Alternative | Ollama + ChromaDB | Full memory stack testable |
| | Docker Compose | without AWS costs |
+-----------------------------+---------------------+------------------------------+
| Infrastructure as Code | Terraform | DynamoDB tables, S3 bucket, |
| | | KB data source, IAM roles |
+-----------------------------+---------------------+------------------------------+
Points clés
- Considérez la fenêtre de contexte comme une ardoise et créez un carnet de notes distinct et persistant pour les connaissances qui doivent survivre à une exécution.
- Séparer la mémoire en niveaux de travail, épisodique, sémantique et procédural, car chacun nécessite un stockage, des mécanismes de récupération et une gestion des pannes différents.
- Résumer les exécutions en comptes-rendus et faits concis ; ne jamais stocker ou rechercher des journaux bruts en tant que mémoire.
- Mettre en place les contrôles les plus stricts sur les niveaux qui modifient le comportement le plus : la confirmation pour les faits, la validation humaine pour les procédures.
- Supprimer toutes les informations personnelles identifiables et assurer l’isolation des utilisateurs au niveau du stockage avant même d’écrire la première donnée dans la mémoire.
- Allouer dès le début des ressources pour gérer l’oubli, le suivi de la récupération, les retards d’ingestion et une réintégration éventuelle.