Un diseño de memoria de cuatro capas para agentes de IA con trazabilidad
División entre memoria a corto y largo plazo, cuatro capas (búfer, episódica, semántica, procedural), opciones de almacenamiento, escrituras asíncronas y medidas de protección contra el envenenamiento de la memoria.
Imagínese un agente que falla en el paso catorce de un largo flujo de trabajo. La culpa no es del modelo; simplemente el agente ya no recuerda lo que ha hecho anteriormente. El fallo sigue un patrón familiar: las llamadas a herramientas son correctas y las instrucciones son claras, pero una vez que la conversación supera cierta longitud invisible, el agente comienza a repetirse. Repite preguntas ya respondidas, obtiene datos que ya posee y se contradice entre el paso seis y el paso doce, como si los primeros cinco pasos nunca hubieran tenido lugar.
Sustituir el modelo por otro rara vez cambia esta situación. El cuello de botella está en la infraestructura que rodea al modelo, y más concretamente en la falta de una verdadera arquitectura de memoria. Pegar los últimos N mensajes en las instrucciones funciona para un chatbot, pero falla cuando un agente debe ejecutar un flujo de trabajo de veinte pasos y recordar en el paso diecisiete lo que decidió en el paso tres.
Un patrón que aborda este problema es una pila de memoria de cuatro capas, con el almacenamiento adecuado detrás de cada capa. A continuación se explica qué hace cada capa, cómo los equipos de GitHub, Anthropic y Berkeley describen sus versiones, y se incluyen ejemplos en Python para cada componente.
Una ventana de contexto no es memoria
Una ventana de contexto es memoria de trabajo: RAM, no un disco duro. Cada token que contiene tiene un costo. Poner toda la historia de una conversación en una ventana de 128K parece seguro hasta que se utiliza un agente con muchas herramientas. Un flujo de trabajo de treinta pasos con llamadas a herramientas, observaciones y rastros de razonamiento consume rápidamente 100K tokens, y cuando la ventana se llena, el contenido más antiguo es eliminado del principio, que suele ser donde se encuentran las restricciones originales del usuario y la definición de la tarea.
Un problema más sutil es aún peor. Un estudio realizado por investigadores de Stanford y UC Berkeley, publicado en una revista de ACL en 2024 bajo el título “Perdidos en el medio: cómo utilizan los modelos de lenguaje contextos largos”, reveló que el rendimiento de los LLM sigue una curva en forma de U a lo largo de la ventana de contexto. Los modelos rinden bien con la información situada al principio y al final del contexto, pero claramente peor con la información oculta en el medio. Este efecto se observó en diversas arquitecturas, tamaños de modelo y longitudes de contexto, con ventanas probadas desde 4K hasta 32K tokens. Un estudio de seguimiento de 2025 realizado por MIT ofreció una explicación matemática, sosteniendo que tanto el enmascaramiento causal de la atención como la codificación de posición rotativa reducen el peso de la atención en los tokens alejados de los extremos de la secuencia.
Para los agentes, la consecuencia es clara: incluso si todo encaja, el modelo no lo tratará de manera uniforme. Una restricción crítica al llegar a los 30K tokens en un contexto de 60K tokens tiene grandes posibilidades de quedar prácticamente invisible.
El modelo olvida porque el sistema le suministra más información de la que puede procesar de forma fiable, y no existe ningún mecanismo que controle qué entra en su ventana de procesamiento.
La división inicial: memoria a corto y largo plazo
Antes de la taxonomía de cuatro capas, existe una división más simple que la mayoría de los ingenieros tienen en mente. La memoria en los sistemas de IA se divide en a corto y largo plazo, y comprender las limitaciones de cada una motiva el diseño de arquitecturas más detalladas.
La memoria a corto plazo está limitada a una sesión y solo existe durante una interacción. Almacena lo que el agente necesita en ese momento: el estado de la tarea actual, la etapa en la que se encuentra, la salida más reciente de una herramienta y el rastro de razonamiento que se está generando. En la práctica, eso corresponde a la ventana de contexto, la cual es altamente dinámica, se actualiza constantemente, tiene un costo elevado por token y se elimina al finalizar la sesión.
La documentación de LangChain lo explica bien, describiendo la memoria a corto plazo como un almacenamiento del historial de conversaciones, las salidas de herramientas y el estado temporal de razonamiento. Lo que no menciona es que esto también es lo primero en verse afectado a gran escala. En un flujo de trabajo con veinte pasos del agente, las salidas intermedias, las reflexiones, los resultados de búsquedas y las respuestas de las herramientas se acumulan y compiten por el mismo espacio limitado. Es similar a intentar mantener un plan complicado en la mente mientras alguien sigue interrumpiendo con nuevos datos.
La memoria a corto plazo tiene dos componentes prácticos:
- Estatus de la tarea es el registro en tiempo real del progreso: el paso actual, las subtareas pendientes, el estado de ejecución de las herramientas y la coordinación entre los agentes. Un agente encargado de planificar viajes mantiene la destinación, los límites presupuestarios, los vuelos explorados y los hoteles seleccionados para no tener que volver a realizar el trabajo del paso dos cuando llegue al paso ocho.
- Salidas intermedias son los resultados del razonamiento durante el proceso: resultados de herramientas, resúmenes generados, salidas SQL, fragmentos de búsquedas y notas temporales. El equipo de ingeniería de Anthropic ha descrito un patrón en los sistemas en producción según el cual los agentes que guardan notas fuera de la ventana de contexto y las recuperan más tarde obtienen mejores resultados en tareas largas que aquellos que intentan mantener todo dentro del contexto. Tomar notas es en sí misma una estrategia de memoria a corto plazo.
La memoria a largo plazo sobrevive más allá de las sesiones. Permite que un agente recuerde que a un usuario le gustan respuestas concisas, que el despliegue de la semana pasada falló debido a un problema de migración, o que el esquema de la base de datos cambió en enero. LangChain la define como memoria que persiste a través de hilos y sesiones, y en términos prácticos significa que esos datos necesitan una base de datos, no una ventana de contexto.
Existen dos tipos básicos:
- Preferencias del usuario son señales de personalización que se van acumulando con el tiempo, incluyendo el estilo de escritura, el tono, la profundidad técnica, las elecciones de herramientas y los hábitos de formato. Un asistente de escritura que recuerde que siempre deseas usar markdown y un nivel técnico elevado resulta útil de una manera que ningún sistema sin estado puede igualar, por mejor que sea su modelo.
Muchos equipos cometen los mismos errores. Implementan una memoria a corto plazo llenando la ventana de contexto y una memoria a largo plazo creando un almacén de vectores. Ambos enfoques son incorrectos, o como mínimo incompletos. El primero falla a gran escala por las razones mencionadas anteriormente. El segundo se descompone porque la búsqueda de similitudes es un método deficiente para obtener registros de sesiones, preferencias o reglas de comportamiento, y cada tipo de memoria a largo plazo requiere un backend diferente. El diseño de cuatro capas es una versión más refinada de la división entre memoria a corto y largo plazo, ya que asigna cada tipo de memoria al almacenamiento que necesita.
Cuatro tipos de memoria, una taxonomía
Los equipos que desarrollan sistemas de IA en producción han llegado independientemente a la misma taxonomía. Los nombres varían, pero la estructura no.
El proyecto MemGPT, presentado por Charles Packer y sus colegas en la UC Berkeley en 2023, popularizó la idea de tratar la ventana de contexto de un LLM como RAM, que es limitada pero valiosa. Un sistema operativo amplía la memoria mediante paginación y almacenamiento en disco, y los marcos de trabajo de agentes pueden simular de manera similar la memoria a largo plazo al mover el contexto más antiguo a un almacenamiento de mayor capacidad y recuperarlo cuando sea necesario. Esa idea se convirtió posteriormente en la base de Letta.
El equipo LangMem de LangChain, el marco ReMem de Google DeepMind y la arquitectura de producción de Mem0 describen todas variantes de cuatro tipos:
- Memoria de trabajo es todo lo que está en el contexto en ese momento: los turnos de la conversación actual, los documentos recuperados y las salidas de las herramientas. Es rápida, tiene un límite y es costosa por token.
- Memoria episódica almacena registros de interacciones pasadas, como lo que el agente hizo para un usuario determinado el martes pasado y cuál fue el resultado. Es cronológica y está limitada a la sesión, generalmente se guarda en un almacenamiento rápido de tipo clave-valor o en una base de datos SQL, y se recupera según la recencia o la similitud semántica.
- Memoria semántica conserva hechos generales, conocimiento del dominio y cosas aprendidas a lo largo del tiempo. Es la capa RAG, a la que se accede mediante similitud de embeddings cada vez que una tarea requiere conocimiento del dominio o un hallazgo anterior.
Estos tipos reflejan los principios de la ciencia cognitiva, lo que explica por qué los equipos recurren a ellos. Corresponden a los sistemas de memoria episódica, semántica y procedural de la neurociencia humana, y la memoria de trabajo como ventana de contexto es más que una metáfora: representa el análogo funcional más cercano disponible.
Elegir el almacenamiento: SQL, vector o grafo
¿Qué base de datos debe albergar la memoria? Antes de escribir código, la respuesta honesta es que depende del tipo de memoria en cuestión.
Bases de datos vectoriales (Weaviate, Pinecone, Qdrant, Chroma) son adecuadas para la memoria semántica, ya que permiten búsquedas de similitud en embeddings de manera rápida a gran escala, y la mayoría ahora ofrecen recuperación híbrida que combina BM25 con vectores, superando claramente a la búsqueda puramente vectorial en texto. Carecen de razonamiento temporal y consultas relacionales, y dos hechos contradictorios con embeddings casi idénticos pueden engañar. Mantenga los registros episódicos y las trazas de auditoría estructuradas fuera de ellas.
Bases de datos SQL como PostgreSQL y SQLite son adecuadas para la memoria episódica. Las consultas están estructuradas por ID de usuario, sesión, tipo de tarea o marca de tiempo; el volumen de escritura es bajo, y las garantías ACID son importantes cuando un agente registra lo que hizo y cuándo. PostgreSQL con la extensión pgvector ofrece consultas relacionales y búsqueda vectorial en un mismo sistema. Para menos de decenas de millones de registros, esta es la opción más práctica, con operaciones familiares, hosting económico y SQL para depuración.
Bases de datos gráficas como Neo4j solo encajan en un sistema de memoria cuando los datos episódicos o semánticos tienen una estructura relacional compleja que requiere consultas de múltiples pasos, como determinar quién informó a quién hace tres meses cuando se escribió un documento. Los registros simples de tareas de agentes no cumplen con estos requisitos. Muchos equipos recurren a bases de datos gráficas y descubren que en realidad necesitan SQL.
El patrón que predomina en las implementaciones de producción es el almacenamiento relacional (SQL o PostgreSQL) para los episodios y perfiles de usuarios, un almacén vectorial para búsquedas semánticas, y un estado interno para el contexto de trabajo; PostgreSQL junto con pgvector constituyen la opción híbrida para equipos que desean menor carga operativa. Elija el nivel de almacenamiento según las funciones que debe realizar el tipo de memoria, y no vectorice todo.
Capa 1: memoria de búfer con resumen
La forma más básica es una cola desordenada con un límite de un número fijo de mensajes:
from collections import deque
from typing import List, Dict
class BufferMemory:
def __init__(self, max_turns: int = 10):
# Each turn = one user message + one assistant message
self.history: deque = deque(maxlen=max_turns * 2)
def add_message(self, role: str, content: str):
self.history.append({"role": role, "content": content})
def get_context(self) -> List[Dict]:
return list(self.history)
def token_estimate(self) -> int:
total_chars = sum(len(m["content"]) for m in self.history)
return total_chars // 4 # rough approximation
def clear(self):
self.history.clear()
Esto funciona para intercambios breves. En conversaciones más largas, los mensajes antiguos simplemente desaparecen al final y el agente pierde el hilo de la conversación. Un enfoque mejor, que LangMem denomina método “hot path”, condensa los mensajes antiguos en un resumen antes de descartarlos:
import os
from collections import deque
from typing import List, Dict
from openai import OpenAI
class SummarizingBufferMemory:
def __init__(self, max_turns: int = 8, summary_batch: int = 4):
self.client = OpenAI(
base_url="https://openrouter.ai/api/v1",
api_key=os.environ["OPENROUTER_API_KEY"],
)
self.model = "deepseek/deepseek-v4-flash-0731"
self.recent: deque = deque(maxlen=max_turns * 2)
self.rolling_summary: str = ""
self.summary_batch = summary_batch
def add_message(self, role: str, content: str):
if len(self.recent) == self.recent.maxlen:
self._compress_oldest()
self.recent.append({"role": role, "content": content})
def _compress_oldest(self):
batch = [self.recent.popleft() for _ in range(min(self.summary_batch * 2, len(self.recent)))]
text = "\n".join(f"{m['role']}: {m['content']}" for m in batch)
response = self.client.chat.completions.create(
model=self.model,
max_tokens=300,
messages=[
{
"role": "system",
"content": "You compress conversation history. Reply with the summary only.",
},
{
"role": "user",
"content": (
"Summarize this conversation segment in 2-4 sentences. "
"Preserve any decisions made, constraints stated, and conclusions reached.\n\n"
f"{text}"
),
},
],
)
new_summary = response.choices[0].message.content.strip()
if self.rolling_summary:
self.rolling_summary = f"{self.rolling_summary} | {new_summary}"
else:
self.rolling_summary = new_summary
def get_context(self) -> List[Dict]:
context = []
if self.rolling_summary:
context.append({
"role": "system",
"content": f"[Prior conversation summary: {self.rolling_summary}]",
})
context.extend(list(self.recent))
return context
LangMem separa la resumen del camino activo, que se ejecuta en tiempo real y genera latencia para el usuario, del resumen en segundo plano que se realiza después de la sesión. En las aplicaciones orientadas al usuario, el método en segundo plano es casi siempre mejor: los resúmenes son de mayor calidad porque se puede analizar toda la conversación, y nadie tiene que esperar. En el caso de agentes autónomos que operan sin supervisión, el camino activo es aceptable ya que de todos modos no hay usuario esperando.
Nivel 2: memoria episódica que verifica sus propias citaciones
Aquí es donde muchos equipos fallan: construyen sistemas de recuperación de información pero nunca registran lo que el agente hizo realmente.
El blog técnico de GitHub Copilot describe un patrón que vale la pena replicar. Cada entrada de memoria incluye referencias, es decir, ubicaciones específicas como rutas de archivos, URLs, nombres de herramientas y números de línea que permiten verificar en el momento de la consulta si la información almacenada sigue siendo válida. GitHub denomina esto validación justo a tiempo. Cuando el agente carga una memoria al inicio de una tarea, comprueba si la fuente citada sigue existiendo y sigue respaldando la afirmación. De no ser así, la memoria se corrige o se descarta, de modo que el conjunto se actualiza por sí mismo sin necesidad de una curaduría fuera de línea.
GitHub reportó un efecto medible: el agente de programación con memoria mostró un aumento del 7% en la tasa de fusión de solicitudes de pull request, pasando del 83% al 90%.
import json
import sqlite3
from datetime import datetime, timedelta
from dataclasses import dataclass, field
@dataclass
class Episode:
task_type: str
user_query: str
outcome: str
tools_used: list
citations: list # source references for validation
duration_seconds: float
success: bool
session_id: str = ""
created_at: str = field(default_factory=lambda: datetime.utcnow().isoformat())
class EpisodicMemory:
EXPIRY_DAYS = 28 # GitHub Copilot's production default
def __init__(self, db_path: str = "episodes.db"):
self.conn = sqlite3.connect(db_path, check_same_thread=False)
self._init_schema()
def _init_schema(self):
self.conn.executescript("""
CREATE TABLE IF NOT EXISTS episodes (
id INTEGER PRIMARY KEY AUTOINCREMENT,
task_type TEXT NOT NULL,
user_query TEXT,
outcome TEXT,
tools_used TEXT,
citations TEXT,
duration_s REAL,
success INTEGER,
session_id TEXT,
created_at TEXT
);
CREATE INDEX IF NOT EXISTS idx_task_type ON episodes (task_type, success, created_at);
""")
self.conn.commit()
def record(self, ep: Episode):
self.conn.execute(
"""
INSERT INTO episodes
(task_type, user_query, outcome, tools_used, citations,
duration_s, success, session_id, created_at)
VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?)
""",
(
ep.task_type, ep.user_query, ep.outcome,
json.dumps(ep.tools_used), json.dumps(ep.citations),
ep.duration_seconds, int(ep.success),
ep.session_id, ep.created_at,
),
)
self.conn.commit()
def recall_similar(self, task_type: str, limit: int = 3) -> list[Episode]:
cutoff = (datetime.utcnow() - timedelta(days=self.EXPIRY_DAYS)).isoformat()
cursor = self.conn.execute(
"""
SELECT task_type, user_query, outcome, tools_used, citations,
duration_s, success, session_id, created_at
FROM episodes
WHERE task_type = ? AND success = 1 AND created_at > ?
ORDER BY created_at DESC
LIMIT ?
""",
(task_type, cutoff, limit),
)
return [
Episode(
task_type=r[0], user_query=r[1], outcome=r[2],
tools_used=json.loads(r[3]), citations=json.loads(r[4]),
duration_seconds=r[5], success=bool(r[6]),
session_id=r[7], created_at=r[8],
)
for r in cursor.fetchall()
]
def validate_citations(self, episode: Episode, validator_fn) -> bool:
"""
validator_fn(citation: str) -> bool
Check if cited sources are still valid (file exists, URL responds, etc.)
Return False if any citation fails; the episode should be discarded.
"""
return all(validator_fn(c) for c in episode.citations)
El vencimiento de 28 días sigue la decisión adoptada en entornos de producción por GitHub. Los episodios antiguos no se eliminan, pero dejan de ser recuperables; además, una consulta LIMIT sobre un índice filtrado por fecha es rápida y no requiere tareas de curación en segundo plano.
Nivel 3: memoria semántica mediante la búsqueda híbrida de Weaviate
Este es el nivel RAG. La mayoría de las guías tratan sobre la configuración y el cese del funcionamiento, mientras que los verdaderos problemas en entornos de producción se relacionan con la calidad de la recuperación de información y su origen.
Un error común al principio es configurar la memoria semántica como solo de lectura. Un agente que opera durante mucho tiempo necesita que sea de lectura-escritura: cuando el agente encuentra un hecho durante una tarea, lo valida y decide que merece ser conservado, debe escribirlo en la memoria semántica para sesiones futuras. El segundo error es confiar únicamente en la similitud coseno. La búsqueda híbrida de Weaviate combina la coincidencia por palabras clave BM25 con la similitud vectorial, y suele superar el rendimiento de cualquiera de los métodos por separado en la recuperación de información para los agentes.
Tercero, y esto es lo que abre la puerta a los envenenamientos de datos, las entradas a menudo no tienen origen comprobable. Cada una necesita una fuente, un puntaje de confianza e un identificador de sesión. Al carecer de estos elementos, resulta imposible realizar búsquedas ponderadas por confianza; las entradas obsoletas o dudosas no pueden ser degradadas en su prioridad, y rastrear una búsqueda incorrecta se convierte en una suposición.
import uuid as uuid_lib
from datetime import datetime
import weaviate
from weaviate.classes.config import Configure, Property, DataType, VectorDistances
from weaviate.classes.query import MetadataQuery, HybridFusion, Filter
def embed(text: str) -> list[float]:
"""
Swap in any embedding source: OpenAI, Cohere, sentence-transformers, etc.
Returns a normalized float vector.
"""
from sentence_transformers import SentenceTransformer
_model = SentenceTransformer("all-MiniLM-L6-v2")
return _model.encode(text, normalize_embeddings=True).tolist()
class WeaviateSemanticMemory:
COLLECTION = "AgentMemory"
def __init__(self, host: str = "localhost", port: int = 8080):
self.client = weaviate.connect_to_local(host=host, port=port)
self._ensure_collection()
def _ensure_collection(self):
if self.client.collections.exists(self.COLLECTION):
return
self.client.collections.create(
name=self.COLLECTION,
# We provide our own vectors; no built-in vectorizer needed.
# Swap to Configure.Vectorizer.text2vec_openai() if you prefer managed embedding.
vectorizer_config=Configure.Vectorizer.none(),
vector_index_config=Configure.VectorIndex.hnsw(
distance_metric=VectorDistances.COSINE
),
properties=[
Property(name="content", data_type=DataType.TEXT),
Property(name="task_type", data_type=DataType.TEXT),
Property(name="source", data_type=DataType.TEXT),
Property(name="citations", data_type=DataType.TEXT),
Property(name="session_id", data_type=DataType.TEXT),
Property(name="confidence", data_type=DataType.NUMBER),
Property(name="created_at", data_type=DataType.TEXT),
],
)
def store(
self,
content: str,
task_type: str = "",
source: str = "agent",
citations: str = "",
session_id: str = "",
confidence: float = 1.0,
) -> str:
collection = self.client.collections.get(self.COLLECTION)
doc_id = str(uuid_lib.uuid4())
collection.data.insert(
properties={
"content": content,
"task_type": task_type,
"source": source,
"citations": citations,
"session_id": session_id,
"confidence": confidence,
"created_at": datetime.utcnow().isoformat(),
},
vector=embed(content),
uuid=doc_id,
)
return doc_id
def retrieve_hybrid(
self,
query: str,
n_results: int = 5,
min_confidence: float = 0.6,
task_type: str = None,
) -> list[dict]:
collection = self.client.collections.get(self.COLLECTION)
# Filter by confidence floor and optionally by task type
confidence_filter = Filter.by_property("confidence").greater_or_equal(min_confidence)
if task_type:
active_filter = (
Filter.by_property("task_type").equal(task_type) & confidence_filter
)
else:
active_filter = confidence_filter
results = collection.query.hybrid(
query=query,
vector=embed(query),
limit=n_results,
fusion_type=HybridFusion.RELATIVE_SCORE,
filters=active_filter,
return_metadata=MetadataQuery(score=True),
)
return [
{
"content": obj.properties["content"],
"score": obj.metadata.score,
"confidence": obj.properties.get("confidence", 1.0),
"source": obj.properties.get("source", ""),
"citations": obj.properties.get("citations", ""),
"uuid": str(obj.uuid),
}
for obj in results.objects
]
def update_confidence(self, doc_id: str, new_confidence: float):
collection = self.client.collections.get(self.COLLECTION)
collection.data.update(
uuid=doc_id,
properties={"confidence": new_confidence},
)
def close(self):
self.client.close()
Dos detalles en retrieve_hybrid faltan en muchas implementaciones. Uno es que el endpoint híbrido de Weaviate recibe tanto el texto de la consulta en bruto (utilizado para BM25) como su representación vectorial (usada para búsquedas vectoriales). Los resultados basados en palabras clave exactas provienen del primero, mientras que la cercanía semántica proviene del segundo, y su combinación ofrece un rendimiento superior al de cualquiera de ellos por separado. El otro detalle es el filtro min_confidence, que excluye de los resultados las memorias de baja fiabilidad, independientemente de cuán cercanas sean semánticamente. En efecto, se trata de un control de seguridad y no solo de un ajuste de relevancia.
El método update_confidence proporciona la vía de retroalimentación. Si el razonamiento propio del agente contradice una memoria recuperada durante una tarea, se debe reducir su nivel de confianza. Si una verificación de citas lo confirma, se debe aumentar dicha confianza. El conjunto de memorias debería aprender a partir del uso real.
Capa 4: memoria procedural y evolución de los prompts
Esta capa es la que más se pasa por alto. El diseño inicial consiste en un archivo JSON con reglas que el agente lee, sigue y amplía tras los fallos. Funciona, pero llega a un límite, ya que las reglas en un archivo separado compiten por el espacio de contexto y requieren una recuperación explícita.
Los ingenieros de LangMem en LangChain publicaron una idea valiosa que merece ser adoptada: en producción, la memoria procedural existe como el propio prompt del sistema. No se almacena junto al prompt ni se recupera de él; es el prompt en sí. Cada vez que un fallo enseña al agente una nueva regla de conducta, el prompt se reescribe, práctica que denominan optimización del prompt como memoria procedural. Los comportamientos aprendidos se convierten en instrucciones en lenguaje natural que el modelo lee al inicio de cada interacción.
La versión más sencilla con archivo JSON es donde la mayoría de los equipos deberían comenzar:
import json
import os
import re
from datetime import datetime, timedelta, timezone
from openai import OpenAI
MODEL = "deepseek/deepseek-v4-flash-0731"
client = OpenAI(
base_url="https://openrouter.ai/api/v1",
api_key=os.environ["OPENROUTER_API_KEY"],
)
class ProceduralMemory:
def __init__(self, rules_path: str = "procedural_rules.json"):
self.rules_path = rules_path
self.rules: list[dict] = self._load()
def _load(self) -> list[dict]:
try:
with open(self.rules_path) as f:
return json.load(f)
except FileNotFoundError:
return []
def add_rule(self, situation: str, action: str, reason: str, confidence: float = 1.0):
self.rules.append({
"situation": situation,
"action": action,
"reason": reason,
"confidence": float(confidence),
"added_at": datetime.now(timezone.utc).isoformat(),
"trigger_count": 0,
})
self._save()
def get_applicable_rules(self, context: str, min_confidence: float = 0.7) -> list[dict]:
relevant = []
ctx_lower = context.lower()
dirty = False
for rule in self.rules:
if rule.get("confidence", 1.0) < min_confidence:
continue
keywords = rule["situation"].lower().split()
if not keywords:
continue
hits = sum(1 for kw in keywords if kw in ctx_lower)
if hits >= max(1, len(keywords) // 3):
rule["trigger_count"] = rule.get("trigger_count", 0) + 1
relevant.append(rule)
dirty = True
if dirty:
self._save()
relevant.sort(key=lambda r: r.get("confidence", 1.0), reverse=True)
return relevant[:5]
def prune_stale(self, max_age_days: int = 60, min_triggers: int = 2):
cutoff = (datetime.now(timezone.utc) - timedelta(days=max_age_days)).isoformat()
self.rules = [
r for r in self.rules
if r.get("added_at", "") > cutoff or r.get("trigger_count", 0) >= min_triggers
]
self._save()
def _save(self):
tmp = f"{self.rules_path}.tmp"
with open(tmp, "w") as f:
json.dump(self.rules, f, indent=2)
os.replace(tmp, self.rules_path)
def _parse_json(text: str) -> dict:
text = text.strip()
fenced = re.search(r"```(?:json)?\s*(.*?)```", text, re.S)
if fenced:
text = fenced.group(1).strip()
return json.loads(text)
def extract_rule_from_failure(failure_trace: str, memory: ProceduralMemory):
response = client.chat.completions.create(
model=MODEL,
max_tokens=250,
response_format={"type": "json_object"},
messages=[
{
"role": "system",
"content": "You return only a JSON object. No prose, no code fences.",
},
{
"role": "user",
"content": (
"A task failed. Extract one behavioral rule to prevent this failure.\n"
"Return ONLY valid JSON with keys: situation, action, reason, "
"confidence (0.0-1.0)\n"
f"Failure trace:\n{failure_trace}"
),
},
],
)
try:
rule = _parse_json(response.choices[0].message.content)
except (json.JSONDecodeError, AttributeError, TypeError):
return None
if not all(k in rule for k in ("situation", "action", "reason")):
return None
memory.add_rule(
situation=str(rule["situation"]),
action=str(rule["action"]),
reason=str(rule["reason"]),
confidence=float(rule.get("confidence", 1.0)),
)
return rule
prune_stale aborda el problema del olvido a nivel procedural. Una regla creada tras un fallo inusual y que nunca se activa nuevamente debería desaparecer, mientras que una regla que se ejecuta con frecuencia permanece sin importar cuán antigua sea. Es una herramienta rudimentaria, pero el olvido es lo más sencillo de implementar precisamente en esta capa.
Una vez que un agente funciona de forma continua, el siguiente paso lógico es incorporar en la propia instrucción del sistema aquellas reglas que son fiables y se utilizan con frecuencia. LangMem automatiza esto mediante un bucle de optimización de instrucciones, y la selección manual es perfectamente adecuada para configuraciones más pequeñas.
Vinculando las capas entre sí
La siguiente lista muestra cómo se integran las cuatro capas en entornos de producción. Hay una restricción innegociable: escribir en la memoria nunca debe retrasar la respuesta.
import json
from concurrent.futures import ThreadPoolExecutor
from dataclasses import dataclass, field
@dataclass
class TaskResult:
task_type: str
query: str
outcome: str
tools_used: list
citations: list
duration_seconds: float
success: bool
new_facts: list[dict] = field(default_factory=list)
session_id: str = ""
class MemoryManager:
def __init__(
self,
weaviate_host: str = "localhost",
episodic_db: str = "episodes.db",
rules_path: str = "procedural_rules.json",
):
self.buffer = SummarizingBufferMemory(max_turns=8)
self.episodic = EpisodicMemory(db_path=episodic_db)
self.semantic = WeaviateSemanticMemory(host=weaviate_host)
self.procedural = ProceduralMemory(rules_path=rules_path)
self._pool = ThreadPoolExecutor(max_workers=2, thread_name_prefix="memory_write")
def build_context(self, query: str, task_type: str) -> list[dict]:
context: list[dict] = []
# Procedural rules first: they constrain behavior throughout the task
rules = self.procedural.get_applicable_rules(query)
if rules:
rules_text = "\n".join(
f"- When '{r['situation']}': {r['action']} (reason: {r['reason']})"
for r in rules
)
context.append({"role": "user", "content": f"[Behavioral rules:\n{rules_text}]"})
# Semantic facts: domain knowledge and past discoveries
facts = self.semantic.retrieve_hybrid(query, n_results=5, min_confidence=0.6)
if facts:
facts_text = "\n".join(f"- {f['content']}" for f in facts)
context.append({"role": "user", "content": f"[Relevant knowledge:\n{facts_text}]"})
# Past episodes: outcome templates for similar tasks
episodes = self.episodic.recall_similar(task_type, limit=3)
if episodes:
ep_text = "\n".join(
f"- Outcome: {e.outcome} (tools: {', '.join(e.tools_used)})"
for e in episodes
)
context.append({"role": "user", "content": f"[Past similar tasks:\n{ep_text}]"})
# Current conversation last: the model reads this most carefully
context.extend(self.buffer.get_context())
return context
def record_turn(self, role: str, content: str):
self.buffer.add_message(role, content)
def persist(self, result: TaskResult):
# Submit to thread pool and return immediately; never block the caller
self._pool.submit(self._persist_worker, result)
def _persist_worker(self, result: TaskResult):
ep = Episode(
task_type=result.task_type,
user_query=result.query,
outcome=result.outcome,
tools_used=result.tools_used,
citations=result.citations,
duration_seconds=result.duration_seconds,
success=result.success,
session_id=result.session_id,
)
self.episodic.record(ep)
for fact in result.new_facts:
self.semantic.store(**fact)
if not result.success:
extract_rule_from_failure(result.outcome, self.procedural)
def shutdown(self):
self._pool.shutdown(wait=True)
self.semantic.close()
El orden del contexto en build_context es intencional. La investigación sobre el sesgo de atención posicional respalda la idea de que los modelos otorgan más importancia al inicio y al final de la ventana. Las reglas de comportamiento se colocan en la parte superior, ya que son restricciones que deben dar forma a todo lo que viene después. La conversación actual se encuentra en la parte inferior porque es lo que el modelo necesita más para responder a continuación. Los hechos semánticos y los episodios anteriores se sitúan en medio, ofreciendo referencias sin saturar la tarea activa.
La línea self._pool.submit(self._persist_worker, result) es la elección de ingeniería más importante en la clase. Cada llamada de escritura devuelve el resultado de inmediato, por lo que la respuesta del LLM ya se está transmitiendo al usuario antes de que comience cualquier escritura en la base de datos.
Tres decisiones que dan forma a la arquitectura
Con cuatro capas implementadas, tres decisiones arquitectónicas influyen en todo lo demás, y muchos equipos cometen al menos un error al respecto.
Escribimientos en el camino principal o en segundo plano. El equipo de LangMem invirtió mucho esfuerzo en esto. Los escritimientos en el camino principal ocurren durante la conversación, cuando el agente detecta un hecho que merece ser almacenado y realiza una escritura directamente. Los escritimientos en segundo plano tienen lugar después de que finaliza la sesión, donde se analiza toda la conversación para determinar qué información merece conservarse. El camino principal genera latencia, ya que cada llamada a semantic.store() dentro del bucle de respuesta representa un commit en la base de datos que el usuario debe esperar. Los escritimientos en segundo plano mejoran la calidad de los resúmenes, ya que permiten examinar los temas y decisiones a lo largo de toda la conversación y almacenar resúmenes más detallados. Para aplicaciones en producción orientadas al usuario, los escritimientos en segundo plano son casi siempre la opción correcta. El pool de hilos en persist() permite obtener la mayor parte de los beneficios evitando al mismo tiempo un proceso completamente asíncrono.
Recuperación o resumen. Estos métodos abordan problemas diferentes, y los equipos a menudo los confunden. La recuperación consiste en encontrar el pasaje relevante dentro de un corpus grande. La consulta es específica y el corpus es amplio y bastante estable, como la documentación, los catálogos de productos o el código. Falla cuando la consulta es compleja, la información relevante está dispersa en muchos pasajes o el corpus cambia más rápido que el índice de embeddings. El resumen implica comprimir una conversación o un documento para adaptarlo al contexto. Funciona cuando la entrada es secuencial, como en el historial de conversaciones, y se puede tolerar cierta pérdida de información; falla cuando los detalles descartados resultan ser importantes. Durante 2025 la industria comenzó a reemplazar ambos términos por “ingeniería de contexto”: ya no se trata de recuperación frente a resumen, sino de determinar qué combinación de conocimiento almacenado, pasajes recuperados e historial comprimido ofrece el mejor resultado.
El contexto para esta llamada de inferencia en particular. El métodobuild_context del gestor de memoria es una forma de ingeniería de contexto, ya que decide cuál de las cuatro memorias leer, cuánto tomar de cada una y el orden para combinar los datos. Procedencia desde el primer día. No se puede añadir posteriormente la información de procedencia a un sistema de memoria. Desde la primera escritura, cada entrada debe contener su fuente, su puntuación de confianza, el ID de sesión y la hora de creación. Sin ellos no es posible realizar búsquedas ponderadas por confianza, filtrar según la reciente fecha ni auditar búsquedas defectuosas. Los equipos que omiten la información de procedencia en el primer sprint terminan pasando semanas añadiéndola a una base de datos que ya contiene miles de entradas sin ella.
Envenenamiento de memoria: patrones de ataque y defensas
Esta es la sección que los guías de diseño de memoria suelen omitir, pero debería estar en todos ellos.
Un artículo de arXiv de 2025 presentó el marco de ataque MINJA, una de las análisis públicos más exhaustivos sobre el envenenamiento de memoria en agentes LLM. Sus autores indican que, a través de interacciones normales de los usuarios sin privilegios elevados, los atacantes pueden lograr tasas de éxito en la inyección superiores al 95% utilizando un patrón de tres pasos. Un paso de conexión vincula un tema inocente y legítimo con el objetivo al que apunta el atacante. Un paso de indicación proporciona pistas sutiles que hacen que la memoria dañina parezca fiable. Un paso de acortamiento progresivo comprime la inyección hasta dejar una huella mínima que sobrevive a la extracción de memoria.
Lo que diferencia al envenenamiento de memoria de la inyección de prompts ordinaria es que las memorias envenenadas trascienden los límites de sesión y se acumulan. Una inyección exitosa en la primera sesión puede influir en todas las sesiones posteriores, de forma indefinida.
Un artículo separado de 2025 sobre defensas bayesianas contra la desconfianza describió un diseño de cinco capas que abarca toda la superficie de ataque:
- Moderación de las entradas en el momento de la escritura. Antes de cada escritura en memoria, se calcula una puntuación de confianza combinada a partir de varios indicadores independientes: el origen de la información, el análisis semántico de los patrones de instrucciones y la detección de anomalías basada en el historial del usuario. Ningún indicador por sí solo es suficiente, y las inyecciones MINJA están diseñadas para superar las verificaciones basadas en un solo indicador.
- Eliminación de instrucciones antes de la escritura. Si una memoria candidata parece ser una orden (“siempre”, “nunca”, “debes”, “no”), se debe limpiar o descartar en lugar de guardarla. Las memorias factuales reales nunca necesitan dar órdenes al agente.
min_confidence en el lado de Weaviate de la Capa 3. Los recuerdos con baja procedencia quedan fuera de contexto independientemente de su similitud. En una implementación más completa, el campo de confianza se actualiza a partir de los resultados de validación, de modo que los recuerdos verificados ganan confianza y aquellos contradictorios la pierden.Para la mayoría de los equipos, el orden lógico es primero aplicar etiquetas de procedencia en cada escritura, luego detectar instrucciones en el momento de la escritura y finalmente realizar búsquedas ponderadas según el nivel de confianza. La supervisión del comportamiento viene al final porque necesita una línea base de comportamientos, y crearla requiere semanas de datos de operación.
Otro punto: investigaciones realizadas con sistemas de memoria del mundo real que contaban con entradas legítimas preexistentes demostraron que las condiciones reales reducen considerablemente la efectividad de los ataques en comparación con los entornos de laboratorio, ya que las entradas contaminadas compiten con las memorias legítimas por los espacios de recuperación. Eso no es motivo para omitir las medidas de defensa, sino una razón más para mantener las entradas legítimas de alta calidad y con alto nivel de confianza, ya que constituyen la primera línea de defensa.
Qué construir primero
No es necesario utilizar inmediatamente las cuatro capas. Para la mayoría de las aplicaciones de IA, dos de ellas cubren alrededor del 80% de los fallos de memoria: la Capa 1 (búfer más resumen) y la Capa 3 (Weaviate con búsqueda híbrida, filtro de confianza y proveniencia). Juntas manejan los dos problemas más frecuentes: un agente que olvida parte del proceso de una tarea y uno que carece de conocimiento en un dominio específico. Construya ambas, compare el grado de completación de las tareas antes y después, y solo entonces determine qué falta.
Agregue memoria episódica cuando se dé cuenta de que se pregunta por qué el agente olvidó lo que logró en la sesión anterior, ya que esa pregunta es la señal. No la agregue antes. Agregue memoria procedural cuando la misma clase de error se repita en distintas tareas y la ingeniería de prompts no lo resuelva, porque un patrón de comportamiento requiere memoria de tipo conductual.
PostgreSQL con pgvector es la opción por defecto más adecuada para la mayoría de los equipos. Una instancia de base de datos con CREATE EXTENSION IF NOT EXISTS vector ofrece cumplimiento de ACID, SQL flexible, búsqueda de similitud vectorial y puntuación híbrida; según algunas estimaciones, el costo de alojamiento es de aproximadamente 45 dólares al mes para 10K de memorias. Una base de datos vectorial independiente solo resulta útil cuando la memoria semántica crece hasta el punto en que el índice HNSW de PostgreSQL se convierte en el factor que ralentiza el sistema, algo que la mayoría de los equipos nunca experimentarán.
Haga que cada escritura en la memoria sea asíncrona. La respuesta al usuario no debería depender de un commit en la base de datos, así que delegue esa tarea en un pool de hilos y continúe con otras actividades.
Evite los almacenes vectoriales en memoria en entornos de producción. Una implementación sin almacenamiento persistente pierde todo el conocimiento acumulado al reiniciar el servidor, un error que pocas personas cometen dos veces.
Leer más
- Memory Semántica de Agentes en LangGraph con Qdrant y Recuperación entre Hilos — Proporcionar a un agente de LangGraph una memoria a largo plazo con Qdrant: incrustar datos del usuario, recuperarlos mediante IDs de hilo y agregar filtros de carga útil para la isolación multiinquilino.
- Un Agente de Programación Local en Python: Herramienta Ollama que Escribe y Ejecuta Código — Crear un pequeño bucle de agente con Ollama y llama3.1:8b que escribe archivos en Python, los ejecuta y corrige errores, sin necesidad de claves API ni costos en la nube.