Long-Term Memory in LangGraph
Long-term memory enables conversational agents to maintain continuity, personal identity, and user context across multiple, independent conversation threads. In this chapter, we explore how LangGraph separates short-term session state from persistent cross-thread long-term stores, examine the base store architecture, implement semantic vector lookups, prevent memory duplication with Pydantic schemas, and transition from volatile prototypes to durable PostgreSQL storage engines.
1. The Core Problem: Why We Need Long-Term Memory
In traditional Large Language Model (LLM) applications, conversation state is tied to an isolated session or thread (often managed via short-term checkpointers). When a user closes a chat window or starts a new thread, the conversation context resets to zero.
Real-world user interactions do not happen in a single monolithic session. A user shares disparate fragments of their identity, ongoing projects, preferences, and worldviews across independent conversations over weeks or months:
Thread A: "Troubleshoot this Python script for an MCP server."
(Context: User codes in Python, builds MCP architecture)
Thread B: "Plan a 3-day itinerary for Mumbai."
(Context: User is traveling to Mumbai in two months)
Thread C: "Explain existentialism versus stoicism."
(Context: Philosophical leaning, concise tone preference)
If the agent only uses short-term memory, Thread B knows nothing about Thread A, and Thread C treats the user as a complete stranger.
The Solution: Cross-Thread Persistent Memory Stores
To build an agent that personalizes responses over time, we introduce a decoupled Memory Store:
When a user asks: "Write a Fibonacci series generator," an agent equipped with long-term memory queries its persistent store, retrieves the user's preference ("Prefers Python"), and writes the solution in Python without requiring explicit prompting.
2. LangGraph Memory Architecture: BaseStore and Its Subclasses
LangGraph manages long-term memory via the BaseStore abstract interface, keeping memory management decoupled from graph orchestration (StateGraph).
The BaseStore Interface
BaseStore defines standard CRUD and search operations:
put(namespace, key, value): Stores or updates a record at a specified path.get(namespace, key): Retrieves a specific record using its precise key.search(namespace, query=None, limit=10): Retrieves items under a namespace prefix, with optional semantic similarity search.delete(namespace, key): Removes an item from the store.
Built-in Store Implementations
3. Organizing Memory: Namespaces and Key-Value Operations
The Mental Model: Namespaces as File Paths
Conceptually: Think of a
BaseStoreas a hierarchical cloud drive (like Google Drive). A Namespace is a tuple of strings defining the directory path where a file resides:("users", user_id, folder_name).
By structuring namespaces hierarchically, you isolate user data while retaining the flexibility to query broadly or narrowly.
Basic CRUD Operations with InMemoryStore
The following implementation initializes an InMemoryStore, writes facts into specific user namespaces using put(), and reads them back using get() and prefix-level search().
from langgraph.store.memory import InMemoryStore
# 1. Instantiate the ephemeral memory store
store = InMemoryStore()
# 2. Define namespaces as tuples
ns_user_1 = ("users", "U1")
ns_user_2 = ("users", "U2")
# 3. Create memories (put: namespace, key, value)
store.put(ns_user_1, key="pref_food", value={"data": "User likes pizza"})
store.put(ns_user_1, key="pref_ui", value={"data": "User prefers dark mode"})
store.put(ns_user_2, key="pref_food", value={"data": "User likes pasta"})
store.put(ns_user_2, key="pref_ui", value={"data": "User prefers grid style navigation"})
# 4. Fetch a specific memory by exact key (get)
u1_food = store.get(ns_user_1, key="pref_food")
print("U1 Specific Fetch:", u1_food.value)
# Output: {'data': 'User likes pizza'}
# 5. Fetch all memories within a namespace (search)
u1_all = store.search(ns_user_1)
print(f"Total items for U1: {len(u1_all)}")
for item in u1_all:
print(f"Key: {item.key} | Value: {item.value}")
4. Scaling Retrieval: Semantic Search in Long-Term Stores
The Limits of Exact-Key and Namespace-Wide Retrieval
get(namespace, key)requires knowing the exact key ahead of time, which is impractical for an LLM parsing freeform user input.search(namespace)returns every record in that namespace. If a user profile contains 300 discrete memories, dumping all of them into the system prompt bloats token usage, drives up latency, and confuses the model with irrelevant facts.
Vector-Backed Semantic Retrieval
When conversational context matters (e.g., the user asks about travel, coding, or diet), the agent needs semantic search: finding stored facts whose semantic embeddings align with the current conversation.
Implementing Vector Indexing with InMemoryStore
To activate semantic similarity matching, configure the store with an index dictionary specifying an embedding model and vector dimension size:
from langchain_openai import OpenAIEmbeddings
from langgraph.store.memory import InMemoryStore
# 1. Initialize the embedding client
embeddings = OpenAIEmbeddings(model="text-embedding-3-small")
# 2. Configure the memory store with an indexing specification
store = InMemoryStore(
index={
"embed": embeddings,
"dims": 1536 # Native dimension size for text-embedding-3-small
}
)
ns = ("users", "U1")
# 3. Seed diverse profile facts
facts = [
("1", "User's name is Nitish."),
("2", "User teaches AI and Machine Learning on YouTube."),
("3", "User prefers concise answers over long explanations."),
("4", "User likes code examples written exclusively in Python."),
("5", "User is actively learning Machine Learning engineering."),
("6", "User prefers dark mode in applications."),
("7", "User prefers bullet points over long prose paragraphs.")
]
for key, fact in facts:
store.put(ns, key=key, value={"text": fact})
# 4. Perform semantic search matching on meaning, not syntax
results = store.search(
ns,
query="What subject is the user currently studying?",
limit=1
)
for match in results:
print(f"Top Semantic Match: {match.value['text']}")
# Output: User is actively learning Machine Learning engineering.
5. Integrating Long-Term Memory with LangGraph Agents
To expose the store to execution nodes, pass the store instance into .compile(store=...). LangGraph automatically injects the compiled store into any node declaring a store: BaseStore argument.
from typing import TypedDict, Annotated
from langchain_core.messages import SystemMessage, HumanMessage
from langchain_core.runnables import RunnableConfig
from langchain_openai import ChatOpenAI
from langgraph.graph import StateGraph, START, END
from langgraph.graph.message import add_messages
from langgraph.store.base import BaseStore
from langgraph.store.memory import InMemoryStore
# Graph State
class AgentState(TypedDict):
messages: Annotated[list, add_messages]
SYSTEM_PROMPT_TEMPLATE = """You are a helpful assistant with access to long-term memory.
Stored User Details:
{user_details}
Guidelines:
1. If the user's name is known, address them by name.
2. Tailor technical depth and language to their known preferences.
3. Suggest 3 relevant follow-up questions reflecting their background.
"""
def chat_node(state: AgentState, config: RunnableConfig, store: BaseStore):
# 1. Extract routing identifier from invocation configuration
user_id = config.get("configurable", {}).get("user_id", "default_user")
namespace = ("users", user_id, "details")
# 2. Retrieve memories from the persistent store
memories = store.search(namespace)
if memories:
details_str = "\n".join([f"- {m.value.get('text')}" for m in memories])
else:
details_str = "No prior history available."
# 3. Format system prompt with long-term context
formatted_system = SYSTEM_PROMPT_TEMPLATE.format(user_details=details_str)
system_message = SystemMessage(content=formatted_system)
# 4. Invoke LLM with injected memories prepended to conversation history
llm = ChatOpenAI(model="gpt-4o-mini", temperature=0.2)
response = llm.invoke([system_message] + state["messages"])
return {"messages": [response]}
# Build the Single-Node Graph
workflow = StateGraph(AgentState)
workflow.add_node("chat", chat_node)
workflow.add_edge(START, "chat")
workflow.add_edge("chat", END)
# Setup memory store with pre-populated records
shared_store = InMemoryStore()
shared_store.put(("users", "U1", "details"), key="1", value={"text": "User name: Nitish"})
shared_store.put(("users", "U1", "details"), key="2", value={"text": "Teaches AI on YouTube"})
shared_store.put(("users", "U1", "details"), key="3", value={"text": "Prefers Python code samples"})
# Compile graph with the shared store bound to its runtime
app = workflow.compile(store=shared_store)
# Execution test
inputs = {"messages": [HumanMessage(content="Explain Generative AI in simple terms.")]}
config = {"configurable": {"user_id": "U1"}}
result = app.invoke(inputs, config=config)
print(result["messages"][-1].content)
The model tailors its response directly to the stored facts: addressing the user as Nitish, using Python-oriented machine learning analogies, and suggesting classroom-oriented follow-up questions.
6. Real-Time Memory Creation and Extraction
An agent must also learn dynamically during conversation. To achieve this, we introduce an extraction node that analyzes user turns and saves meaningful personal information to the store.
The Pitfall: Memory Duplication
A naive extractor adds entries on every turn. Repeating a message creates redundant, conflicting records:
Over time, duplicate writes pollute the search index and degrade context quality.
The Solution: Pydantic-Enforced Deduplication
To fix this, pass existing memories alongside the incoming message to the extractor LLM. The model checks whether each extracted fact is truly new or already known:
import uuid
from typing import List
from pydantic import BaseModel, Field
from langchain_core.messages import SystemMessage
from langchain_openai import ChatOpenAI
from langgraph.store.base import BaseStore
from langchain_core.runnables import RunnableConfig
# 1. Fine-grained memory schema
class MemoryItem(BaseModel):
text: str = Field(description="Atomic, factual statement about user profile or preference.")
is_new: bool = Field(
description="True ONLY IF this fact is completely new and not present in existing memories. False if redundant."
)
class MemoryDecision(BaseModel):
should_write: bool = Field(description="Set to True if any memory requires persistence, else False.")
memories: List[MemoryItem] = Field(default_factory=list, description="Extracted memory items.")
# 2. Node performing deduplicated extraction
MEMORY_EXTRACTOR_PROMPT = """You manage user memories.
Current Stored Knowledge:
{existing_memories}
Task:
Analyze the user's latest message. Extract stable personal facts, identities, or preferences.
Do NOT record transient context (e.g., 'user said hello').
For each item, set `is_new=True` only if it represents information not captured in Current Stored Knowledge.
"""
def remember_node(state: dict, config: RunnableConfig, store: BaseStore):
user_id = config.get("configurable", {}).get("user_id", "default_user")
namespace = ("users", user_id, "details")
# Retrieve current records for comparison
existing = store.search(namespace)
existing_text = "\n".join([f"- {item.value['text']}" for item in existing]) if existing else "None"
last_user_message = state["messages"][-1].content
# Configure extractor with structured output
extractor_llm = ChatOpenAI(model="gpt-4o-mini", temperature=0).with_structured_output(MemoryDecision)
sys_prompt = SYSTEM_PROMPT_TEMPLATE.format(existing_memories=existing_text)
decision: MemoryDecision = extractor_llm.invoke([
SystemMessage(content=sys_prompt),
HumanMessage(content=last_user_message)
])
# Selective write: write only items tagged with is_new=True
if decision.should_write:
for item in decision.memories:
if item.is_new:
memory_id = str(uuid.uuid4())
store.put(namespace, key=memory_id, value={"text": item.text})
return state
7. The End-to-End Production Pattern: Read, Write, Deduplicate
Combining dynamic extraction with memory-conditioned responses yields a production-ready conversational architecture:
import uuid
from typing import TypedDict, Annotated, List
from pydantic import BaseModel, Field
from langchain_core.messages import SystemMessage, HumanMessage
from langchain_core.runnables import RunnableConfig
from langchain_openai import ChatOpenAI
from langgraph.graph import StateGraph, START, END
from langgraph.graph.message import add_messages
from langgraph.store.base import BaseStore
from langgraph.store.memory import InMemoryStore
# --- Schema Definitions ---
class MemoryItem(BaseModel):
text: str = Field(description="Factual, atomic piece of knowledge about the user.")
is_new: bool = Field(description="True if not present in current memories, False otherwise.")
class MemoryDecision(BaseModel):
should_write: bool = Field(description="True if any fact needs writing, False otherwise.")
memories: List[MemoryItem] = Field(default_factory=list)
class AgentState(TypedDict):
messages: Annotated[list, add_messages]
# --- Nodes ---
def remember_node(state: AgentState, config: RunnableConfig, store: BaseStore):
user_id = config.get("configurable", {}).get("user_id", "default_user")
namespace = ("users", user_id, "details")
existing = store.search(namespace)
existing_str = "\n".join([f"- {m.value['text']}" for m in existing]) if existing else "Empty"
user_msg = state["messages"][-1].content
extractor = ChatOpenAI(model="gpt-4o-mini", temperature=0).with_structured_output(MemoryDecision)
prompt = (
f"You are a memory extraction worker.\n"
f"Known User Memories:\n{existing_str}\n\n"
f"Extract stable facts from the latest turn. Tag novel items with is_new=True."
)
decision: MemoryDecision = extractor.invoke([
SystemMessage(content=prompt),
HumanMessage(content=user_msg)
])
if decision.should_write:
for mem in decision.memories:
if mem.is_new:
store.put(namespace, key=str(uuid.uuid4()), value={"text": mem.text})
return {}
def chat_node(state: AgentState, config: RunnableConfig, store: BaseStore):
user_id = config.get("configurable", {}).get("user_id", "default_user")
namespace = ("users", user_id, "details")
records = store.search(namespace)
memories_str = "\n".join([f"- {r.value['text']}" for r in records]) if records else "No records found."
prompt = (
f"You are a personalized assistant. Use the following user facts when answering:\n"
f"{memories_str}\n"
f"If the user's name is known, address them by name."
)
llm = ChatOpenAI(model="gpt-4o-mini", temperature=0.7)
response = llm.invoke([SystemMessage(content=prompt)] + state["messages"])
return {"messages": [response]}
# --- Graph Assembly ---
builder = StateGraph(AgentState)
builder.add_node("remember", remember_node)
builder.add_node("chat", chat_node)
builder.add_edge(START, "remember")
builder.add_edge("remember", "chat")
builder.add_edge("chat", END)
global_store = InMemoryStore()
agent = builder.compile(store=global_store)
8. Moving to Production: Durable Storage with PostgresStore
The Danger of Ephemeral Memory
InMemoryStore stores data in the running process's RAM. If the application server restarts, scales down, or crashes, all user memories are permanently lost.
To guarantee durability, enterprise systems require an external database like PostgreSQL.
PostgreSQL Local Container Setup
Run a local PostgreSQL container using Docker:
# 1. Verify Docker daemon is running
docker --version
# 2. Pull and start a PostgreSQL 16 instance
docker run --name langgraph-postgres \
-e POSTGRES_USER=postgres \
-e POSTGRES_PASSWORD=postgres \
-e POSTGRES_DB=memory_db \
-p 5432:5432 \
-d postgres:16
# 3. Confirm container status
docker ps --filter "name=langgraph-postgres"
Production Setup with PostgresStore
Install the required driver package:
pip install langgraph-checkpoint-postgres psycopg[binary,pool]
Use PostgresStore.from_conn_string() as a context manager to handle connection pooling and schema initialization automatically:
from langgraph.store.postgres import PostgresStore
# Standard connection URI
DB_URI = "postgresql://postgres:postgres@localhost:5432/memory_db"
# Establish a connection-pooled store context
with PostgresStore.from_conn_string(DB_URI) as store:
# 1. Run migrations to set up required schema tables
store.setup()
# 2. Compile the graph with PostgresStore
agent = builder.compile(store=store)
# 3. Invoke with user thread parameters
cfg = {"configurable": {"user_id": "U1"}}
# Turn 1: Introduce facts
msg1 = {"messages": [HumanMessage(content="Hello! My name is Nitish and I teach AI on YouTube.")]}
agent.invoke(msg1, config=cfg)
# Turn 2: Verify dynamic recall
msg2 = {"messages": [HumanMessage(content="Explain prompt injection attacks simply.")]}
res = agent.invoke(msg2, config=cfg)
print("Agent Response:\n", res["messages"][-1].content)
Proving Durability
If you terminate the Python process completely, reopen a new terminal, and run a standalone script querying the same database URI, store.search(("users", "U1", "details")) returns the stored facts intact. The data persists safely on disk in PostgreSQL.
9. Summary and Key Takeaways
- Short-Term vs. Long-Term Distinction: Short-term memory (checkpointers) handles state within a single thread. Long-term memory (
BaseStore) persists knowledge across independent threads and conversational sessions. - Hierarchical Namespacing: Namespaces act like filesystem directories (e.g.,
("users", user_id, "profile")), keeping user profiles partitioned while allowing scoped or global searches. - Three Core Retrieval Methods:
get()handles point lookups with known keys.search()lists entire namespaces.search(query="...", limit=k)performs vector-based semantic retrieval using embeddings.
- Preventing Memory Bloat: Unchecked memory extraction leads to repetitive, bloated stores. Pairing structured Pydantic schemas with an
is_newcheck ensures only novel facts are written. - Decoupled Node Architecture: A clean separation of concerns splits memory updates and generation into distinct graph nodes:
remember_nodehandles fact extraction, whilechat_nodefocuses on generating personalized answers. - Runtime Injection via
.compile(): Passingstore=...into graph compilation injects the store directly into node function signatures viastore: BaseStore. - Production Durability: While
InMemoryStoreis ideal for local development and unit tests, production deployments require durable backends likePostgresStore(orRedisStore) to ensure facts survive process recycles and host reboots.