Module 6: Memory Systems for Agents
8. Project: Research Agent v3 — Persistent Memory
Project Overview
In Module 5 you built Research Agent v2: planning with priorities, reflection against 5 criteria, conditional re-planning, and reasoning traces. It's an agent that thinks, researches, evaluates its own answer, and improves when it detects gaps. But it has a problem you discover the moment you use it for real: it remembers nothing.
If the research gets interrupted at step 7 of 12 (because your laptop went to sleep, the server restarted, you closed the terminal), you lose steps 1-6. If you've told it 5 times that you prefer reports in bullet points, it asks you again the sixth time. If you run two research sessions back to back, the second one has no context from the first.
In this project you turn v2 into Research Agent v3 by adding three memory systems:
-
Checkpointing with PostgresSaver: Every step persists in PostgreSQL. If the process dies, you pick up exactly where you left off — plan, partial data, reasoning trace, all intact.
-
Long-term memory: The agent remembers the user's preferences across different sessions. Output format, preferred sources, topics researched. They get injected automatically into the system prompt.
-
Conversation management: For long research sessions (30+ messages), it trims old messages and generates a summary to keep the context window manageable.
-
Time-travel: Inspect any past step of a research run and replay from a specific checkpoint for debugging.
Estimated time: 60-90 minutes (includes bringing up Docker for Postgres).
Project Goal
Extend Research Agent v2 (M5) with durable checkpointing, long-term memory for preferences, conversation management with trimming, and time-travel debugging.
By the end you'll be able to:
- Replace MemorySaver with PostgresSaver without changing the agent's logic
- Implement long-term memory with InMemoryStore for user preferences
- Implement trimming + summarization for long conversations
- Inspect the state history and replay from checkpoints
- Verify that a research run survives a process restart
What Changes vs v2 (M5)
Research Agent v2's foundation stays — planning, research, analysis, synthesis, reflection, re-planning. You don't rewrite those nodes. But you add three layers of memory.
New state fields
| Field | Type | What for |
|---|---|---|
user_id | str | Identifying the user for long-term memory |
conversation_summary | str | A summary of old, trimmed messages |
preferences_loaded | bool | Avoiding loading preferences more than once |
New components
| Component | Type | What it does |
|---|---|---|
PostgresSaver | Checkpointer | Persists state in PostgreSQL |
InMemoryStore | Store | Cross-session user preferences |
summarize_node | Node | Trims + summarizes messages when they exceed the limit |
time_travel_* | Functions | Checkpoint inspection and replay |
Changes to the flow
v2: planning → research ⇄ analysis → synthesis → reflection ─┐
▲ │
└──── replan ◄─────────────────────────────────┘
v3: [summarize] → planning → research ⇄ analysis → synthesis → reflection ─┐
▲ │ │
trim + resume │ │
loads prefs │
from InMemoryStore │
└──────────────── replan ◄─────────────────────────────┘
+ PostgresSaver (every step persists in PostgreSQL)
+ InMemoryStore (cross-session preferences)
+ time-travel (inspection and replay)
Technical Specifications
Stack
| Technology | Version | Use |
|---|---|---|
| Python | 3.11+ | Runtime |
| langchain, langchain-openai | v1.2+ | LLM, prompts |
| langgraph | v1.0+ | StateGraph, checkpointing, store |
| langgraph-checkpoint-postgres | latest | PostgresSaver |
| tavily-python | latest | Web search |
| Docker | any | Local PostgreSQL |
pip install langchain langchain-openai langgraph langgraph-checkpoint-postgres tavily-python python-dotenvDocker Compose for PostgreSQL
# docker-compose.yml
services:
postgres:
image: postgres:16
container_name: research_agent_postgres
environment:
POSTGRES_USER: agents
POSTGRES_PASSWORD: agents_secret
POSTGRES_DB: agents_db
ports:
- "5432:5432"
volumes:
- pgdata:/var/lib/postgresql/data
volumes:
pgdata:docker compose up -d
docker compose psEnvironment variables
# .env
OPENAI_API_KEY=sk-proj-your-api-key-here
TAVILY_API_KEY=tvly-your-api-key-here
DATABASE_URL=postgresql://agents:agents_secret@localhost:5432/agents_dbv3 Architecture
quality >= threshold
┌───────────────────────────────────┐
│ ▼
[START] → [SUMMARIZE] → [PLANNING] → [RESEARCH] ⇄ [ANALYSIS] → [SYNTHESIS] → [REFLECTION] → [END]
│ │ ▲ score<0.7 │ │
trim + resume │ │ & iter<max │ │ quality < threshold
│ └──────────────┘ │ & revisions < max
loads preferences │
from InMemoryStore │
└──────────────── [RE-PLAN] ◄───────────────────────┘
┌─────────────────────────────────────────────────────────────────────────┐
│ PERSISTENCE │
│ PostgresSaver InMemoryStore Time-Travel │
│ (checkpoints/step) (cross-session prefs) (get_state_history/replay) │
└─────────────────────────────────────────────────────────────────────────┘
Three levels of memory:
-
Checkpointing (PostgresSaver): Every node generates a checkpoint in Postgres. If the process dies between research and analysis, on restart it continues from research with everything intact.
-
Long-term (InMemoryStore): Preferences stored under a
user_idnamespace. They get loaded at the start of every run and injected into the planning node's system prompt. -
Conversation management (summarize_node): Before planning, it checks whether the history exceeds a limit. If it does, it trims the old messages and generates a compact summary.
Step 1: Add PostgresSaver
The simplest and most impactful change. The same API as MemorySaver, durable persistence.
from dotenv import load_dotenv
load_dotenv()
import os
from langgraph.checkpoint.postgres import PostgresSaver
DB_URI = os.environ["DATABASE_URL"]
def create_checkpointer():
"""PostgresSaver if there's a DB, MemorySaver if not."""
if os.environ.get("DATABASE_URL"):
checkpointer = PostgresSaver.from_conn_string(DB_URI)
checkpointer.setup()
return checkpointer
from langgraph.checkpoint.memory import MemorySaver
return MemorySaver()In v2 you compiled without a checkpointer. In v3:
checkpointer = create_checkpointer()
agent_v3 = build_research_agent_v3().compile(
checkpointer=checkpointer,
store=store,
)Every conversation uses a thread_id that isolates its state:
config = {"configurable": {"thread_id": "research-session-001"}}
result = agent_v3.invoke(initial_state, config)If the process gets interrupted, you invoke again with the same thread_id and the agent continues from the last checkpoint — PostgresSaver recovers it automatically.
Step 2: Implement Long-term Memory
PostgresSaver solves "don't lose my work." Long-term memory solves "remember who I am."
The store and preference functions
from langgraph.store.memory import InMemoryStore
store = InMemoryStore()
def save_user_preferences(st, user_id: str, preferences: dict):
existing = load_user_preferences(st, user_id)
merged = {**existing, **preferences}
st.put(
namespace=(user_id, "preferences"),
key="current",
value={"data": merged},
)
return merged
def load_user_preferences(st, user_id: str) -> dict:
try:
item = st.get(namespace=(user_id, "preferences"), key="current")
if item and item.value:
return item.value.get("data", {})
except Exception:
pass
return {}
def save_research_history(st, user_id: str, query: str, summary: str):
from datetime import datetime, timezone
history = load_research_history(st, user_id)
history.append({
"query": query,
"summary": summary[:500],
"timestamp": datetime.now(timezone.utc).isoformat(),
})
if len(history) > 20:
history = history[-20:]
st.put(
namespace=(user_id, "research_history"),
key="all",
value={"entries": history},
)
def load_research_history(st, user_id: str) -> list:
try:
item = st.get(namespace=(user_id, "research_history"), key="all")
if item and item.value:
return item.value.get("entries", [])
except Exception:
pass
return []Injecting preferences into the planning node
v2's planning doesn't know the user. In v3, it loads preferences and history:
def planning_node(state: dict, store) -> dict:
"""Generate a research plan — with the user's preferences."""
query = ""
for msg in reversed(state["messages"]):
if isinstance(msg, HumanMessage):
query = msg.content
break
user_id = state.get("user_id", "default")
preferences = load_user_preferences(store, user_id)
past_research = load_research_history(store, user_id)
reflection_notes = state.get("reflection_notes", [])
context = _build_context_prefix(state)
pref_context = ""
if preferences:
pref_context = (
"\nUSER PREFERENCES:\n"
f"- Format: {preferences.get('output_format', 'not specified')}\n"
f"- Sources: {preferences.get('preferred_sources', 'not specified')}\n"
f"- Detail: {preferences.get('detail_level', 'not specified')}\n"
)
history_context = ""
if past_research:
topics = [r["query"][:80] for r in past_research[-3:]]
history_context = (
f"\nPREVIOUS TOPICS: {json.dumps(topics, ensure_ascii=False)}\n"
"Avoid repeating research that's already been done.\n"
)
replan_context = ""
if reflection_notes:
replan_context = (
f"\nRE-PLANNING:\n"
f"Notes: {json.dumps(reflection_notes, ensure_ascii=False)}\n"
"Generate DIFFERENT sub-questions that cover the gaps.\n"
)
response = model.invoke([
SystemMessage(content=(
f"{context}"
"You are an expert researcher. Break the question into "
"2-5 sub-questions with priorities.\n"
f"{pref_context}{history_context}{replan_context}\n"
"Answer with ONLY JSON:\n"
'{"sub_questions": ["q1","q2"], "priority_order": [0,1], '
'"success_criteria": "...", "reasoning": "..."}'
)),
HumanMessage(content=query),
])
parsed = _parse_json(response.content)
sub_qs = parsed.get("sub_questions", [query])
prio = parsed.get("priority_order", list(range(len(sub_qs))))
criteria = parsed.get("success_criteria", "A complete answer")
plan = {
"main_query": query,
"sub_questions": sub_qs,
"completed_questions": state.get("plan", {}).get("completed_questions", []),
"priority_order": prio,
"success_criteria": criteria,
}
text = f"Plan: {query}\nCriterion: {criteria}\n"
for idx in prio:
if idx < len(sub_qs):
text += f" [P{prio.index(idx)+1}] {sub_qs[idx]}\n"
return {
"plan": plan,
"preferences_loaded": True,
"messages": [AIMessage(content=text)],
"reasoning_trace": _trace(
state, "planning",
"re-plan" if reflection_notes else "initial_plan",
parsed.get("reasoning", ""),
preferences_loaded=bool(preferences),
),
}The planning_node(state, store) signature is the key: LangGraph detects the store parameter and injects it automatically. The nodes that don't need the store keep their original (state) -> dict signature.
Detecting preferences after the research run
def extract_and_save_preferences(st, user_id, messages):
keywords = {
"bullet points": {"output_format": "bullet_points"},
"markdown format": {"output_format": "markdown"},
"short answers": {"detail_level": "concise"},
"detailed answers": {"detail_level": "detailed"},
"academic sources": {"preferred_sources": "academic"},
"official sources": {"preferred_sources": "official"},
}
for msg in messages:
if isinstance(msg, HumanMessage):
lower = msg.content.lower()
for kw, pref in keywords.items():
if kw in lower:
save_user_preferences(st, user_id, pref)A simple keyword-based detector. In production you'd use an LLM, but for the project this demonstrates the pattern without adding latency.
Step 3: Conversation Management
Long research sessions generate a lot of messages. A run with 3 iterations + reflection + re-plan generates ~20 messages. Two runs back to back: 40. The context window fills up and the cost grows.
The summarization node
It runs at the start, before planning. If there are too many messages, it trims the old ones and generates a summary.
from langchain_core.messages import RemoveMessage
MAX_MESSAGES = 20
SUMMARY_THRESHOLD = 15
def summarize_node(state: dict) -> dict:
messages = state.get("messages", [])
existing_summary = state.get("conversation_summary", "")
if len(messages) <= SUMMARY_THRESHOLD:
return {}
messages_to_summarize = messages[:-10]
summary_input = ""
if existing_summary:
summary_input = f"PREVIOUS SUMMARY: {existing_summary}\n\n"
summary_input += "MESSAGES TO SUMMARIZE:\n"
for msg in messages_to_summarize:
role = "User" if isinstance(msg, HumanMessage) else "Agent"
summary_input += f"[{role}]: {msg.content[:200]}\n"
response = model.invoke([
SystemMessage(content=(
"Summarize the conversation in 3-5 sentences. Capture:\n"
"- What was researched and what was found\n"
"- What the user prefers\n"
"- Where the research currently stands\n"
"Be concise."
)),
HumanMessage(content=summary_input),
])
delete_msgs = [RemoveMessage(id=m.id) for m in messages_to_summarize if m.id]
return {
"conversation_summary": response.content,
"messages": delete_msgs,
"reasoning_trace": _trace(
state, "summarize", "trim_and_summarize",
f"Trimmed {len(messages_to_summarize)} msgs",
),
}RemoveMessage is a special LangGraph type. When a node returns RemoveMessage(id=msg.id) in the messages list, the add_messages reducer removes that message from the state. It's declarative trimming.
Injecting the summary as context
The nodes that invoke the model include the summary to maintain coherence:
def _build_context_prefix(state: dict) -> str:
summary = state.get("conversation_summary", "")
if not summary:
return ""
return (
f"PREVIOUS CONTEXT:\n{summary}\n\n"
"Use this context. Don't repeat research that's already been done.\n\n"
)Step 4: Time-travel Capabilities
With PostgresSaver, every step generates a checkpoint. Time-travel lets you navigate those checkpoints for debugging and replay.
Inspecting the history
def inspect_investigation(agent, thread_id: str):
config = {"configurable": {"thread_id": thread_id}}
history = list(agent.get_state_history(config))
print(f"\n{'='*60}")
print(f" RESEARCH RUN: {thread_id}")
print(f" Total checkpoints: {len(history)}")
print(f"{'='*60}")
for i, snapshot in enumerate(reversed(history)):
values = snapshot.values
node = snapshot.metadata.get("source", "unknown")
step = snapshot.metadata.get("step", "?")
score = values.get("quality_score", 0.0)
data_count = len(values.get("research_data", []))
print(f"\n [{step}] {node} — score={score:.2f}, data={data_count}")
trace = values.get("reasoning_trace", [])
if trace:
last = trace[-1]
print(f" trace: {last.get('node', '?')} → {last.get('decision', '?')}")
print(f"\n{'='*60}")
return historyReplaying from a checkpoint
If the research took a wrong turn, you can go back to a point and continue differently:
def replay_from_checkpoint(agent, thread_id: str, checkpoint_index: int):
"""checkpoint_index: 0 = most recent, 1 = the one before, etc."""
config = {"configurable": {"thread_id": thread_id}}
history = list(agent.get_state_history(config))
if checkpoint_index >= len(history):
print(f"There are only {len(history)} checkpoints.")
return None, None
target = history[checkpoint_index]
print(f" Replay from step {target.metadata.get('step', '?')}")
print(f" Node: {target.metadata.get('source', '?')}")
return target.values, target.config
def fork_investigation(agent, thread_id: str, new_thread_id: str, checkpoint_index: int):
"""Create a new research run from an existing checkpoint."""
state, _ = replay_from_checkpoint(agent, thread_id, checkpoint_index)
if state is None:
return None
new_config = {"configurable": {"thread_id": new_thread_id}}
agent.update_state(new_config, state)
print(f" Fork created: {new_thread_id} ← {thread_id}[{checkpoint_index}]")
return new_config
def compare_investigations(agent, thread_ids: list[str]):
print(f"\n{'='*60}")
print(f" COMPARISON")
print(f"{'='*60}")
for tid in thread_ids:
state = agent.get_state({"configurable": {"thread_id": tid}})
if state.values:
v = state.values
print(f"\n {tid}")
print(f" Score: {v.get('quality_score', 0.0):.2f}")
print(f" Iters: {v.get('iteration_count', 0)}, "
f"Re-plans: {v.get('plan_revisions', 0)}")
print(f" Data: {len(v.get('research_data', []))}")
print(f"\n{'='*60}")Step 5: Update the Graph
Extend the state
from typing import Annotated, TypedDict
from langchain_core.messages import BaseMessage
from langgraph.graph.message import add_messages
class ResearchPlan(TypedDict):
main_query: str
sub_questions: list[str]
completed_questions: list[str]
priority_order: list[int]
success_criteria: str
class AgentState(TypedDict):
# v1/v2 fields
messages: Annotated[list[BaseMessage], add_messages]
plan: ResearchPlan
iteration_count: int
max_iterations: int
research_data: list[str]
quality_score: float
final_answer: str
metadata: dict
reflection_notes: list[str]
plan_revisions: int
max_plan_revisions: int
reasoning_trace: list[dict]
# v3: Memory Systems
user_id: str
conversation_summary: str
preferences_loaded: boolBuild the graph
from langgraph.graph import StateGraph, START, END
def should_continue_research(state: dict) -> str:
if state.get("quality_score", 0.0) >= 0.7:
return "synthesis"
if state.get("iteration_count", 0) >= state.get("max_iterations", 3):
return "synthesis"
return "research"
def should_replan_or_finish(state: dict) -> str:
if state.get("quality_score", 0.0) >= 0.7:
return "end"
if state.get("plan_revisions", 0) < state.get("max_plan_revisions", 2):
return "replan"
return "end"
def build_research_agent_v3():
graph = StateGraph(AgentState)
graph.add_node("summarize", summarize_node)
graph.add_node("planning", planning_node)
graph.add_node("research", research_node)
graph.add_node("analysis", analysis_node)
graph.add_node("synthesis", synthesis_node)
graph.add_node("reflection", reflection_node)
graph.add_node("replan", replan_node)
graph.add_edge(START, "summarize")
graph.add_edge("summarize", "planning")
graph.add_edge("planning", "research")
graph.add_edge("research", "analysis")
graph.add_edge("synthesis", "reflection")
graph.add_edge("replan", "research")
graph.add_conditional_edges(
"analysis", should_continue_research,
{"research": "research", "synthesis": "synthesis"},
)
graph.add_conditional_edges(
"reflection", should_replan_or_finish,
{"end": END, "replan": "replan"},
)
return graphThe new summarize node gets inserted between START and planning. v2's nodes (research, analysis, synthesis, reflection, replan) don't change — only planning gets the store as a second argument.
The Complete Research Agent v3
The code from steps 1-5 is everything you need. v2's nodes (research_node, analysis_node, synthesis_node, reflection_node, replan_node) stay identical — copy them from the M5 project. The _parse_json and _trace helpers stay too.
What's new in v3:
create_checkpointer()— PostgresSaver or MemorySaver depending on the environmentsave/load_user_preferencesandsave/load_research_history— long-term memorysummarize_node— trimming + summarization- The updated
planning_node— acceptsstore, loads preferences and history inspect_investigation,replay_from_checkpoint,fork_investigation— time-travel
To run it end-to-end:
from dotenv import load_dotenv
load_dotenv()
import os
import json
from datetime import datetime, timezone
from langchain.chat_models import init_chat_model
from langchain_core.messages import AIMessage, HumanMessage, SystemMessage
from langchain_community.tools.tavily_search import TavilySearchResults
from langgraph.checkpoint.postgres import PostgresSaver
from langgraph.store.memory import InMemoryStore
model = init_chat_model("openai:gpt-4.1-mini")
search_tool = TavilySearchResults(max_results=3)
store = InMemoryStore()
DB_URI = os.environ["DATABASE_URL"]
# ... (include: AgentState, ResearchPlan, _parse_json, _trace,
# _build_context_prefix, every memory function,
# every node, build_research_agent_v3) ...
with PostgresSaver.from_conn_string(DB_URI) as checkpointer:
checkpointer.setup()
agent_v3 = build_research_agent_v3().compile(
checkpointer=checkpointer,
store=store,
)
def run_research_v3(
query: str,
user_id: str = "default",
thread_id: str | None = None,
max_iterations: int = 3,
max_plan_revisions: int = 2,
verbose: bool = True,
) -> dict:
if thread_id is None:
thread_id = f"research-{datetime.now().strftime('%Y%m%d-%H%M%S')}"
config = {"configurable": {"thread_id": thread_id}}
# Detect whether there's previous state (resume)
existing = agent_v3.get_state(config)
resuming = (existing.values
and existing.values.get("plan", {}).get("main_query"))
if resuming:
if verbose:
score = existing.values.get("quality_score", 0.0)
data = len(existing.values.get("research_data", []))
print(f"\n Resuming: {thread_id} (score={score:.2f}, data={data})")
result = agent_v3.invoke(
{"messages": [HumanMessage(content=query)]},
config,
)
else:
initial_state = {
"messages": [HumanMessage(content=query)],
"plan": {
"main_query": "", "sub_questions": [],
"completed_questions": [], "priority_order": [],
"success_criteria": "",
},
"iteration_count": 0,
"max_iterations": max_iterations,
"research_data": [],
"quality_score": 0.0,
"final_answer": "",
"metadata": {},
"reflection_notes": [],
"plan_revisions": 0,
"max_plan_revisions": max_plan_revisions,
"reasoning_trace": [],
"user_id": user_id,
"conversation_summary": "",
"preferences_loaded": False,
}
if verbose:
print(f"\n{'='*60}")
print(f" Research Agent v3")
print(f" Query: {query}")
print(f" User: {user_id} | Thread: {thread_id}")
print(f"{'='*60}")
result = agent_v3.invoke(initial_state, config)
# Post-run: save to long-term memory
save_research_history(store, user_id, query, result.get("final_answer", "")[:500])
extract_and_save_preferences(store, user_id, result.get("messages", []))
if verbose:
score = result.get("quality_score", 0.0)
prefs = load_user_preferences(store, user_id)
history = load_research_history(store, user_id)
print(f"\n Score: {score:.2f} | Thread: {thread_id}")
print(f" Preferences: {list(prefs.keys()) if prefs else 'none'}")
print(f" Research runs: {len(history)}")
return result
# ── Execution ────────────────────────────────────────────
if __name__ == "__main__":
save_user_preferences(store, "mike", {
"output_format": "bullet_points",
"preferred_sources": "official",
"detail_level": "detailed",
})
result = run_research_v3(
query="What are the trends in AI agents and how do LangGraph and CrewAI compare?",
user_id="mike",
thread_id="demo-v3-001",
)
print(f"\n{result.get('final_answer', 'No answer')[:500]}")
inspect_investigation(agent_v3, "demo-v3-001")Expected output (schematic)
============================================================
Research Agent v3
Query: What are the trends in AI agents...?
User: mike | Thread: demo-v3-001
============================================================
Plan: 3 sub-questions (trends, LangGraph, CrewAI)
Preferences loaded: [output_format, preferred_sources, detail_level]
Research → Analysis (iter 1): score=0.45
Research → Analysis (iter 2): score=0.72 → Synthesis
Reflection: score=0.68 → Re-plan #1
Research → Analysis (iter 1): score=0.78 → Synthesis
Reflection: score=0.82 → APPROVED
Score: 0.82 | Thread: demo-v3-001
Preferences: ['output_format', 'preferred_sources', 'detail_level']
Research runs: 1
RESEARCH RUN: demo-v3-001
Total checkpoints: 14
[1] summarize — score=0.00, data=0
[2] planning — score=0.00, data=0
...
[14] reflection — score=0.82, data=9
============================================================
Recommended Tests
Test 1: The research survives a restart (checkpointing)
# Session 1: run the research
result = run_research_v3(
query="What is the Model Context Protocol?",
user_id="test", thread_id="persist-001",
)
# CLOSE the Python process (Ctrl+C or exit())
# Session 2: new terminal, re-import everything, re-create agent_v3
config = {"configurable": {"thread_id": "persist-001"}}
state = agent_v3.get_state(config)
print(f"Plan: {state.values.get('plan', {}).get('main_query', 'EMPTY')}")
print(f"Data: {len(state.values.get('research_data', []))}")What it validates: The state should show the plan and data — recovered from PostgreSQL after the restart.
Test 2: Cross-session memory (long-term)
save_user_preferences(store, "carlos", {
"output_format": "bullet_points",
"preferred_sources": "academic",
})
result1 = run_research_v3(query="What is RAG?", user_id="carlos", thread_id="s-a")
result2 = run_research_v3(query="How does fine-tuning work?", user_id="carlos", thread_id="s-b")
# In session-b, planning should load the preferences from session-a
prefs = load_user_preferences(store, "carlos")
history = load_research_history(store, "carlos")
print(f"Prefs: {prefs}")
print(f"Research runs: {[h['query'] for h in history]}")What it validates: In session-b, planning loads session-a's preferences. The history shows both research runs.
Test 3: A long conversation (trimming)
thread = "long-convo-test"
queries = ["What is LangGraph?", "How does it compare to CrewAI?",
"What are the advantages for production?", "Give me more detail"]
for q in queries:
run_research_v3(query=q, user_id="trim-test", thread_id=thread)
state = agent_v3.get_state({"configurable": {"thread_id": thread}})
msgs = len(state.values.get("messages", []))
has_summary = bool(state.values.get("conversation_summary"))
print(f"Msgs: {msgs} | Summary: {has_summary}")What it validates: The message count stabilizes. When it passes SUMMARY_THRESHOLD, a summary appears.
Test 4: Time-travel and debugging
result = run_research_v3(query="The best frameworks for AI agents",
user_id="debug", thread_id="debug-001")
history = inspect_investigation(agent_v3, "debug-001")
fork_investigation(agent_v3, "debug-001", "debug-001-fork", checkpoint_index=3)
compare_investigations(agent_v3, ["debug-001", "debug-001-fork"])What it validates: inspect_investigation shows every step. fork_investigation creates a branch. compare_investigations shows the differences.
Success Criteria
-
The research survives a restart. Close Python, open a new session, query the thread_id — the plan, data, and score are all intact.
-
The agent remembers preferences between sessions. Two research runs with the same
user_id, a differentthread_id. The second loads the first one's preferences. -
Long conversations stay within budget. After 4+ invocations, the messages don't exceed
MAX_MESSAGES. -
You can inspect and replay past steps.
inspect_investigationshows the history.fork_investigationcreates branches. -
The trace includes the v3 nodes.
summarizeshows up when it triggers. Planning records whether it loaded preferences.
Checklist
- Docker Compose brings up PostgreSQL
-
create_checkpointer()returns a PostgresSaver - The agent compiles with
checkpointerandstore -
run_research_v3acceptsuser_idandthread_id - Invoking with an existing
thread_idresumes the research -
save/load_user_preferencesworks with InMemoryStore -
planning_node(state, store)loads preferences and injects them into the prompt -
save_research_historyrecords past research runs -
summarize_nodetrims when messages > SUMMARY_THRESHOLD -
RemoveMessageremoves messages correctly -
inspect_investigationshows the complete history -
fork_investigationcreates a branch from a checkpoint - The state survives a process restart
- Preferences persist across the same user's threads
- A long conversation doesn't exceed MAX_MESSAGES
Common Errors
Error 1: PostgreSQL isn't running
Symptom: psycopg.OperationalError: connection refused when compiling.
Cause: Docker Compose isn't up, or port 5432 is taken.
Solution: docker compose up -d && docker compose ps. If the port is taken, change it to 5433:5432 in docker-compose.yml and update the connection string.
Error 2: The agent doesn't resume interrupted research
Symptom: You invoke with the same thread_id after a restart but it starts from zero.
Cause: You pass initial_state instead of letting the checkpointer recover the state, or the thread_id doesn't match exactly.
Solution: run_research_v3 checks for existing state with agent.get_state(config). If there's state with a plan, it invokes with just the new message. Verify the thread_id is the same string.
Error 3: Preferences don't load in planning
Symptom: Planning doesn't mention preferences even though you saved them.
Cause: planning_node doesn't have store as its second parameter, or the user_id doesn't match.
Solution: The signature must be planning_node(state, store). LangGraph injects the store automatically. Verify that user_id in the initial_state matches the one in save_user_preferences.
Error 4: RemoveMessage doesn't remove messages
Symptom: After summarize_node, the count keeps growing.
Cause: The messages have no id. RemoveMessage needs an id to know what to remove.
Solution: LangGraph assigns IDs automatically when messages pass through add_messages. If you create messages manually without going through the state, they have no ID. Check with [m.id for m in state["messages"]].
Error 5: The store loses data across restarts
Symptom: The preferences disappear when you restart the process.
Cause: InMemoryStore lives in RAM — it's lost on restart, just like MemorySaver.
Solution: This is expected. InMemoryStore demonstrates the pattern; in production you'd use a durable backend. For this project, pre-load the preferences at the start of the script. M10 covers persisting the store.
Error 6: Time-travel shows checkpoints with partial values
Symptom: inspect_investigation lists checkpoints but some values are empty.
Cause: The initial_state doesn't include defaults for every AgentState field.
Solution: Verify that initial_state in run_research_v3 has every field. LangGraph persists the complete state in each checkpoint, but if a field was never initialized, it shows up as None.
Connection to M7
Your Research Agent v3 remembers — checkpoints last across restarts, preferences persist across sessions, conversations stay manageable. But the tools are fixed: Tavily for web search, and nothing else. If you need to search GitHub, read an API's documentation, or query a database, you have to add each tool manually.
In Module 7 (MCP — Model Context Protocol):
- MCP servers expose tools as standardized services. A GitHub server exposes
search_repos,read_file. A Postgres server exposesquery,list_tables. - MCP clients connect the agent to servers dynamically — the agent discovers the available tools at runtime.
- Research Agent v4 connects to MCP servers to expand its capabilities without modifying the graph.
The transition: "Your agent remembers between sessions (M6) → now give it access to an ecosystem of external tools via MCP (M7)." v3's memory is especially useful with MCP — the agent can remember which servers gave good results and prioritize them.
Resources
- LangGraph Persistence Docs — Checkpointing, backends, configuration
- PostgresSaver Reference — The complete API
- LangGraph Store (Memory) — InMemoryStore, BaseStore, long-term
- LangGraph Time Travel — get_state_history, update_state, replay
- LangGraph Message Trimming — RemoveMessage, trimming
- Docker Compose for PostgreSQL — The official image
- psycopg 3 Documentation — The PostgreSQL driver