Module 6: Memory Systems for Agents
5. Long-term Memory: Cross-Session
Overview
So far, every kind of memory you've implemented is tied to one conversation. Checkpointing with MemorySaver or PostgresSaver persists a thread's state — messages, research plan, partial results. But when the user opens a new conversation (a new thread_id), the agent starts from zero. It doesn't know who the user is, it doesn't remember they prefer bullet points, it has no idea that last week it already researched "RAG patterns" and doesn't need to explain the basics. Every thread is an island.
Long-term memory solves this. It's memory that persists between different conversations — across threads, across sessions, across days or weeks. When the user opens a new conversation, the agent already knows things: "this user prefers academic sources, works in fintech, and last time researched vector databases." It isn't the chat history (that's short-term). It isn't a thread's checkpoint (that's durable execution). It's accumulated knowledge that transcends any individual conversation.
LangGraph implements long-term memory through its Store API: InMemoryStore and BaseStore. It's a key-value store with namespaces that lives separately from the checkpointer. In this capsule you'll implement user preferences, episodic memory, and the complete integration with your agent.
Checkpointing vs Long-term Memory
Two systems for two different problems
This is the most common point of confusion. Checkpointing and long-term memory aren't the same thing — they solve different problems, use different APIs, and get stored separately.
Checkpointing (capsules 03-04) saves the complete state of one conversation:
Thread "research-001"
├── Checkpoint 1: [user: "Research RAG"] → state with a plan
├── Checkpoint 2: [assistant: "I found 3 papers"] → state with results
└── Checkpoint 3: [user: "Give me more detail"] → state with the follow-up
If the user comes back with thread_id: "research-001", the agent picks up where it left off. But if they open thread_id: "research-002", it knows nothing about the previous thread.
Long-term memory saves knowledge that crosses conversations:
User "user-456" (across ALL threads)
├── Preferences: {format: "bullet_points", sources: "academic"}
├── History: [{topic: "RAG", date: "2025-01-15"}, {topic: "vector DBs", date: "2025-01-22"}]
└── Context: {company: "fintech", role: "ML engineer"}
Think of it this way. Checkpointing is like a hotel room: when you step out, your luggage is still there. But when you check out and come back on another trip, they give you a new room — cleaned. Long-term memory is the hotel's customer profile: "This guest prefers high floors, an extra pillow, and late checkout." It doesn't matter how many times they come or which room they're assigned.
| Aspect | Checkpointing | Long-term Memory |
|---|---|---|
| Scope | One thread | All of a user's threads |
| What it saves | The complete state (messages, plan) | Accumulated knowledge (preferences, facts) |
| API | BaseCheckpointSaver | BaseStore (InMemoryStore) |
| Key | thread_id | namespace + key |
| When it's written | Automatically on every step | Explicitly from your code |
| Passed to | graph.compile(checkpointer=...) | graph.compile(store=...) |
Both get passed to compile() as separate parameters:
agent = graph.compile(
checkpointer=checkpointer, # per-thread state
store=store # cross-thread knowledge
)
The LangGraph Store API
InMemoryStore: the starting point
InMemoryStore is to the Store what MemorySaver is to the Checkpointer: an in-RAM implementation, perfect for development, lost on restart.
from langgraph.store.memory import InMemoryStore
store = InMemoryStore()
Namespaces: hierarchical organization
Data gets organized in namespaces — tuples of strings that work like directories:
("users", "user-456", "preferences") # the user's preferences
("users", "user-456", "research_history") # research history
("users", "user-456", "profile") # professional context
The hierarchical structure lets you search at different levels. You can search everything for the user ("users", "user-456") or just their preferences ("users", "user-456", "preferences").
The fundamental operations: put, get, search, delete
put — Save or update an item:
store.put(
namespace=("users", "user-456", "preferences"),
key="output_format",
value={"format": "bullet_points", "max_length": "concise"}
)
Each item has a namespace (where it lives), a key (a unique identifier within the namespace), and a value (a dictionary). If the key already exists, it gets overwritten.
get — Retrieve a specific item:
item = store.get(namespace=("users", "user-456", "preferences"), key="output_format")
if item:
print(item.value) # {"format": "bullet_points", "max_length": "concise"}
print(item.key) # "output_format"
print(item.created_at) # creation timestamp
It returns None if the item doesn't exist. Always check before accessing .value.
search — Find items by namespace prefix:
items = store.search(namespace_prefix=("users", "user-456", "preferences"))
for item in items:
print(f"{item.key}: {item.value}")
It returns every item whose namespace starts with the given prefix.
delete — Remove an item:
store.delete(namespace=("users", "user-456", "preferences"), key="output_format")
A complete example
from langgraph.store.memory import InMemoryStore
store = InMemoryStore()
ns = ("users", "user-456")
store.put((*ns, "preferences"), "output_format", {"format": "bullet_points", "language": "en"})
store.put((*ns, "preferences"), "sources", {"preferred": "academic", "avoid": ["medium_blogs"]})
store.put((*ns, "profile"), "context", {"company": "fintech startup", "role": "ML engineer"})
print(f"Preferences: {len(store.search((*ns, 'preferences')))} items")
print(f"Everything for the user: {len(store.search(ns))} items")
Implementing User Preferences
Which preferences are worth remembering
| Category | Examples | Impact |
|---|---|---|
| Output format | Bullet points vs paragraphs, length, language | How it formats every answer |
| Preferred sources | Academic vs blogs, specific domains | Which results it prioritizes |
| Level of detail | Technical vs executive, with/without code | The level of abstraction |
| Topics researched | RAG, vector DBs, fine-tuning | Avoids repeating the basics |
| Professional context | Industry, role, tech stack | Personalizes the examples |
Saving preferences with a tool
The cleanest pattern is a tool the agent calls when it detects a preference:
from langchain_core.tools import tool
@tool
def save_user_preference(category: str, key: str, value: str) -> str:
"""Save a user preference for future conversations.
Args:
category: Category (output_format, sources, detail_level, profile)
key: The preference's name
value: The preference's value
"""
return f"Preference saved: {category}/{key} = {value}"
In the node that processes tool calls, you write to the store:
def process_tool_calls(state: State, config: RunnableConfig, *, store: BaseStore):
user_id = config["configurable"].get("user_id", "anonymous")
results = []
for tc in state["messages"][-1].tool_calls:
if tc["name"] == "save_user_preference":
args = tc["args"]
store.put(("users", user_id, "preferences"), args["key"],
{"category": args["category"], "value": args["value"]})
results.append({"role": "tool", "content": f"Saved: {args['key']}", "tool_call_id": tc["id"]})
return {"messages": results}
The flow in action
Conversation 1 (thread: "abc-001")
User: "Research RAG. Give me the info in bullet points, academic sources only."
Agent: [detects → save_user_preference("output_format", "style", "bullet_points")]
Agent: [detects → save_user_preference("sources", "type", "academic_only")]
Agent: "• RAG (Retrieval-Augmented Generation) combines..."
Conversation 2 (thread: "abc-002") — days later
User: "Research fine-tuning of LLMs"
Agent: [reads the store → bullet_points, academic_only]
Agent: "• Fine-tuning lets you adapt..." ← already uses the preferred format
Implementing Episodic Memory
What episodic memory is
Preferences are static. Episodic memory records past events and experiences: topics researched, key findings, successful strategies, accumulated context.
Saving and querying episodes
from datetime import datetime
from langgraph.store.base import BaseStore
def save_research_episode(store: BaseStore, user_id: str, topic: str,
summary: str, key_findings: list[str]):
namespace = ("users", user_id, "research_history")
episode_key = f"research_{datetime.now().strftime('%Y%m%d_%H%M%S')}"
store.put(namespace, episode_key, {
"topic": topic, "summary": summary, "key_findings": key_findings,
"timestamp": datetime.now().isoformat(),
"depth": "deep" if len(key_findings) > 3 else "surface"
})
def get_research_context(store: BaseStore, user_id: str) -> str:
episodes = store.search(("users", user_id, "research_history"))
if not episodes:
return "No previous research on record."
recent = sorted(episodes, key=lambda e: e.value.get("timestamp", ""), reverse=True)[:10]
parts = ["The user's previous research:"]
for ep in recent:
v = ep.value
parts.append(f"- [{v.get('timestamp', '')[:10]}] {v['topic']}: {v['summary'][:100]}")
return "\n".join(parts)
The agent that remembers
With episodic memory, when the user asks "Explain how to combine RAG with fine-tuning", the agent checks the history, sees that both topics were already researched, and gets right to the point: "Since you already explored RAG and fine-tuning separately, I'll go straight to how they combine." Without it, it would start with "RAG is Retrieval-Augmented Generation..." — redundant information.
Integration with the Agent
The complete pattern: an agent with long-term memory
Any node can receive the store as an injected parameter. The key signature: store: BaseStore after the * (a keyword-only argument). LangGraph automatically injects the instance passed to compile().
from typing import TypedDict, Annotated
from langgraph.graph import StateGraph, START, END
from langgraph.graph.message import add_messages
from langgraph.store.memory import InMemoryStore
from langgraph.store.base import BaseStore
from langgraph.checkpoint.memory import MemorySaver
from langchain.chat_models import init_chat_model
from langchain_core.runnables import RunnableConfig
from langchain_core.messages import SystemMessage
model = init_chat_model("openai:gpt-4.1-mini")
class State(TypedDict):
messages: Annotated[list, add_messages]
def build_memory_context(store: BaseStore, user_id: str) -> str:
sections = []
prefs = store.search(("users", user_id, "preferences"))
if prefs:
sections.append("## User preferences")
sections.extend(f"- {i.key}: {i.value.get('value', i.value)}" for i in prefs)
history = store.search(("users", user_id, "research_history"))
if history:
recent = sorted(history, key=lambda h: h.value.get("timestamp", ""), reverse=True)[:5]
sections.append("\n## Recent research")
sections.extend(f"- {e.value['topic']} ({e.value.get('timestamp', '')[:10]})" for e in recent)
return "\n".join(sections) if sections else "No previous information about the user."
def chat_node(state: State, config: RunnableConfig, *, store: BaseStore):
user_id = config["configurable"].get("user_id", "anonymous")
memory_context = build_memory_context(store, user_id)
system = SystemMessage(content=f"""You are a research assistant with persistent memory.
You know this information about the user (from previous conversations):
{memory_context}
Use this information to personalize your answers.
Don't repeat explanations of topics already researched. Respect their format and source preferences.""")
response = model.invoke([system] + list(state["messages"]))
return {"messages": [response]}
graph = StateGraph(State)
graph.add_node("chat", chat_node)
graph.add_edge(START, "chat")
graph.add_edge("chat", END)
store = InMemoryStore()
agent = graph.compile(checkpointer=MemorySaver(), store=store)
Invoking with a user_id
The key: pass a user_id in the config, separate from the thread_id:
config = {
"configurable": {
"thread_id": "session-2025-03-08-001", # changes with each conversation
"user_id": "user-456" # constant for the user
}
}
result = agent.invoke(
{"messages": [("user", "Research function calling patterns")]},
config
)
The thread_id changes with each conversation. The user_id stays constant — it connects the conversations to each other.
When to Remember and When to Forget
The infinite memory problem
Every item you inject into the system prompt burns tokens. If a user has 200 episodes and 50 preferences, injecting everything is expensive and ineffective (the model loses focus with too much context).
Strategy 1: Relevance filtering
Select only what's relevant to the current conversation:
def get_relevant_memories(store: BaseStore, user_id: str, current_topic: str, max_items: int = 5):
all_episodes = store.search(("users", user_id, "research_history"))
topic_lower = current_topic.lower()
relevant = [
ep for ep in all_episodes
if topic_lower in ep.value.get("topic", "").lower()
or topic_lower in " ".join(ep.value.get("key_findings", [])).lower()
]
relevant.sort(key=lambda e: e.value.get("timestamp", ""), reverse=True)
return relevant[:max_items]
Strategy 2: TTL (Time-to-Live)
Old episodes lose relevance:
from datetime import datetime, timedelta
def cleanup_old_memories(store: BaseStore, user_id: str, max_age_days: int = 90):
namespace = ("users", user_id, "research_history")
episodes = store.search(namespace)
cutoff = datetime.now() - timedelta(days=max_age_days)
deleted = 0
for ep in episodes:
ts = ep.value.get("timestamp", "")
if ts and datetime.fromisoformat(ts) < cutoff:
store.delete(namespace, ep.key)
deleted += 1
return deleted
Strategy 3: Consolidation
Instead of deleting, consolidate. Multiple episodes on the same topic get compressed into one summary using the LLM: you find every episode for the topic, pass them to the model with a consolidation prompt, delete the originals, and save a single one of type "consolidated". This preserves the knowledge without eating so many items.
Prioritizing by type
| Type | Lifetime | Cleanup |
|---|---|---|
| Format preferences | Permanent | Only if the user changes them |
| Professional profile | Permanent | Manual update |
| Topics researched | 90 days | TTL + consolidation |
| Specific findings | 30 days | TTL |
| Temporal context | 7 days | Aggressive TTL |
A token budget for memory
A model with a 128k context window:
├── Base system prompt: ~500 tokens
├── Long-term memory context: ~500-1000 tokens ← TARGET
├── Conversation history: ~2000-4000 tokens
├── Tool results: variable
└── Room for the answer: ~2000 tokens
500-1000 tokens of long-term memory = ~10 preferences + ~5 summarized episodes. Enough for effective personalization.
Connection to the Project
In this module's project (capsule 08, Research Agent with Persistent Memory):
- InMemoryStore gets added to the Research Agent to save preferences and episodes. Every completed research run gets recorded with a topic, summary, and key findings.
- The preferences get injected into the system prompt at the start of every conversation. The agent respects format, sources, and level of detail.
- Episodic memory lets the agent say "you already researched RAG last week — want me to go deeper on a specific aspect?" instead of starting from scratch.
- The
user_idgets passed in the config alongside thethread_id, connecting all of the same user's conversations.
In later modules, M8 (Multi-Agent) uses stores shared between sub-agents so the researcher and the writer know the same preferences, and M10 (Production) tackles persisting the store beyond InMemoryStore.
Troubleshooting
Problem 1: TypeError: unexpected keyword argument 'store'
Symptom: When running the graph, the node fails because it receives store but doesn't expect it.
Cause: The node's signature doesn't include store: BaseStore as a keyword-only argument (after the *).
Solution:
# Incorrect
def my_node(state: State, config: RunnableConfig, store: BaseStore): ...
# Correct — store after the *
def my_node(state: State, config: RunnableConfig, *, store: BaseStore): ...
Problem 2: store.get() returns None for an item I just saved
Symptom: A put followed by a get returns None.
Cause: The namespace doesn't match exactly. ("users", "123") and ("users", "123", "preferences") are different. get matches exactly, not by prefix.
Solution:
store.put(("users", "123", "preferences"), "format", {"style": "bullets"})
store.get(("users", "123"), "format") # None ← different namespace
store.get(("users", "123", "preferences"), "format") # Item ← correct
store.search(("users", "123")) # [Item] ← search by prefix
Problem 3: KeyError: 'user_id' in the config
Symptom: config["configurable"]["user_id"] fails because there's only a thread_id.
Solution: Add user_id to the config and use .get() with a fallback:
config = {"configurable": {"thread_id": "session-001", "user_id": "user-456"}}
# In the node:
user_id = config["configurable"].get("user_id", "anonymous")
Problem 4: Too much memory eats the context window
Symptom: Poor answers when the user has a lot of history.
Cause: You inject all of the long-term memory without filtering.
Solution: Cap the injection:
def build_memory_context(store, user_id, max_prefs=10, max_episodes=5):
prefs = store.search(("users", user_id, "preferences"))[:max_prefs]
episodes = store.search(("users", user_id, "research_history"))
episodes = sorted(episodes, key=lambda e: e.value.get("timestamp", ""), reverse=True)
return episodes[:max_episodes]
Problem 5: InMemoryStore doesn't persist across restarts
Symptom: You restart the process and all the long-term memory disappears.
Cause: InMemoryStore lives in RAM — it's lost on restart, just like MemorySaver.
Solution: For development, that's acceptable. For production, you need a store with persistence (LangGraph Cloud offers persistent stores). As a workaround, you can serialize to disk:
import json
def save_store_snapshot(store: InMemoryStore, path: str):
data = {}
for item in store.search(()):
ns_key = "/".join(item.namespace)
data.setdefault(ns_key, {})[item.key] = item.value
with open(path, "w") as f:
json.dump(data, f, indent=2, default=str)
Exercises
Exercise 1: A basic store with preferences (Easy)
Create an InMemoryStore, save 3 preferences for a user (output format, source type, language), and retrieve them with search. Print each preference with its key and value.
See solution
from langgraph.store.memory import InMemoryStore
store = InMemoryStore()
ns = ("users", "user-456", "preferences")
store.put(ns, "output_format", {
"value": "bullet_points",
"description": "Answers in bullet-point format"
})
store.put(ns, "source_type", {
"value": "academic",
"description": "Papers and official documentation over blogs"
})
store.put(ns, "language", {
"value": "en",
"description": "Answers in English"
})
prefs = store.search(ns)
print(f"Preferences: {len(prefs)} items\n")
for item in prefs:
print(f" {item.key}: {item.value['value']} — {item.value['description']}")
specific = store.get(ns, "output_format")
print(f"\nDirect lookup: {specific.key} = {specific.value['value']}")
You should see the 3 preferences listed and the specific lookup of output_format.
Exercise 2: Episodic memory with history (Easy)
Simulate 3 completed research runs: save each one as an episode with a topic, summary, key_findings, and timestamp. Query the history sorted by date (most recent first).
See solution
from langgraph.store.memory import InMemoryStore
from datetime import datetime, timedelta
store = InMemoryStore()
ns = ("users", "user-456", "research_history")
data = [
("research_0115", "RAG patterns", "RAG architectures: naive, sentence-window", 45),
("research_0122", "Vector databases", "Comparing Pinecone, Weaviate, Chroma", 38),
("research_0305", "Fine-tuning LLMs", "Techniques: LoRA, QLoRA, full", 3),
]
for key, topic, summary, age in data:
store.put(ns, key, {
"topic": topic, "summary": summary,
"key_findings": [f"Key finding about {topic}"],
"timestamp": (datetime.now() - timedelta(days=age)).isoformat()
})
sorted_eps = sorted(store.search(ns), key=lambda e: e.value["timestamp"], reverse=True)
for ep in sorted_eps:
v = ep.value
print(f"[{v['timestamp'][:10]}] {v['topic']}: {v['summary']}")
Exercise 3: An agent with long-term memory integrated (Medium)
Build a LangGraph graph with a chat node that reads preferences from the store and injects them into the system prompt. Compile it with InMemoryStore and MemorySaver. Pre-load preferences. Invoke it with two different thread_ids and verify the preferences apply in both conversations.
See solution
from typing import TypedDict, Annotated
from langgraph.graph import StateGraph, START, END
from langgraph.graph.message import add_messages
from langgraph.store.memory import InMemoryStore
from langgraph.store.base import BaseStore
from langgraph.checkpoint.memory import MemorySaver
from langchain.chat_models import init_chat_model
from langchain_core.runnables import RunnableConfig
from langchain_core.messages import SystemMessage
model = init_chat_model("openai:gpt-4.1-mini")
class State(TypedDict):
messages: Annotated[list, add_messages]
def chat_node(state: State, config: RunnableConfig, *, store: BaseStore):
user_id = config["configurable"].get("user_id", "anonymous")
prefs = store.search(("users", user_id, "preferences"))
prefs_text = "\n".join(
f"- {item.key}: {item.value.get('value', '')}" for item in prefs
) if prefs else "No preferences on record."
system = SystemMessage(content=f"""You are a research assistant.
The user's preferences:
{prefs_text}
IMPORTANT: Respect these preferences in ALL your answers.""")
response = model.invoke([system] + list(state["messages"]))
return {"messages": [response]}
graph_builder = StateGraph(State)
graph_builder.add_node("chat", chat_node)
graph_builder.add_edge(START, "chat")
graph_builder.add_edge("chat", END)
store = InMemoryStore()
ns = ("users", "user-456", "preferences")
store.put(ns, "output_format", {"value": "bullet_points — always use bullets"})
store.put(ns, "sources", {"value": "academic sources only (arxiv, papers)"})
agent = graph_builder.compile(checkpointer=MemorySaver(), store=store)
config_a = {"configurable": {"thread_id": "session-A", "user_id": "user-456"}}
result_a = agent.invoke({"messages": [("user", "What is RAG?")]}, config_a)
print("=== Conversation A ===")
print(result_a["messages"][-1].content[:400])
config_b = {"configurable": {"thread_id": "session-B", "user_id": "user-456"}}
result_b = agent.invoke({"messages": [("user", "Compare LoRA vs QLoRA")]}, config_b)
print("\n=== Conversation B ===")
print(result_b["messages"][-1].content[:400])
Both conversations should use bullet points and prefer academic sources.
Exercise 4: Cleanup with TTL (Medium)
Create a store with 10 episodes (some from 120 days ago, others recent). Implement cleanup_old_episodes(store, user_id, max_age_days) that deletes the old ones. Show the count before and after.
See solution
from langgraph.store.memory import InMemoryStore
from datetime import datetime, timedelta
store = InMemoryStore()
user_id = "user-456"
ns = ("users", user_id, "research_history")
topics = ["RAG", "Fine-tuning", "RAG", "Embeddings", "RAG",
"Agents", "Fine-tuning", "MCP", "RAG", "Agents"]
ages = [120, 100, 80, 60, 45, 30, 20, 10, 5, 2]
for i, (topic, age) in enumerate(zip(topics, ages)):
store.put(ns, f"ep_{i:03d}", {
"topic": topic,
"summary": f"Research on {topic}",
"timestamp": (datetime.now() - timedelta(days=age)).isoformat()
})
def cleanup_old_episodes(store, user_id, max_age_days=90):
ns = ("users", user_id, "research_history")
episodes = store.search(ns)
cutoff = datetime.now() - timedelta(days=max_age_days)
deleted = 0
for ep in episodes:
ts = ep.value.get("timestamp", "")
if ts and datetime.fromisoformat(ts) < cutoff:
store.delete(ns, ep.key)
deleted += 1
return deleted
before = len(store.search(ns))
deleted = cleanup_old_episodes(store, user_id, max_age_days=90)
after = len(store.search(ns))
print(f"Before: {before} episodes")
print(f"Deleted: {deleted} (>90 days)")
print(f"After: {after} episodes")
for ep in store.search(ns):
print(f" {ep.key}: {ep.value['topic']} ({ep.value['timestamp'][:10]})")
The episodes at 120 and 100 days get deleted. 8 of 10 remain.
Exercise 5: A complete system with auto-detected preferences (Hard)
Build an agent that: (a) detects the user's implicit preferences, (b) saves them to the store using a tool, (c) applies them automatically in future conversations. The first conversation establishes the preferences, the second uses them without the user mentioning them.
See solution
from typing import TypedDict, Annotated
from langgraph.graph import StateGraph, START, END
from langgraph.graph.message import add_messages
from langgraph.store.memory import InMemoryStore
from langgraph.store.base import BaseStore
from langgraph.checkpoint.memory import MemorySaver
from langchain.chat_models import init_chat_model
from langchain_core.tools import tool
from langchain_core.runnables import RunnableConfig
from langchain_core.messages import SystemMessage, ToolMessage
model = init_chat_model("openai:gpt-4.1-mini")
@tool
def save_preference(category: str, key: str, value: str) -> str:
"""Save a user preference for future conversations."""
return f"Preference '{key}' saved."
model_with_tools = model.bind_tools([save_preference])
class State(TypedDict):
messages: Annotated[list, add_messages]
def chat_node(state: State, config: RunnableConfig, *, store: BaseStore):
user_id = config["configurable"].get("user_id", "anonymous")
prefs = store.search(("users", user_id, "preferences"))
prefs_text = "\n".join(f"- {i.key}: {i.value.get('value', '')}" for i in prefs) or "None."
system = SystemMessage(content=f"""You are a research assistant with memory.
Known preferences: {prefs_text}
If the user expresses a preference, use save_preference. ALWAYS respect the saved ones.""")
return {"messages": [model_with_tools.invoke([system] + list(state["messages"]))]}
def tool_node(state: State, config: RunnableConfig, *, store: BaseStore):
user_id = config["configurable"].get("user_id", "anonymous")
results = []
for tc in state["messages"][-1].tool_calls:
args = tc["args"]
store.put(("users", user_id, "preferences"), args["key"],
{"category": args["category"], "value": args["value"]})
results.append(ToolMessage(content=f"Saved: {args['key']}", tool_call_id=tc["id"]))
return {"messages": results}
def should_continue(state: State):
last = state["messages"][-1]
return "tools" if hasattr(last, "tool_calls") and last.tool_calls else END
graph = StateGraph(State)
graph.add_node("chat", chat_node)
graph.add_node("tools", tool_node)
graph.add_edge(START, "chat")
graph.add_conditional_edges("chat", should_continue, {"tools": "tools", END: END})
graph.add_edge("tools", "chat")
store = InMemoryStore()
agent = graph.compile(checkpointer=MemorySaver(), store=store)
r1 = agent.invoke(
{"messages": [("user", "Explain RAG to me. I want bullet points and academic papers only.")]},
{"configurable": {"thread_id": "c-001", "user_id": "user-789"}}
)
print("Conv 1:", r1["messages"][-1].content[:200])
r2 = agent.invoke(
{"messages": [("user", "Now explain fine-tuning")]},
{"configurable": {"thread_id": "c-002", "user_id": "user-789"}}
)
print("Conv 2:", r2["messages"][-1].content[:200])
In conversation 2, the agent uses bullet points and academic sources automatically.
Summary
In this capsule you learned:
-
Checkpointing and long-term memory are distinct systems. The checkpointer saves one conversation's state (a thread). The store saves knowledge that crosses every conversation (cross-thread). Both get passed to
graph.compile()separately. -
The LangGraph Store API works with namespaces and key-value pairs.
InMemoryStoregives you four operations:put,get,search,delete. Namespaces are tuples that organize data hierarchically — by user, by type, by category. -
User preferences apply cross-session. Output format, preferred sources, level of detail. They get saved when the agent detects them and injected into the system prompt at the start of every new conversation.
-
Episodic memory records past experiences. Topics researched, key findings, successful strategies. It lets the agent avoid repetition and build on previous research.
-
The integration happens via injection into nodes. Add
*, store: BaseStoreto the node's signature, and LangGraph injects the store automatically. Theuser_idin the config connects the same user's conversations. -
Memory needs active management. Filter by relevance, apply a TTL for old episodes, consolidate duplicate memories, and respect a token budget. More memory isn't better memory.
Next capsule: Time-Travel Debugging — how to navigate an agent's state history, replay from checkpoints, and use the combination of checkpointing + store for advanced debugging.
Additional Resources
- LangGraph Memory Concepts — Official memory documentation: short-term, long-term, the Store API
- LangGraph Store Guide — How-to for shared state and cross-thread memory
- BaseStore API Reference — The complete reference: InMemoryStore, put, get, search, delete
- LangGraph Persistence — How checkpointing and the store work together in persistence
- Building Agents with Memory (LangChain Blog) — Memory patterns for agents: preferences, episodic, semantic