Un agente de IA sin memoria es como hablar con alguien que cada vez que entras por la puerta no sabe quién eres. Da igual lo que le hayas contado ayer, lo que haya decidido o lo que hayas acordado con él: empieza desde cero. Para muchos casos de uso esto es un problema serio, y la solución no pasa por meter más mensajes en el contexto, sino por implementar memoria episódica.
Tipos de memoria en un agente
Antes de entrar en código conviene aclarar de qué hablamos cuando hablamos de memoria en agentes. Hay tres tipos principales:
- Semántica: conocimiento general del mundo, lo que sabe el modelo por entrenamiento. "Python es un lenguaje interpretado" es memoria semántica.
- Episódica: eventos concretos. "El usuario me dijo el martes que prefiere respuestas cortas y que trabaja con FastAPI." Eso es un episodio.
- Procedimental: saber cómo hacer algo. Los propios pesos del modelo codifican buena parte de esto.
Este artículo se centra en la episódica: qué pasó, cuándo ocurrió y qué decidió el agente en sesiones anteriores.
Memoria episódica frente a historial de chat
Mucha gente confunde las dos cosas. El historial de chat son los últimos N mensajes de la conversación actual, los que metes en el array messages al llamar al LLM. Es efímero: cuando cierras la sesión, desaparece.
La memoria episódica es otra cosa. Son resúmenes de sesiones pasadas, preferencias detectadas, hechos relevantes del usuario o patrones de comportamiento que vale la pena conservar entre conversaciones. No la guardas en el historial de chat, sino en una base de datos externa, y la recuperas al inicio de cada sesión nueva.
Dicho de otro modo: el historial de chat es lo que está pasando ahora mismo; la memoria episódica es lo que pasó antes.
Implementar memoria episódica con LanceDB
LanceDB es una base de datos vectorial embebida, sin servidor, que funciona de forma parecida a SQLite pero para vectores. No necesitas levantar ningún proceso adicional; funciona directamente desde Python leyendo y escribiendo archivos locales.
Instala las dependencias:
pip install lancedb openai sentence-transformersPrimero defines el esquema de tus episodios:
import lancedb
import pyarrow as pa
from datetime import datetime
db = lancedb.connect("./memoria_agente")
schema = pa.schema([
pa.field("vector", pa.list_(pa.float32(), 384)),
pa.field("text", pa.string()),
pa.field("session_id", pa.string()),
pa.field("timestamp", pa.string()),
pa.field("user_id", pa.string()),
])
table = db.create_table("episodios", schema=schema, exist_ok=True)Aquí vector almacena el embedding del texto del episodio, con 384 dimensiones si usas all-MiniLM-L6-v2. Ajusta ese número si cambias de modelo de embeddings.
Guardar un episodio
Al acabar cada sesión, resumes la conversación con el LLM y guardas ese resumen como episodio:
from sentence_transformers import SentenceTransformer
encoder = SentenceTransformer("all-MiniLM-L6-v2")
def guardar_episodio(texto: str, session_id: str, user_id: str):
vector = encoder.encode(texto).tolist()
table.add([{
"vector": vector,
"text": texto,
"session_id": session_id,
"timestamp": datetime.utcnow().isoformat(),
"user_id": user_id,
}])Recuperar episodios relevantes
Cuando el usuario empieza una nueva sesión, generates un embedding de su primera pregunta y buscas los episodios más similares:
def recuperar_episodios(query: str, user_id: str, k: int = 5) -> list[str]:
vector_query = encoder.encode(query).tolist()
resultados = (
table.search(vector_query)
.where(f"user_id = '{user_id}'")
.limit(k)
.to_list()
)
return [r["text"] for r in resultados]La búsqueda por similitud semántica te permite recuperar episodios relevantes aunque el usuario no use las mismas palabras que en la sesión anterior. No es búsqueda exacta; es búsqueda por significado.
Cuándo guardar un episodio
Aquí la mayoría de implementaciones fallan: guardan demasiado o demasiado poco. Estas son las tres situaciones donde tiene sentido persistir un episodio:
- Al final de cada sesión: pide al LLM que resuma en 2-3 frases qué se trató, qué decidió el usuario y qué información relevante apareció.
- Cuando el usuario comparte datos personales o preferencias: nombre, stack tecnológico, tipo de proyectos, restricciones importantes.
- Cuando se toma una decisión significativa: "decidimos usar Redis para las colas", "descartamos FastAPI porque ya tienen Django".
Para detectar automáticamente qué vale la pena guardar, puedes pedírselo al propio LLM:
def extraer_hechos_relevantes(conversacion: str, cliente_llm) -> str:
prompt = f"""
Analiza esta conversación y extrae, en 3-5 frases, los hechos que un agente
debería recordar en futuras sesiones con este usuario.
Incluye: preferencias, decisiones tomadas, contexto técnico relevante.
Omite detalles triviales o pasajeros.
Conversación:
{conversacion}
"""
respuesta = cliente_llm.chat.completions.create(
model="gpt-4o-mini",
messages=[{"role": "user", "content": prompt}]
)
return respuesta.choices[0].message.contentIntegrar con PostgreSQL y pgvector
Si ya tienes PostgreSQL en tu infraestructura, no necesitas añadir LanceDB. La extensión pgvector añade soporte para vectores directamente en Postgres y funciona bien para volúmenes moderados (hasta varios millones de vectores).
Instala la extensión y crea la tabla:
-- En PostgreSQL
CREATE EXTENSION IF NOT EXISTS vector;
CREATE TABLE episodios (
id SERIAL PRIMARY KEY,
user_id TEXT NOT NULL,
session_id TEXT NOT NULL,
content TEXT NOT NULL,
embedding vector(384),
created_at TIMESTAMPTZ DEFAULT NOW()
);
CREATE INDEX ON episodios USING ivfflat (embedding vector_cosine_ops)
WITH (lists = 100);El índice ivfflat acelera las búsquedas aproximadas por similitud. Para tablas pequeñas puedes omitirlo; para tablas grandes es casi obligatorio.
Guardar y recuperar desde Python:
import psycopg2
import json
conn = psycopg2.connect("postgresql://usuario:contraseña@localhost/agente_db")
def guardar_episodio_pg(texto: str, session_id: str, user_id: str):
vector = encoder.encode(texto).tolist()
with conn.cursor() as cur:
cur.execute(
"INSERT INTO episodios (user_id, session_id, content, embedding) VALUES (%s, %s, %s, %s)",
(user_id, session_id, texto, json.dumps(vector))
)
conn.commit()
def recuperar_episodios_pg(query: str, user_id: str, k: int = 5) -> list[str]:
vector_query = json.dumps(encoder.encode(query).tolist())
with conn.cursor() as cur:
cur.execute("""
SELECT content FROM episodios
WHERE user_id = %s
ORDER BY embedding <=> %s
LIMIT %s
""", (user_id, vector_query, k))
return [fila[0] for fila in cur.fetchall()]El operador <=> es distancia coseno. También tienes <-> para distancia euclidiana y <#> para producto interior. Para texto, coseno suele dar mejores resultados.
Inyectar la memoria en el system prompt
Recuperar episodios es solo la mitad del trabajo. Tienes que inyectarlos en el contexto antes de que el LLM responda:
def construir_system_prompt(query: str, user_id: str) -> str:
episodios = recuperar_episodios(query, user_id, k=5)
if not episodios:
contexto = ""
else:
contexto = "nnLo que sabes de este usuario (sesiones anteriores):n"
for i, ep in enumerate(episodios, 1):
contexto += f"- {ep}n"
return f"""Eres un asistente técnico especializado en Python y desarrollo backend.
Responde de forma directa y práctica.{contexto}"""Un par de puntos importantes aquí. Primero, no incluyas episodios irrelevantes: si el usuario pregunta por Docker y los episodios hablan de CSS, mejor no meterlos. El filtro por similitud semántica ya ayuda, pero puedes añadir un umbral de puntuación mínima para descartar los que tienen poca relevancia. Segundo, ten cuidado con la privacidad. Guarda solo lo que el usuario esperaría que guardes, define un periodo de retención y ofrece algún mecanismo para que el usuario pueda borrar su memoria.
Clase EpisodicMemory completa
Para que sea más fácil de usar en un proyecto real, aquí tienes una clase que encapsula todo lo anterior con LanceDB:
import lancedb
import pyarrow as pa
from sentence_transformers import SentenceTransformer
from datetime import datetime
from typing import Optional
class EpisodicMemory:
def __init__(self, db_path: str = "./memoria", model: str = "all-MiniLM-L6-v2"):
self.db = lancedb.connect(db_path)
self.encoder = SentenceTransformer(model)
self._init_table()
def _init_table(self):
schema = pa.schema([
pa.field("vector", pa.list_(pa.float32(), 384)),
pa.field("text", pa.string()),
pa.field("session_id", pa.string()),
pa.field("user_id", pa.string()),
pa.field("timestamp", pa.string()),
])
self.table = self.db.create_table("episodios", schema=schema, exist_ok=True)
def save(self, text: str, session_id: str, user_id: str = "default"):
"""Guarda un episodio nuevo."""
vector = self.encoder.encode(text).tolist()
self.table.add([{
"vector": vector,
"text": text,
"session_id": session_id,
"user_id": user_id,
"timestamp": datetime.utcnow().isoformat(),
}])
def recall(self, query: str, user_id: str = "default", k: int = 5,
min_score: Optional[float] = None) -> list[str]:
"""Recupera los k episodios más relevantes para la query."""
vector = self.encoder.encode(query).tolist()
resultados = (
self.table.search(vector)
.where(f"user_id = '{user_id}'")
.limit(k)
.to_list()
)
textos = []
for r in resultados:
if min_score is not None and r.get("_distance", 1.0) > min_score:
continue
textos.append(r["text"])
return textos
def get_context(self, query: str, user_id: str = "default", k: int = 5) -> str:
"""Devuelve el contexto formateado para inyectar en el system prompt."""
episodios = self.recall(query, user_id, k)
if not episodios:
return ""
lineas = "n".join(f"- {ep}" for ep in episodios)
return f"nContexto de sesiones anteriores:n{lineas}"Usarlo en un agente es sencillo:
from openai import OpenAI
memoria = EpisodicMemory()
cliente = OpenAI()
def responder(query: str, user_id: str, session_id: str) -> str:
contexto = memoria.get_context(query, user_id)
system = f"Eres un asistente técnico en Python.{contexto}"
respuesta = cliente.chat.completions.create(
model="gpt-4o-mini",
messages=[
{"role": "system", "content": system},
{"role": "user", "content": query},
]
)
return respuesta.choices[0].message.content
# Al cerrar la sesión, guardas el resumen
def cerrar_sesion(resumen: str, user_id: str, session_id: str):
memoria.save(resumen, session_id=session_id, user_id=user_id)LanceDB o pgvector: cuál elegir
Si estás arrancando un proyecto nuevo sin infraestructura previa, LanceDB es la opción más rápida: sin configuración, sin servidor, funciona en local y en producción. Si ya tienes Postgres y no quieres añadir otra tecnología, pgvector es la elección obvia. El rendimiento de ambos es más que suficiente para la mayoría de agentes: empieza a notar la diferencia a partir de varios millones de vectores.
Para implementaciones más avanzadas puedes representar la memoria como un grafo: episodios como nodos, relaciones entre sesiones como aristas. NetworkX te permite visualizar eso durante el desarrollo y detectar si el agente está acumulando información duplicada o contradictoria. Pero para empezar, una tabla vectorial simple funciona muy bien.
Si quieres explorar más sobre Python para machine learning e IA o necesitas construir interfaces para aplicaciones de IA con Python, tienes más artículos en esta misma web.
Imagen: Pexels / Google DeepMind