Module 8: Memory and Persistence
Evolving Project: Persistence and Memory (v3)
Project overview
In Module 7, you built v2 of the AI Research Assistant: a robust agent with retry logic, protected parallel search, merge with deduplication, and graceful degradation. If an API fails, it retries. If they all fail, it degrades gracefully. It's production-ready when it comes to resilience.
But it has one fundamental problem: it's amnesiac. Every run starts from zero. If you researched "RAG techniques" yesterday and today you ask about "retrieval augmented generation", the agent doesn't connect the dots. If the process is interrupted halfway through a research run, you lose everything. If a second user comes along, there's no isolation.
v3 adds three layers on top of v2: checkpointing with MemorySaver so the agent saves progress at every step and can resume after interruptions, long-term memory with InMemoryStore so it remembers preferences and history across sessions, and multi-user support so each user gets isolated context.
The moment that defines this project: you run the agent, research a topic, close the terminal. You open a new terminal, run the agent as the same user. The agent says: "Welcome back! Last time you researched RAG techniques. Want to continue or start something new?" That moment — the agent that remembers — is what separates a prototype from a product.
Project goal
Evolve the AI Research Assistant from v2 (robust but amnesiac) to v3 (robust with memory), adding checkpointing, long-term memory, and multi-user support.
By the end of this project:
- 🔧 You'll implement checkpointing with MemorySaver to save progress at every step
- 🔧 You'll add InMemoryStore for preferences and history across sessions
- 🔧 You'll create a personalized greeting that uses the user's memory
- 🔧 You'll implement detection and resumption of interrupted sessions
- 🔧 You'll support multiple users with complete isolation
- 🔧 You'll persist memory to JSON so it survives restarts
Before and after
v2 (Module 7): robust but amnesiac
Session 1: "Research RAG" → Report → END
Session 2: "Research RAG" → Starts from zero. Doesn't know you already researched it.
Crash halfway → Everything lost.
User A and B → Same context. No isolation.
v3 (This module): robust with memory
Session 1: "Research RAG" → Report → Saved to history
Session 2: "Research RAG" → "You already researched this. Continue or take a new angle?"
Crash halfway → Resumes from the last checkpoint.
User A → Their isolated context. User B → Theirs.
Technical specs
| Component | Version | Purpose |
|---|---|---|
| Python | 3.11+ | Runtime |
| LangChain | v1.2+ | LLM framework |
| LangGraph | v1.0+ | Functional API + Store + Checkpointer |
| langchain-openai | latest | Model provider |
| pydantic | v2+ | Structured models |
Project structure
research-assistant/
├── .env
├── requirements.txt
├── agents/
│ └── researcher.py # MODIFIED — v3 with store + greeting
├── tools/
│ ├── web_search.py # UNCHANGED (from v2)
│ └── calculator.py # UNCHANGED (from v1)
├── state/
│ └── research_state.py # EXTENDED — UserProfile model
├── config/
│ └── settings.py # EXTENDED — memory config
├── memory/ # NEW
│ └── user_store.py # InMemoryStore + JSON persistence
├── utils/
│ ├── retry.py # UNCHANGED (from v2)
│ └── logger.py # UNCHANGED (from v2)
└── main.py # MODIFIED — multi-user CLI with memory
Step 1: Memory configuration (config/settings.py)
"""
config/settings.py
Configuration for the 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"]Step 2: User memory management (memory/user_store.py)
The central module of v3. It handles the InMemoryStore, JSON persistence, and high-level operations on preferences and history.
"""
memory/user_store.py
Long-term memory for the AI Research Assistant v3.
InMemoryStore with JSON persistence for development.
"""
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:
"""Creates an InMemoryStore and loads persisted data if it exists."""
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] Loaded {len(data)} items from {filepath}")
else:
print(f" [Memory] Empty store — {filepath} not found")
return store
def persist_store(store: InMemoryStore, persist_file: str = None):
"""Serializes the InMemoryStore to 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] Persisted {len(all_items)} items to {filepath}")
def record_session(store: InMemoryStore, user_id: str, topic: str, summary: str):
"""Records a completed research session."""
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:
"""Builds the user's complete profile."""
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:
"""The user's preferences, with 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-detects preferences from the user's input."""
t = text.lower()
if "bullet" in t or "list" in t or "points" in t:
save_preference(store, user_id, "format", "bullet_points", f"Detected: '{text[:50]}'")
if "detailed" in t or "in depth" in t or "thorough" in t:
save_preference(store, user_id, "detail_level", "detailed", f"Detected: '{text[:50]}'")
if "paper" in t or "arxiv" in t or "academic" in t:
save_preference(store, user_id, "sources_priority", ["academic", "web", "news"], f"Detected: '{text[:50]}'")
if "brief" in t or "short" in t or "summarized" in t:
save_preference(store, user_id, "detail_level", "brief", f"Detected: '{text[:50]}'")create_store loads from JSON on startup. persist_store saves on shutdown. That's what gives you the "close the terminal, open it again, the agent remembers" experience — without PostgreSQL.
Step 3: Extend the data models (state/research_state.py)
"""
state/research_state.py
Data models for the 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):
"""The user's profile from 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: strStep 4: The v3 agent with memory (agents/researcher.py)
The @entrypoint now has access to the store, generates a personalized greeting, detects preferences, and records sessions.
"""
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 "Hi! I'm your research assistant. I'll learn your preferences over time."
parts = ["Welcome back!"]
if profile["last_session"]:
topic = profile["last_session"].get("topic", "")
date = profile["last_session"].get("timestamp", "")[:10]
parts.append(f"Last research: '{topic}' ({date}).")
if profile["total_sessions"] > 0:
parts.append(f"You've had {profile['total_sessions']} sessions.")
if profile["top_topics"] and profile["top_topics"][0][1] >= 2:
top = profile["top_topics"][0]
parts.append(f"Favorite topic: '{top[0]}' ({top[1]} times).")
prefs = profile.get("preferences", {})
if "format" in prefs:
parts.append(f"Format: {prefs['format'].get('value', 'standard')}.")
return " ".join(parts)
@task
def decompose_query(topic: str) -> list[dict]:
response = model.invoke(
f"Break this topic down into {MAX_SUB_QUERIES} researchable sub-questions.\n\n"
f"Topic: {topic}\n\n"
f'Answer in JSON: [{{"query": "...", "rationale": "..."}}]\nJSON only.'
)
try:
return json.loads(response.content)[:MAX_SUB_QUERIES]
except json.JSONDecodeError:
return [{"query": topic, "rationale": "Fallback"}, {"query": f"advances in {topic}", "rationale": "Trends"}]
@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"Source {i} ({r['source_type']}): {r['content']}" for i, r in enumerate(results, 1))
response = model.invoke(
f"Identify 3-5 key findings about '{topic}'.\n\nSources:\n{text}\n\n"
f'JSON: [{{"title": "...", "description": "...", "confidence": 0.8}}]\nJSON only.'
)
try:
return json.loads(response.content)[:5]
except json.JSONDecodeError:
return [{"title": "General finding", "description": f"Relevant research on {topic}.", "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 = "Use bullet points. "
if user_prefs.get("detail_level") == "brief":
fmt += "Be brief (1-2 sentences). "
elif user_prefs.get("detail_level") == "detailed":
fmt += "Be detailed (4-5 sentences). "
response = model.invoke(
f"Executive summary about '{topic}'.\nFindings:\n{findings_text}\n"
f"Sources: {source_info}\n{fmt}The summary only."
)
return response.content.strip()
def create_research_agent(checkpointer, store):
"""Factory that creates the v3 research agent."""
@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" Topic: {topic} | User: {user_id} | Request: {request_id}")
print(f"{'=' * 60}")
detect_preferences_from_input(store, user_id, topic)
# Decompose
print(f"\n📋 Breaking the topic down...")
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🔍 Searching {len(SEARCH_SOURCES)} sources...")
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 ❌ Not enough sources ({total_ok}). Aborting.")
return {"error": f"Only {total_ok} sources. Minimum: {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)} unique")
findings_raw = synthesize_findings(topic, unique).result()
findings = [KeyFinding(**f) for f in findings_raw]
print(f"🧠 {len(findings)} findings identified")
source_info = f"{total_ok}/{total_ok + total_failed} sources 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📄 v3 report generated. Session #{report.session_number}. ({pipeline_ms:.0f}ms)")
print(f"{'=' * 60}")
return report.model_dump()
return research_agentStep 5: Implement resumption of interrupted sessions
The checkpointer saves state at every step of the pipeline. If the process is interrupted, the next run with the same thread_id can detect the incomplete session. The detection logic lives in the CLI:
def check_incomplete_sessions(checkpointer, user_id: str) -> list[dict]:
"""Looks for the user's threads that might be incomplete."""
# In production with PostgresSaver, this would be a DB query.
# With MemorySaver in development, checkpoints are lost on restart.
# Real detection only works within the same process.
return []With MemorySaver, checkpoints live in RAM — they're lost on restart. The long-term store (persisted to JSON) is what survives. For real crash resumption, you need PostgresSaver + PostgresStore in production.
Step 6: Multi-user CLI with persistence (main.py)
"""
main.py
CLI for the AI Research Assistant v3. Multi-user with 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" 📄 REPORT v3 | {report.get('user_id')} | Session #{report.get('session_number')}".center(58) + "║",
"╚" + "═" * 58 + "╝",
f"\n📌 Topic: {report['topic']}",
f"🎯 Confidence: {report['confidence']:.0%}",
f"\n{'─' * 60}", "📋 SUMMARY", f"{'─' * 60}", report["summary"],
f"\n{'─' * 60}", "💡 FINDINGS", 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"📚 SOURCES ({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}", "🔌 AVAILABILITY", f"{'─' * 60}"])
for sa in report["source_availability"]:
icon = "✅" if sa["status"] == "ok" else "❌"
retry = f" ({sa['attempts']} attempts)" 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 — Persistent memory")
print("=" * 60)
store = create_store()
checkpointer = MemorySaver()
agent = create_research_agent(checkpointer, store)
current_user = None
print("\n Commands: user <name> | profile | pref <k> <v> | exit\n")
while True:
if not current_user:
try:
current_user = input("👤 User: ").strip()
except (KeyboardInterrupt, EOFError):
break
if not current_user:
continue
profile = get_user_profile(store, current_user)
if profile["is_new_user"]:
print(f" Welcome, {current_user}! First time here.\n")
else:
print(f" Welcome back, {current_user}! ({profile['total_sessions']} sessions)")
if profile["last_session"]:
print(f" Last: '{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 ("exit", "quit", "bye"):
break
if user_input.lower().startswith("user "):
current_user = user_input[5:].strip()
profile = get_user_profile(store, current_user)
status = "new" if profile["is_new_user"] else f"{profile['total_sessions']} sessions"
print(f" Switched to {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 Sessions: {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 Saving memory...")
persist_store(store)
print(" See you next time!")
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()Running it: the powerful moment
First time: a new user
cd research-assistant
python main.py [Memory] Empty store — memory_store.json not found
👤 User: mike
Welcome, mike! First time here.
🔎 [mike] RAG techniques
🔬 AI Research Assistant v3
Hi! I'm your research assistant. I'll learn your preferences over time.
... (pipeline) ...
📄 v3 report generated. Session #1.
🔎 [mike] exit
[Memory] Persisted 3 items to memory_store.json
Second time: the agent remembers
python main.py [Memory] Loaded 3 items from memory_store.json
👤 User: mike
Welcome back, mike! (1 sessions)
Last: 'RAG techniques' (2026-03-08)
🔎 [mike] prompt engineering
🔬 AI Research Assistant v3
Welcome back! Last research: 'RAG techniques' (2026-03-08). You've had 1 sessions.
...
You closed the terminal. You opened a new one. The agent remembers.
Multi-user
🔎 [mike] user ana
Switched to ana (new)
🔎 [ana] machine learning
Hi! I'm your research assistant...
🔎 [ana] user mike
Switched to mike (2 sessions)
Ana and mike have completely isolated contexts.
Success criteria
- ✅ Resumes after interruptions — the checkpointer saves progress at every step
- ✅ Remembers preferences across sessions — the store persists to JSON and loads on startup
- ✅ Personalized greeting — new and returning users get different greetings
- ✅ Automatic detection — "papers from arxiv" saves a preference for academic sources
- ✅ Multi-user — each user has isolated data
- ✅ Memory survives restarts — close the terminal, open a new one, preferences persist
- ✅ Reports with metadata — user_id and session_number in the JSON
Test scenarios
Test 1: Cross-session memory
python main.py --user mike "RAG techniques"
# → Generates a report, saves it to memory_store.json
python main.py --user mike "prompt engineering"
# → The greeting includes: "Last research: 'RAG techniques'"Test 2: Preference detection
🔎 [mike] I want bullet points and academic sources on AI
# → Detects format=bullet_points, sources_priority=academic
# → The summary uses bullet points
Test 3: Isolated multi-user
👤 alice → researches quantum computing (session #1)
user bob → researches ML (session #1, knows nothing about alice)
user alice → profile → shows 1 session on quantum computing
Test 4: A user with a long history
for topic in "RAG" "Embeddings" "RAG" "Fine-tuning" "RAG"; do
python main.py --user mike "$topic"
done
python main.py --user mike "AI agents"
# → "Favorite topic: 'RAG' (3 times)."Common mistakes
1. Memory doesn't persist between runs
Cause: persist_store() was never called before exiting. A Ctrl+C can skip it.
Solution: Wrap the CLI in a try/finally:
try:
run_interactive()
finally:
persist_store(store)2. TypeError: got unexpected keyword argument 'store'
Cause: The @entrypoint doesn't have store= in its decorator.
Solution: Check the signature: @entrypoint(checkpointer=checkpointer, store=store) and the *, store: BaseStore parameter.
3. One user's data shows up in another's
Cause: The namespace doesn't include user_id.
Solution: Every access uses ("users", user_id, ...). Check that user_id comes from config["configurable"]["user_id"].
4. The persistence JSON grows without limit
Cause: Every session adds items and persist_store saves them all.
Solution: Implement retention:
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() returns None for user_id
Cause: You didn't pass user_id in the config when invoking.
Solution: Always include it:
config={"configurable": {"thread_id": thread_id, "user_id": "mike"}}6. Auto-detected preferences overwrite manual ones
Cause: detect_preferences_from_input runs on every invocation.
Solution: Add a manual flag and check it:
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 # Don't overwrite a manual preference
store.put(("users", uid, "preferences"), key, {"value": value, "manual": manual, ...})What's coming: Module 9 — Human-in-the-Loop
Your Research Assistant v3 has memory: it remembers preferences, accumulates history, personalizes every interaction. But it makes every decision on its own. If it decides to search an expensive source, it does so without asking.
Module 9 adds human-in-the-loop — the agent pauses and asks for approval before expensive actions. "I'm about to search 5 premium sources. Do you confirm?" This module's persistence is the prerequisite: the agent needs to save its state in order to pause indefinitely and resume when it gets your approval.
Project resources
- LangGraph Memory Store — Official long-term memory
- LangGraph Persistence — Checkpointing and MemorySaver
- LangGraph Cross-thread Memory — Memory across threads
- LangGraph Functional API — Store in @entrypoint
- InMemoryStore Reference — API reference
- LangGraph Human-in-the-Loop — Preview of Module 9
Module 8 — LangChain & LangGraph: From Chains to Agents