Módulo 8: Memoria y Persistencia
Proyecto Evolutivo: Persistencia y Memoria (v3)
Descripción del proyecto
En el Módulo 7, construiste la v2 del AI Research Assistant: un agente robusto con retry logic, búsqueda paralela protegida, merge con deduplicación, y graceful degradation. Si una API falla, reintenta. Si todas fallan, degrada gracefully. Es production-ready en resiliencia.
Pero tiene un problema fundamental: es amnésico. Cada ejecución empieza de cero. Si investigaste "RAG techniques" ayer y hoy preguntas por "retrieval augmented generation", el agente no conecta los puntos. Si el proceso se interrumpe a mitad de una investigación, pierdes todo. Si un segundo usuario usa el sistema, no hay aislamiento.
La v3 agrega tres capas encima de la v2: checkpointing con MemorySaver para que el agente guarde progreso en cada paso y pueda resumir ante interrupciones, long-term memory con InMemoryStore para que recuerde preferencias e historial entre sesiones, y soporte multi-usuario para que cada usuario tenga contexto aislado.
El momento que define este proyecto: ejecutas el agente, investigas un tema, cierras la terminal. Abres una nueva terminal, ejecutas el agente con el mismo usuario. El agente dice: "¡Bienvenido de vuelta! La última vez investigaste RAG techniques. ¿Quieres continuar o empezar algo nuevo?" Ese momento — el agente que recuerda — es lo que separa un prototipo de un producto.
Objetivo del proyecto
Evolucionar el AI Research Assistant de v2 (robusto pero amnésico) a v3 (robusto con memoria), agregando checkpointing, long-term memory, y soporte multi-usuario.
Al completar este proyecto:
- 🔧 Implementarás checkpointing con MemorySaver para guardar progreso en cada paso
- 🔧 Agregarás InMemoryStore para preferencias e historial entre sesiones
- 🔧 Crearás un saludo personalizado que usa la memoria del usuario
- 🔧 Implementarás detección y reanudación de sesiones interrumpidas
- 🔧 Soportarás múltiples usuarios con aislamiento completo
- 🔧 Persistirás la memoria a JSON para sobrevivir reinicios
Antes y después
v2 (Módulo 7): robusto pero amnésico
Sesión 1: "Investiga RAG" → Reporte → FIN
Sesión 2: "Investiga RAG" → Empieza de cero. No sabe que ya investigaste.
Crash a mitad → Todo perdido.
Usuario A y B → Mismo contexto. Sin aislamiento.
v3 (Este módulo): robusto con memoria
Sesión 1: "Investiga RAG" → Reporte → Guarda en historial
Sesión 2: "Investiga RAG" → "Ya investigaste esto. ¿Continuar o nuevo enfoque?"
Crash a mitad → Resume desde el último checkpoint.
Usuario A → Su contexto aislado. Usuario B → El suyo.
Especificaciones técnicas
| Componente | Versión | Propósito |
|---|---|---|
| Python | 3.11+ | Runtime |
| LangChain | v1.2+ | Framework de LLMs |
| LangGraph | v1.0+ | Functional API + Store + Checkpointer |
| langchain-openai | latest | Proveedor de modelos |
| pydantic | v2+ | Modelos structured |
Estructura del proyecto
research-assistant/
├── .env
├── requirements.txt
├── agents/
│ └── researcher.py # MODIFICADO — v3 con store + greeting
├── tools/
│ ├── web_search.py # SIN CAMBIOS (de v2)
│ └── calculator.py # SIN CAMBIOS (de v1)
├── state/
│ └── research_state.py # EXTENDIDO — UserProfile model
├── config/
│ └── settings.py # EXTENDIDO — config de memoria
├── memory/ # NUEVO
│ └── user_store.py # InMemoryStore + persistencia a JSON
├── utils/
│ ├── retry.py # SIN CAMBIOS (de v2)
│ └── logger.py # SIN CAMBIOS (de v2)
└── main.py # MODIFICADO — CLI multi-usuario con memoria
Paso 1: Configuración de memoria (config/settings.py)
"""
config/settings.py
Configuración del AI Research Assistant v3.
"""
from dotenv import load_dotenv
load_dotenv()
MODEL_NAME = "openai:gpt-4.1-mini"
MODEL_TEMPERATURE = 0.2
MAX_SUB_QUERIES = 4
SEARCH_SOURCES = ["web", "academic", "news"]
RETRY_MAX_ATTEMPTS = 3
RETRY_BASE_DELAY = 1.0
RETRY_MAX_DELAY = 10.0
RETRY_JITTER = True
MIN_SOURCES_FOR_REPORT = 1
# v3: Memory
MAX_CONVERSATION_MESSAGES = 20
MEMORY_PERSIST_FILE = "memory_store.json"
DEFAULT_FORMAT = "paragraphs"
DEFAULT_DETAIL_LEVEL = "standard"
DEFAULT_SOURCES_PRIORITY = ["web", "academic", "news"]Paso 2: Gestión de memoria del usuario (memory/user_store.py)
El módulo central de la v3. Maneja InMemoryStore, persistencia a JSON, y operaciones de alto nivel sobre preferencias e historial.
"""
memory/user_store.py
Long-term memory para el AI Research Assistant v3.
InMemoryStore con persistencia a JSON para desarrollo.
"""
import json
import os
from datetime import datetime
from langgraph.store.memory import InMemoryStore
from config.settings import MEMORY_PERSIST_FILE
def create_store(persist_file: str = None) -> InMemoryStore:
"""Crea un InMemoryStore y carga datos persistidos si existen."""
store = InMemoryStore()
filepath = persist_file or MEMORY_PERSIST_FILE
if os.path.exists(filepath):
with open(filepath, "r", encoding="utf-8") as f:
data = json.load(f)
for entry in data:
store.put(tuple(entry["namespace"]), entry["key"], entry["value"])
print(f" [Memory] Cargados {len(data)} items desde {filepath}")
else:
print(f" [Memory] Store vacío — {filepath} no encontrado")
return store
def persist_store(store: InMemoryStore, persist_file: str = None):
"""Serializa el InMemoryStore a JSON."""
filepath = persist_file or MEMORY_PERSIST_FILE
all_items = []
for item in store.search(("users",)):
all_items.append({
"namespace": list(item.namespace),
"key": item.key,
"value": item.value,
})
with open(filepath, "w", encoding="utf-8") as f:
json.dump(all_items, f, indent=2, ensure_ascii=False, default=str)
print(f" [Memory] Persistidos {len(all_items)} items a {filepath}")
def record_session(store: InMemoryStore, user_id: str, topic: str, summary: str):
"""Registra una sesión de investigación completada."""
history_ns = ("users", user_id, "history")
sessions = store.search(history_ns)
session_num = len(sessions) + 1
store.put(history_ns, f"session_{session_num:04d}", {
"topic": topic,
"summary": summary[:200],
"timestamp": datetime.now().isoformat(),
"session_number": session_num,
})
topics_ns = ("users", user_id, "topics")
existing = store.get(topics_ns, topic)
count = (existing.value["count"] + 1) if existing else 1
store.put(topics_ns, topic, {"count": count, "last_researched": datetime.now().isoformat()})
def get_user_profile(store: InMemoryStore, user_id: str) -> dict:
"""Construye el perfil completo del usuario."""
history = store.search(("users", user_id, "history"))
topics = store.search(("users", user_id, "topics"))
prefs = store.search(("users", user_id, "preferences"))
sorted_topics = sorted(topics, key=lambda x: x.value["count"], reverse=True)
last_session = None
if history:
sorted_h = sorted(history, key=lambda x: x.value.get("timestamp", ""), reverse=True)
last_session = sorted_h[0].value
return {
"total_sessions": len(history),
"top_topics": [(t.key, t.value["count"]) for t in sorted_topics[:5]],
"last_session": last_session,
"preferences": {item.key: item.value for item in prefs},
"is_new_user": len(history) == 0,
}
def get_preferences(store: InMemoryStore, user_id: str) -> dict:
"""Preferencias del usuario con defaults."""
from config.settings import DEFAULT_FORMAT, DEFAULT_DETAIL_LEVEL, DEFAULT_SOURCES_PRIORITY
defaults = {"format": DEFAULT_FORMAT, "detail_level": DEFAULT_DETAIL_LEVEL, "sources_priority": DEFAULT_SOURCES_PRIORITY}
items = store.search(("users", user_id, "preferences"))
user_prefs = {item.key: item.value.get("value", item.value) for item in items}
return {**defaults, **user_prefs}
def save_preference(store: InMemoryStore, user_id: str, key: str, value, reason: str = ""):
store.put(("users", user_id, "preferences"), key, {
"value": value, "reason": reason, "updated_at": datetime.now().isoformat(),
})
def detect_preferences_from_input(store: InMemoryStore, user_id: str, text: str):
"""Auto-detecta preferencias del input del usuario."""
t = text.lower()
if "bullet" in t or "lista" in t or "puntos" in t:
save_preference(store, user_id, "format", "bullet_points", f"Detected: '{text[:50]}'")
if "detallado" in t or "profundidad" in t or "completo" in t:
save_preference(store, user_id, "detail_level", "detailed", f"Detected: '{text[:50]}'")
if "paper" in t or "arxiv" in t or "académic" in t:
save_preference(store, user_id, "sources_priority", ["academic", "web", "news"], f"Detected: '{text[:50]}'")
if "breve" in t or "corto" in t or "resumido" in t:
save_preference(store, user_id, "detail_level", "brief", f"Detected: '{text[:50]}'")create_store carga desde JSON al iniciar. persist_store guarda al terminar. Esto da la experiencia de "cierra la terminal, ábrela de nuevo, el agente recuerda" — sin PostgreSQL.
Paso 3: Extender modelos de datos (state/research_state.py)
"""
state/research_state.py
Modelos de datos para el AI Research Assistant v3.
"""
from pydantic import BaseModel, Field
from datetime import datetime
class Source(BaseModel):
name: str
source_type: str
content: str
class KeyFinding(BaseModel):
title: str
description: str
confidence: float = Field(ge=0.0, le=1.0)
class SourceStatus(BaseModel):
source_type: str
status: str
attempts: int = Field(default=1)
error: str = Field(default="")
duration_ms: float = Field(default=0)
class UserProfile(BaseModel):
"""Perfil del usuario desde long-term memory (v3)."""
user_id: str
total_sessions: int = 0
top_topics: list[tuple[str, int]] = Field(default_factory=list)
last_topic: str = ""
preferred_format: str = "paragraphs"
is_new_user: bool = True
class ResearchReport(BaseModel):
topic: str
summary: str
key_findings: list[KeyFinding] = Field(min_length=1)
sources: list[Source] = Field(min_length=1)
sub_queries: list[str]
confidence: float = Field(ge=0.0, le=1.0)
generated_at: str = Field(default_factory=lambda: datetime.now().isoformat())
source_availability: list[SourceStatus] = Field(default_factory=list)
version: str = Field(default="v3")
user_id: str = Field(default="anonymous")
session_number: int = Field(default=0)
class SubQuery(BaseModel):
query: str
rationale: strPaso 4: Agente v3 con memoria (agents/researcher.py)
El @entrypoint ahora tiene acceso al store, genera un saludo personalizado, detecta preferencias, y registra sesiones.
"""
agents/researcher.py
AI Research Assistant v3 — long-term memory, greeting, multi-user.
"""
import json
import time
import uuid
from langchain.chat_models import init_chat_model
from langgraph.func import entrypoint, task
from langgraph.checkpoint.memory import MemorySaver
from langgraph.store.base import BaseStore
import sys
sys.path.insert(0, ".")
from config.settings import (
MODEL_NAME, MODEL_TEMPERATURE, MAX_SUB_QUERIES,
SEARCH_SOURCES, MIN_SOURCES_FOR_REPORT,
)
from state.research_state import ResearchReport, Source, KeyFinding, SubQuery, SourceStatus
from tools.web_search import search_with_retry
from tools.calculator import calculate_confidence
from utils.logger import ResearchLogger
from memory.user_store import (
get_user_profile, get_preferences, record_session, detect_preferences_from_input,
)
model = init_chat_model(MODEL_NAME, temperature=MODEL_TEMPERATURE)
agent_logger = ResearchLogger("research_agent_v3")
@task
def generate_greeting(user_id: str, profile: dict) -> str:
if profile["is_new_user"]:
return "¡Hola! Soy tu asistente de investigación. Aprenderé tus preferencias con el tiempo."
parts = ["¡Bienvenido de vuelta!"]
if profile["last_session"]:
topic = profile["last_session"].get("topic", "")
date = profile["last_session"].get("timestamp", "")[:10]
parts.append(f"Última investigación: '{topic}' ({date}).")
if profile["total_sessions"] > 0:
parts.append(f"Llevas {profile['total_sessions']} sesiones.")
if profile["top_topics"] and profile["top_topics"][0][1] >= 2:
top = profile["top_topics"][0]
parts.append(f"Tema favorito: '{top[0]}' ({top[1]} veces).")
prefs = profile.get("preferences", {})
if "format" in prefs:
parts.append(f"Formato: {prefs['format'].get('value', 'estándar')}.")
return " ".join(parts)
@task
def decompose_query(topic: str) -> list[dict]:
response = model.invoke(
f"Descompone este tema en {MAX_SUB_QUERIES} sub-preguntas investigables.\n\n"
f"Tema: {topic}\n\n"
f'Responde en JSON: [{{"query": "...", "rationale": "..."}}]\nSolo JSON.'
)
try:
return json.loads(response.content)[:MAX_SUB_QUERIES]
except json.JSONDecodeError:
return [{"query": topic, "rationale": "Fallback"}, {"query": f"avances en {topic}", "rationale": "Tendencias"}]
@task
def search_all_sources_v2(query: str, logger: ResearchLogger) -> list[dict]:
start = time.time()
futures = [search_with_retry(query, src, logger) for src in SEARCH_SOURCES]
results = [f.result() for f in futures]
return results
@task
def merge_and_deduplicate(all_results: list[dict]) -> list[dict]:
seen, unique = set(), []
for r in all_results:
if r["search_status"] != "ok":
continue
key = f"{r['source_type']}:{r['content'][:100]}"
if key not in seen:
seen.add(key)
unique.append(r)
unique.sort(key=lambda r: r.get("relevance", 0), reverse=True)
return unique
@task
def synthesize_findings(topic: str, results: list[dict]) -> list[dict]:
text = "\n".join(f"Fuente {i} ({r['source_type']}): {r['content']}" for i, r in enumerate(results, 1))
response = model.invoke(
f"Identifica 3-5 hallazgos clave sobre '{topic}'.\n\nFuentes:\n{text}\n\n"
f'JSON: [{{"title": "...", "description": "...", "confidence": 0.8}}]\nSolo JSON.'
)
try:
return json.loads(response.content)[:5]
except json.JSONDecodeError:
return [{"title": "Hallazgo general", "description": f"Investigación sobre {topic} relevante.", "confidence": 0.6}]
@task
def generate_summary(topic: str, findings: list[dict], source_info: str, user_prefs: dict) -> str:
findings_text = "\n".join(f"- {f['title']}: {f['description']}" for f in findings)
fmt = ""
if user_prefs.get("format") == "bullet_points":
fmt = "Usa bullet points. "
if user_prefs.get("detail_level") == "brief":
fmt += "Sé breve (1-2 oraciones). "
elif user_prefs.get("detail_level") == "detailed":
fmt += "Sé detallado (4-5 oraciones). "
response = model.invoke(
f"Resumen ejecutivo sobre '{topic}'.\nHallazgos:\n{findings_text}\n"
f"Fuentes: {source_info}\n{fmt}Solo el resumen."
)
return response.content.strip()
def create_research_agent(checkpointer, store):
"""Factory que crea el research agent v3."""
@entrypoint(checkpointer=checkpointer, store=store)
def research_agent(topic: str, *, store: BaseStore) -> dict:
config = entrypoint.get_config()
user_id = config["configurable"].get("user_id", "anonymous")
request_id = uuid.uuid4().hex[:8]
agent_logger.set_request_id(request_id)
pipeline_start = time.time()
profile = get_user_profile(store, user_id)
user_prefs = get_preferences(store, user_id)
greeting = generate_greeting(user_id, profile).result()
print(f"\n{'=' * 60}")
print(f" 🔬 AI Research Assistant v3")
print(f" {greeting}")
print(f" Tema: {topic} | User: {user_id} | Request: {request_id}")
print(f"{'=' * 60}")
detect_preferences_from_input(store, user_id, topic)
# Decompose
print(f"\n📋 Descomponiendo tema...")
sub_queries_raw = decompose_query(topic).result()
sub_queries = [SubQuery(**sq) for sq in sub_queries_raw]
for i, sq in enumerate(sub_queries, 1):
print(f" {i}. {sq.query}")
# Search
print(f"\n🔍 Buscando en {len(SEARCH_SOURCES)} fuentes...")
all_raw, source_statuses = [], []
search_futures = [search_all_sources_v2(sq.query, agent_logger) for sq in sub_queries]
for i, future in enumerate(search_futures):
results = future.result()
all_raw.extend(results)
ok = sum(1 for r in results if r["search_status"] == "ok")
print(f" Sub-query {i + 1}: {ok}/{len(results)} OK")
for r in results:
source_statuses.append(SourceStatus(
source_type=r["source_type"], status=r["search_status"],
attempts=r.get("attempts", 1), error=r.get("error", ""),
duration_ms=r.get("duration_ms", 0),
))
total_ok = sum(1 for r in all_raw if r["search_status"] == "ok")
total_failed = len(all_raw) - total_ok
if total_ok < MIN_SOURCES_FOR_REPORT:
print(f"\n ❌ Fuentes insuficientes ({total_ok}). Abortando.")
return {"error": f"Solo {total_ok} fuentes. Mínimo: {MIN_SOURCES_FOR_REPORT}", "version": "v3"}
# Merge + Synthesize + Summary
unique = merge_and_deduplicate(all_raw).result()
print(f"\n🔀 {len(all_raw)} raw → {len(unique)} únicos")
findings_raw = synthesize_findings(topic, unique).result()
findings = [KeyFinding(**f) for f in findings_raw]
print(f"🧠 {len(findings)} hallazgos identificados")
source_info = f"{total_ok}/{total_ok + total_failed} fuentes OK"
summary = generate_summary(topic, findings_raw, source_info, user_prefs).result()
# Confidence
avg_rel = sum(r["relevance"] for r in unique) / len(unique) if unique else 0.5
base_conf = calculate_confidence(len(unique), avg_rel, len(findings)).result()
avail_factor = total_ok / (total_ok + total_failed) if (total_ok + total_failed) > 0 else 0.5
confidence = round(base_conf * (0.7 + 0.3 * avail_factor), 2)
# Build report + record session
sources = [Source(name=r["source_name"], source_type=r["source_type"], content=r["content"]) for r in unique]
report = ResearchReport(
topic=topic, summary=summary, key_findings=findings, sources=sources,
sub_queries=[sq.query for sq in sub_queries], confidence=confidence,
source_availability=source_statuses, version="v3",
user_id=user_id, session_number=profile["total_sessions"] + 1,
)
record_session(store, user_id, topic, summary[:200])
pipeline_ms = (time.time() - pipeline_start) * 1000
print(f"\n📄 Reporte v3 generado. Sesión #{report.session_number}. ({pipeline_ms:.0f}ms)")
print(f"{'=' * 60}")
return report.model_dump()
return research_agentPaso 5: Implementar reanudación de sesiones interrumpidas
El checkpointer guarda el estado en cada paso del pipeline. Si el proceso se interrumpe, la siguiente ejecución con el mismo thread_id puede detectar la sesión incompleta. La lógica de detección vive en el CLI:
def check_incomplete_sessions(checkpointer, user_id: str) -> list[dict]:
"""Busca threads del usuario que podrían estar incompletos."""
# En producción con PostgresSaver, esto sería una query a la DB.
# Con MemorySaver en desarrollo, los checkpoints se pierden al reiniciar.
# La detección real funciona dentro del mismo proceso.
return []Con MemorySaver, los checkpoints viven en RAM — se pierden al reiniciar. El long-term store (persistido a JSON) es lo que sobrevive. Para reanudación real ante crashes, necesitas PostgresSaver + PostgresStore en producción.
Paso 6: CLI multi-usuario con persistencia (main.py)
"""
main.py
CLI del AI Research Assistant v3. Multi-user con long-term memory.
"""
import sys
import json
import uuid
sys.path.insert(0, ".")
from langgraph.checkpoint.memory import MemorySaver
from memory.user_store import create_store, persist_store, get_user_profile, get_preferences, save_preference
from agents.researcher import create_research_agent
def format_report(report: dict) -> str:
if "error" in report:
return f"\n ❌ {report['error']}"
lines = [
"", "╔" + "═" * 58 + "╗",
"║" + f" 📄 REPORTE v3 | {report.get('user_id')} | Sesión #{report.get('session_number')}".center(58) + "║",
"╚" + "═" * 58 + "╝",
f"\n📌 Tema: {report['topic']}",
f"🎯 Confianza: {report['confidence']:.0%}",
f"\n{'─' * 60}", "📋 RESUMEN", f"{'─' * 60}", report["summary"],
f"\n{'─' * 60}", "💡 HALLAZGOS", f"{'─' * 60}",
]
for i, f in enumerate(report["key_findings"], 1):
lines.append(f" {i}. {f['title']} [{f['confidence']:.0%}]")
lines.append(f" {f['description']}")
lines.extend([f"\n{'─' * 60}", f"📚 FUENTES ({len(report['sources'])})", f"{'─' * 60}"])
for s in report["sources"]:
lines.append(f" • [{s['source_type'].upper()}] {s['name']}")
if report.get("source_availability"):
lines.extend([f"\n{'─' * 60}", "🔌 DISPONIBILIDAD", f"{'─' * 60}"])
for sa in report["source_availability"]:
icon = "✅" if sa["status"] == "ok" else "❌"
retry = f" ({sa['attempts']} intentos)" if sa["attempts"] > 1 else ""
lines.append(f" {icon} {sa['source_type']}: {sa['status']}{retry}")
lines.append(f"\n{'═' * 60}")
return "\n".join(lines)
def run_interactive():
print("=" * 60)
print(" 🔬 AI Research Assistant v3 — Memoria persistente")
print("=" * 60)
store = create_store()
checkpointer = MemorySaver()
agent = create_research_agent(checkpointer, store)
current_user = None
print("\n Comandos: user <nombre> | profile | pref <k> <v> | salir\n")
while True:
if not current_user:
try:
current_user = input("👤 Usuario: ").strip()
except (KeyboardInterrupt, EOFError):
break
if not current_user:
continue
profile = get_user_profile(store, current_user)
if profile["is_new_user"]:
print(f" ¡Bienvenido, {current_user}! Primera vez aquí.\n")
else:
print(f" ¡Bienvenido, {current_user}! ({profile['total_sessions']} sesiones)")
if profile["last_session"]:
print(f" Última: '{profile['last_session']['topic']}' ({profile['last_session']['timestamp'][:10]})\n")
continue
try:
user_input = input(f"🔎 [{current_user}] ").strip()
except (KeyboardInterrupt, EOFError):
break
if not user_input:
continue
if user_input.lower() in ("salir", "exit", "quit"):
break
if user_input.lower().startswith("user "):
current_user = user_input[5:].strip()
profile = get_user_profile(store, current_user)
status = "nuevo" if profile["is_new_user"] else f"{profile['total_sessions']} sesiones"
print(f" Cambiado a {current_user} ({status})\n")
continue
if user_input.lower() == "profile":
profile = get_user_profile(store, current_user)
prefs = get_preferences(store, current_user)
print(f"\n Sesiones: {profile['total_sessions']}")
for t, c in profile["top_topics"]:
print(f" - {t} ({c}x)")
print(f" Prefs: {json.dumps(prefs, ensure_ascii=False)}\n")
continue
if user_input.lower().startswith("pref "):
parts = user_input.split(maxsplit=2)
if len(parts) == 3:
save_preference(store, current_user, parts[1], parts[2], "Manual")
print(f" ✓ {parts[1]} = {parts[2]}\n")
continue
thread_id = f"v3-{current_user}-{uuid.uuid4().hex[:8]}"
try:
report = agent.invoke(
user_input,
config={"configurable": {"thread_id": thread_id, "user_id": current_user}},
)
print(format_report(report))
except Exception as e:
print(f"\n ❌ Error: {e}\n")
print("\n Guardando memoria...")
persist_store(store)
print(" ¡Hasta luego!")
if __name__ == "__main__":
if len(sys.argv) > 1:
user = "default"
args = sys.argv[1:]
if args[0] == "--user" and len(args) >= 3:
user, args = args[1], args[2:]
store = create_store()
agent = create_research_agent(MemorySaver(), store)
thread_id = f"v3-{user}-{uuid.uuid4().hex[:8]}"
report = agent.invoke(" ".join(args), config={"configurable": {"thread_id": thread_id, "user_id": user}})
print(format_report(report))
persist_store(store)
else:
run_interactive()Ejecución: el momento poderoso
Primera vez: usuario nuevo
cd research-assistant
python main.py [Memory] Store vacío — memory_store.json no encontrado
👤 Usuario: mike
¡Bienvenido, mike! Primera vez aquí.
🔎 [mike] RAG techniques
🔬 AI Research Assistant v3
¡Hola! Soy tu asistente de investigación. Aprenderé tus preferencias con el tiempo.
... (pipeline) ...
📄 Reporte v3 generado. Sesión #1.
🔎 [mike] salir
[Memory] Persistidos 3 items a memory_store.json
Segunda vez: el agente recuerda
python main.py [Memory] Cargados 3 items desde memory_store.json
👤 Usuario: mike
¡Bienvenido, mike! (1 sesiones)
Última: 'RAG techniques' (2026-03-08)
🔎 [mike] prompt engineering
🔬 AI Research Assistant v3
¡Bienvenido de vuelta! Última investigación: 'RAG techniques' (2026-03-08). Llevas 1 sesiones.
...
Cerraste la terminal. Abriste una nueva. El agente recuerda.
Multi-usuario
🔎 [mike] user ana
Cambiado a ana (nuevo)
🔎 [ana] machine learning
¡Hola! Soy tu asistente de investigación...
🔎 [ana] user mike
Cambiado a mike (2 sesiones)
Ana y mike tienen contextos completamente aislados.
Criterios de éxito
- ✅ Resume ante interrupciones — checkpointer guarda progreso en cada paso
- ✅ Recuerda preferencias entre sesiones — store persiste a JSON, se carga al iniciar
- ✅ Saludo personalizado — usuario nuevo vs recurrente reciben saludos diferentes
- ✅ Detección automática — "papers de arxiv" guarda preferencia por fuentes académicas
- ✅ Multi-usuario — cada usuario tiene datos aislados
- ✅ Memoria sobrevive reinicios — cierra terminal, abre nueva, preferencias persisten
- ✅ Reportes con metadata — user_id y session_number en el JSON
Escenarios de prueba
Test 1: Memoria cross-session
python main.py --user mike "RAG techniques"
# → Genera reporte, guarda en memory_store.json
python main.py --user mike "prompt engineering"
# → Saludo incluye: "Última investigación: 'RAG techniques'"Test 2: Detección de preferencias
🔎 [mike] Quiero bullet points y fuentes académicas sobre AI
# → Detecta format=bullet_points, sources_priority=academic
# → Resumen usa bullet points
Test 3: Multi-usuario aislado
👤 alice → investiga quantum computing (sesión #1)
user bob → investiga ML (sesión #1, no sabe de alice)
user alice → profile → muestra 1 sesión sobre quantum computing
Test 4: Usuario con historial extenso
for topic in "RAG" "Embeddings" "RAG" "Fine-tuning" "RAG"; do
python main.py --user mike "$topic"
done
python main.py --user mike "AI agents"
# → "Tema favorito: 'RAG' (3 veces)."Errores comunes
1. La memoria no persiste entre ejecuciones
Causa: persist_store() no se llamó antes de terminar. Con Ctrl+C puede saltarse.
Solución: Envuelve el CLI en try/finally:
try:
run_interactive()
finally:
persist_store(store)2. TypeError: got unexpected keyword argument 'store'
Causa: El @entrypoint no tiene store= en su decorador.
Solución: Verifica la firma: @entrypoint(checkpointer=checkpointer, store=store) y el parámetro *, store: BaseStore.
3. Datos de un usuario aparecen en otro
Causa: El namespace no incluye user_id.
Solución: Todos los accesos usan ("users", user_id, ...). Verifica que user_id viene de config["configurable"]["user_id"].
4. JSON de persistencia crece sin límite
Causa: Cada sesión agrega items y persist_store guarda todo.
Solución: Implementa retención:
def cleanup_old_sessions(store, user_id, max_sessions=100):
sessions = store.search(("users", user_id, "history"))
if len(sessions) > max_sessions:
oldest = sorted(sessions, key=lambda x: x.value.get("timestamp", ""))
for old in oldest[:-max_sessions]:
store.delete(("users", user_id, "history"), old.key)5. entrypoint.get_config() retorna None para user_id
Causa: No pasaste user_id en el config al invocar.
Solución: Incluye siempre:
config={"configurable": {"thread_id": thread_id, "user_id": "mike"}}6. Preferencias auto-detectadas sobrescriben manuales
Causa: detect_preferences_from_input corre en cada invocación.
Solución: Agrega flag manual y verifica:
def save_preference(store, uid, key, value, reason="", manual=False):
existing = store.get(("users", uid, "preferences"), key)
if existing and existing.value.get("manual") and not manual:
return # No sobrescribir preferencia manual
store.put(("users", uid, "preferences"), key, {"value": value, "manual": manual, ...})Lo que viene: Módulo 9 — Human-in-the-Loop
Tu Research Assistant v3 tiene memoria: recuerda preferencias, acumula historial, personaliza cada interacción. Pero toma todas las decisiones solo. Si decide buscar en una fuente costosa, lo hace sin preguntar.
El Módulo 9 agrega human-in-the-loop — el agente pausa y pide aprobación antes de acciones costosas. "Voy a buscar en 5 fuentes premium. ¿Confirmas?" La persistencia de este módulo es el prerequisito: el agente necesita guardar su estado para pausarse indefinidamente y resumir cuando reciba tu aprobación.
Recursos del proyecto
- LangGraph Memory Store — Long-term memory oficial
- LangGraph Persistence — Checkpointing y MemorySaver
- LangGraph Cross-thread Memory — Memoria entre threads
- LangGraph Functional API — Store en @entrypoint
- InMemoryStore Reference — API reference
- LangGraph Human-in-the-Loop — Preview del Módulo 9
Módulo 8 — LangChain & LangGraph: From Chains to Agents