Diseño de memoria para agentes de cuatro capas con LangGraph y Amazon Bedrock
Aprenda a dotar a los agentes de LLM de memoria episódica, semántica y procedural funcional en Bedrock y LangGraph, así como a protegerla contra el envenenamiento, las fugas de PII y la fuga de datos entre usuarios.
Fuentes: Agentic Architectures — Article 7: Agent Memory Architectures
Tarde o temprano, un agente de producción comienza a hacer el mismo trabajo dos veces. No hay nada roto y la solicitud está bien; simplemente el agente no recuerda haber resuelto un problema casi idéntico la semana pasada, por lo que sigue los mismos pasos, repite los mismos errores intermedios, llega a la misma respuesta, y usted tiene que pagar de nuevo por cada token. Esta guía muestra cómo solucionar eso con una infraestructura de memoria real en lugar de una ventana de contexto más grande: cuatro niveles de memoria distintos creados con LangGraph en Amazon Bedrock, además de las medidas de seguridad para evitar que la memoria se convierta en un problema.
Imagínese un agente de análisis de contratos con una estructura sólida y una configuración multiagente funcional que aún así dedica sus primeras llamadas a herramientas en cada contrato similar a redescubrir patrones estructurales que ya había analizado anteriormente. Lo que le falta es memoria.
Una ventana de contexto más grande no es memoria
Cuando el contexto se llena, la reacción habitual es utilizar una ventana más grande o recurrir a una resumen más agresivo. Ambos métodos son útiles durante una sesión; sin embargo, ninguno constituye memoria. Una analogía útil: la ventana de contexto es una pizarra blanca; resulta práctica durante una reunión y se borra al finalizarla. La memoria, en cambio, es el cuaderno que llevas de una reunión a otra.
La gestión del contexto tiene limitaciones en tres aspectos específicos:
- No conecta las sesiones entre sí. Cuando el agente termina y su proceso finaliza, el contexto desaparece. Todo lo aprendido se pierde a menos que se guarde explícitamente.
- No trasciende los límites de los agentes. En un sistema multiagente, cada trabajador tiene su propio contexto. Lo que descubre un trabajador encargado de la revisión de seguridad es invisible para uno dedicado al análisis de código, a menos que se dirija intencionadamente esa información hacia allí.
La memoria aborda una necesidad diferente: un almacén de conocimiento persistente y consultable que sobrevive a las ejecuciones individuales y puede compartirse en cualquier topología de agentes que se utilice.
Los cuatro niveles de la memoria del agente
Cada nivel implica una infraestructura, patrones de recuperación y modos de fallo diferentes:
+---------------------+--------------------------------------------------+----------------------------+
| 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 | |
+---------------------+--------------------------------------------------+----------------------------+
Los niveles interactúan en un ciclo. La memoria de trabajo almacena la ejecución en curso. Cuando esta finaliza, un paso de destilación escribe los datos en los tres niveles persistentes. Cuando comienza la siguiente ejecución, un paso de recuperación extrae el material relevante:
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 memoria episódica indica en la siguiente ejecución qué ocurrió, la memoria semántica lo que se sabe sobre el dominio, y la memoria procedural cómo abordar la tarea. Las tres se recuperan al iniciar y se colocan en la memoria de trabajo antes de que el agente realice su primer paso de razonamiento.
Nivel 1: gestionar la memoria de trabajo durante una ejecución
La memoria de trabajo es todo lo que existe durante una sola ejecución, y ya la estás gestionando independientemente de si le pones ese nombre o no. En LangGraph, el estado del grafo es la memoria de trabajo: cada mensaje, salida de herramienta y resultado intermedio se almacena allí.
El problema es que sigue creciendo. La compresión de contexto entre las etapas del pipeline, una técnica común en sistemas multiagente, es en realidad gestión de memoria de trabajo bajo otro nombre. La clase que se muestra a continuación convierte esa idea en un componente reutilizable. maybe_compress se ejecuta antes de cada paso del agente; una vez que la lista de mensajes supera un umbral (30 por defecto), mantiene los mensajes del sistema y los 15 mensajes no relacionados con el sistema más recientes tal cual, y reemplaza todo lo anterior por un único resumen generado por un modelo pequeño y económico. Un segundo método, extract_run_artifacts, se ejecuta al final de la ejecución y solicita al modelo un registro estructurado en JSON con información sobre la tarea, el resultado, los hallazgos, el enfoque, las herramientas y los patrones reutilizables, el cual se convierte en la entrada para la memoria episódica. El fragmento está escrito en Python, no 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": []}
La instrucción de resumen descarta los intentos fallidos y los errores del herramienta, y el análisis JSON recurre a un resultado UNKNOWN con listas vacías, por lo que una respuesta mal formada nunca interrumpe el proceso. Los identificadores del modelo, como el ID de Haiku que se muestra aquí, cambian con el tiempo y pueden requerir un perfil de inferencia específico para la región; por eso, verifique la lista actual de modelos Bedrock antes de copiarlos.
Para el desarrollo local, puede reemplazar el resumidor de Bedrock por un modelo proporcionado por Ollama:
# Local dev alternative
# ollama pull llama3.1:8b
from langchain_ollama import ChatOllama
summarizer = ChatOllama(model="llama3.1:8b", temperature=0, format="json")
Nivel 2: memoria episódica de ejecuciones anteriores
La memoria episódica responde a la pregunta “¿qué sucedió antes?”. Al inicio de una ejecución, el agente obtiene resúmenes de ejecuciones anteriores similares a la tarea actual y los utiliza para omitir trabajos redundantes y evitar errores conocidos.
El diseño almacena resúmenes condensados en lugar de registros brutos de ejecución. Los registros completos son costosos de mantener y lentos para buscar; los resúmenes compactos son económicos y rápidos. El almacenamiento se realiza en DynamoDB, y la recuperación utiliza embeddings de Bedrock para la similitud semántica.
La tabla, definida en Terraform, utiliza agent_id como clave de partición y run_id como clave de ordenamiento. Un índice secundario global en agent_id y created_at permite realizar consultas como “los episodios más recientes de este agente”, y el TTL de DynamoDB en expires_at elimina automáticamente los registros antiguos:
# 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"
}
}
La tienda misma incrusta el resumen de la tarea junto con los hallazgos clave, escribe el registro con una fecha de vencimiento de 90 días y, al recuperarlo, compara la incrustación de la tarea actual con episodios recientes utilizando similitud coseno. Solo se conservan los episodios cuya similitud supera el umbral de 0.75; en caso de empate, se prefiere el resultado COMPLETED; además, format_for_injection muestra a los ganadores como un bloque etiquetado para el prompt del sistema:
# 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 consulta obtiene como máximo 100 episodios recientes y calcula la similitud en el código de la aplicación; como indica el comentario, las implementaciones más grandes deberían usar paginación y una base de datos vectorial. Tenga también en cuenta que el recurso DynamoDB de boto3 rechaza los valores float de Python, por lo que la lista de embeddings debe convertirse a Decimal (o serializarse, por ejemplo como JSON o binario) antes de que put_item la acepte. El modelo de embeddings utilizado aquí se describe en la documentación de Amazon Titan Embeddings.
Nivel 3: memoria semántica con bases de conocimiento Bedrock
La memoria episódica registra eventos; la memoria semántica registra conocimientos. La diferencia es sutil pero importante. Un episodio podría decir: “Un contrato con un proveedor analizado el martes pasado tenía una cláusula 7.3 no estándar”. El hecho semántico correspondiente dice: “La cláusula 7.3 en los contratos con proveedores suele referirse a límites de responsabilidad; el rango normal es el doble del valor del contrato, y cualquier cantidad inferior a ese valor es una señal de alerta”.
Los hechos semánticos tienen una duración mayor que los episodios. No caducan después de 90 días; se refinan con el tiempo y son útiles en muchos tipos de tareas, no solo en aquellas similares a ejecuciones anteriores.
Bedrock Knowledge Bases es el lugar ideal para este nivel, ya que maneja la división en fragmentos, el incrustado, el almacenamiento y la recuperación de información. Lo que la mayoría de los equipos aún no ha desarrollado es la vía de escritura, ya que las bases de conocimiento suelen tratarse como de solo lectura. La solución presentada abarca ambas direcciones:
retrieverealiza una búsqueda vectorial semántica, opcionalmente filtrada por una clave de metadatosdomain, y descarta los resultados con una puntuación inferior a 0.6.store_factescribe el hecho como un documento JSON en S3 e inicia un proceso de ingestión para que la Base de Conocimientos lo recoja.extract_facts_from_episodepide a un modelo que extraiga hechos generalizables de una ejecución completada, omitiendo los detalles específicos de la tarea que pertenecen a la memoria episódica y conservando únicamente los hechos con un nivel de confianza de al menos 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"]
Los nuevos datos no se pueden recuperar de inmediato. El proceso de ingestión lleva algo de tiempo, generalmente unos minutos según el comentario del código, por lo que las cargas de trabajo sensibles al tiempo necesitan una búsqueda rápida (por ejemplo, en DynamoDB) además de la Base de Conocimientos. Hay dos puntos más que merecen verificación contra la documentación de ingestión de la Base de Conocimientos antes de confiar en este código. Primero, el filtrado de metadatos en las Bases de Conocimientos funciona con los atributos de metadatos proporcionados en un archivo de metadatos separado junto a cada documento S3, no con los campos incrustados dentro del cuerpo del documento, por lo que el filtro domain podría no coincidir con nada tal como está escrito. Segundo, iniciar un trabajo de ingestión para cada dato individual puede chocar con otro trabajo que ya esté en ejecución sobre la misma fuente de datos; agrupar las escrituras y sincronizarlas según un horario suele ser más eficaz.
Nivel 4: memoria procedural, con una puerta de validación
La memoria procedural es tanto el nivel más poderoso como el más peligroso. Almacena cómo hacer algo, no solo lo que se hizo: un enfoque exitoso para un tipo de tarea recurrente se transforma en una lista reutilizable de pasos que se integra en las solicitudes futuras del sistema.
El peligro radica en que un procedimiento defectuoso, una vez almacenado, se siga sin cuestionamientos en cada tarea posterior. Por eso este nivel necesita una puerta de validación en la ruta de escritura. En el almacén a continuación, propose_procedure solo se ejecuta para ejecuciones completadas y guarda nuevos procedimientos con validated establecido en false, lo que significa que están pendientes. Otro agente o un humano debe aprobar un procedimiento antes de que get_procedure lo devuelva, y además get_procedure requiere al menos tres aplicaciones previas. record_application_result registra con qué frecuencia se ha aplicado un procedimiento y cuántas veces ha tenido éxito, mientras que format_for_injection presenta los pasos junto con la tasa de éxito y un recordatorio de que son un punto de partida, no 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)
Considere este fragmento como un boceto de diseño y no como código listo para usar, ya que algunos detalles requieren ajustes. Las expresiones de actualización de DynamoDB solo admiten suma y resta, por lo que la división de success_rate debe calcularse en el código de la aplicación (o al momento de la lectura) en lugar de dentro de UpdateExpression. La consulta utiliza task_type como clave, mientras que las actualizaciones emplean procedure_id; por eso la tabla necesita un esquema de clave o un índice secundario que soporte ambos patrones de acceso. Las marcas de tiempo en formato flotante y success_rate requieren la misma conversión a Decimal mencionada anteriormente. Finalmente, combinar “validado” con “aplicado al menos tres veces” puede generar una situación de tipo “huevo y gallina”: hay que decidir cómo una procedimiento recién validado obtiene sus primeras aplicaciones, por ejemplo a través de un modo de prueba o una excepción aprobada por un humano.
Olvidar intencionadamente: el deterioro de la memoria
Mantener cada recuerdo para siempre genera dos problemas. El almacenamiento de cientos de miles de resúmenes de ejecuciones en DynamoDB y S3 conlleva un costo real. De manera más sutil, los recuerdos obsoletos inducen a error de forma activa: un patrón que funcionó hace seis meses podría estar equivocado ahora porque el dominio ha evolucionado. Un sistema de olvido bien diseñado es una ventaja, no un defecto.
El gestor de deterioro aplica una política diferente según el nivel. Para episodios anteriores a un período mínimo de retención (siete días por defecto), elimina los registros cuyo resultado fue FAILED. En el caso de procedimientos, marca en lugar de eliminar: todo lo que no se haya actualizado en 60 días, o todo lo que se haya utilizado al menos tres veces con una tasa de éxito inferior al 0.6, recibe un marcador needs_review para que lo examine un humano. Este código es en Python, a pesar de estar etiquetado como 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
Perder un episodio implica perder algo de contexto, mientras que eliminar silenciosamente un procedimiento podría descartar un enfoque obtenido con esfuerzo y validado por humanos, de ahí la asimetría. flag_stale_procedures utiliza un escaneo de tabla, lo cual es adecuado para tareas por lotes programadas (como Lambda) pero no para rutas de alto tráfico.
Integración al inicio y final de la ejecución
Todos los niveles se reúnen al comienzo de cada ejecución. La clase de arranque consulta el almacenamiento de episodios en busca de hasta tres episodios similares, la Base de Conocimientos en busca de hasta cinco hechos relacionados con el dominio, y el almacenamiento de procedimientos en busca de un procedimiento validado que se ajuste al tipo de tarea. Luego envuelve todo lo encontrado en marcadores claramente delimitados para que el modelo pueda distinguir la memoria recuperada de la tarea en sí:
# 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"
)
La ruta de escritura está deliberadamente separada de la ruta de lectura. Al final de una ejecución, el consolidador extrae los artefactos de la memoria de trabajo, almacena el episodio, extrae y almacena los hechos semánticos solo cuando la ejecución finaliza, y propone un procedimiento que esperará la validación. Utiliza un modelo más grande que el de compresión durante la ejecución, ya que este paso determina el comportamiento futuro; verifique que el ID del modelo siga estando disponible en su región. El fragmento también utiliza Optional sin importarlo desde typing, así que agregue esa importación si lo vuelve a utilizar:
# 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)
Reforzamiento en producción
Protección contra el envenenamiento de memoria
Una ejecución defectuosa puede registrar un hecho incorrecto en la memoria semántica o proponer un procedimiento defectuoso, y a gran escala el error se amplifica, ya que cada ejecución futura que recupere ese hecho o siga dicho procedimiento lo hereda. Las defensas son en capas:
- Solo registrar datos de ejecuciones
COMPLETEDcuyos resultados tengan un alto nivel de confianza. - Requerir observaciones repetidas e independientes antes de promover un hecho: una sola ejecución solo sugiere algo, mientras que el consenso en tres ejecuciones constituye evidencia real.
- Incluir una revisión humana antes de que cualquier procedimiento alcance un estado validado.
Una forma sencilla de implementar la segunda capa es utilizar un contador de confirmación basado en el hash del contenido del hecho. Cada observación incrementa atómicamente el contador con if_not_exists, y el hecho solo se promueve una vez que la cuenta alcanza el umbral establecido:
# 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
Dado que el texto exacto se hashea, las versiones con palabras diferentes de un mismo hecho se contabilizan por separado; normalice o agrupe los hechos primero si las promociones resultan ser demasiado escasas.
Mantener la PII fuera de la memoria
La memoria del agente recopilará información de identificación personal a menos que usted lo impida activamente. Un agente de análisis de contratos almacena datos sobre contratos, y estos mencionan a personas, salarios y otros detalles personales. El enfoque más adecuado es limpiar el texto antes de escribir cualquier cosa en los niveles episódico o semántico. Amazon Comprehend ofrece una API de detección de PII que funciona bien con texto en inglés. La función a continuación reemplaza cada entidad detectada por su etiqueta de tipo, iterando desde el final de la cadena para que los índices anteriores sigan siendo válidos:
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
Tenga en cuenta que la llamada solo examina los primeros 5,000 caracteres, por lo que todo lo que está más allá se almacena sin procesar. Los textos más largos deben dividirse en fragmentos y cada fragmento debe procesarse por separado. Los tipos de entidades admitidos y sus límites se detallan en la documentación de Comprehend PII.
Aislamiento de usuarios
Si los agentes atienden a varios clientes en una infraestructura compartida, sus memorias deben estar completamente aisladas: un proceso para un cliente nunca debe recuperar las memorias de otro cliente. El patrón más sencillo es asignar un nombre de espacio de nombres a cada clave de memoria con un identificador de usuario y establecer los límites en la capa de almacenamiento, por ejemplo mediante expresiones condicionales de DynamoDB, en lugar de depender únicamente del filtrado a nivel de aplicación:
# 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
Dado que agent_id es la clave de partición episódica, el prefijo mantiene las consultas normales dentro de un mismo arrendatario. Aplique el mismo esquema de nombres a la Base de Conocimientos, la tabla procedural y el registro de confirmaciones.
Qué esperar una vez que la memoria esté activa
La memoria hace que los agentes mejoren de manera significativa con el tiempo, pero también introduce modos de fallo que son más difíciles de depurar que cualquier cosa que produzca un sistema sin estado:
- Las memorias defectuosas son peores que no tener ninguna. Un agente sin estado confundido arruina una ejecución. Un agente con datos erróneos arruina todas las ejecuciones que los recuperan, posiblemente durante meses. Invierta en mecanismos de validación antes de ampliar la cobertura de memoria.
Arquitectura de referencia
El ciclo de vida completo de una ejecución, desde el arranque hasta su consolidación en los tres almacenamientos persistentes, se ve así:
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 |
+------------------+ +------------------+ +------------------+
Y esta es la estructura completa de infraestructura, incluyendo la alternativa para desarrollo local basada en Ollama, ChromaDB y Docker Compose que permite probar la pila de memoria sin costos de 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 |
+-----------------------------+---------------------+------------------------------+
Puntos clave
- Trate la ventana de contexto como una pizarra y cree un cuaderno separado y persistente para el conocimiento que debe sobrevivir a una ejecución.
- Divida la memoria en niveles de trabajo, episódico, semántico y procedural, ya que cada uno requiere un almacenamiento, recuperación y manejo de fallos diferentes.
- Reduzca los procesos a resúmenes y hechos compactos; nunca almacene ni busque registros en bruto como memoria.
- Coloque las medidas de control más estrictas en los niveles que modifican el comportamiento con mayor intensidad: la confirmación es importante para los hechos, mientras que la validación humana lo es para los procedimientos.
- Limpie la información de identificación personal y aplique aislamiento entre usuarios en la capa de almacenamiento antes de escribir cualquier memoria.
- Asigne presupuesto desde el primer día para el olvido, la supervisión de la recuperación, los retrasos en la incorporación de datos y su reintegración eventual.
Lecturas relacionadas
- Dentro de un agente AI: Modelos, herramientas, memoria y razonamiento explicados — Explica la arquitectura básica de los agentes AI: objetivos, modelos, herramientas, memoria y razonamiento, y presenta un ejemplo práctico de un agente de investigación basado en Python.
- Diseño de grafos de agentes AI resilientes: reintentos, soluciones de recursión y GraphRAG — Aprenda cómo crear flujos de trabajo de agentes AI tolerantes a fallos utilizando ramas de error explícitas, lógica de reintentos, LangGraph, y cuándo GraphRAG supera a los métodos RAG tradicionales o a los bucles de agente.