Implementing Short-Term Memory in LangGraph

Large Language Models (LLMs) are fundamentally stateless; every API invocation operates in complete isolation with zero native retention of past interactions. Building continuous, multi-turn conversational agents requires orchestrating an external state management layer known as Short-Term Memory (STM).

This guide covers implementing short-term memory in LangGraph, transitioning from ephemeral in-memory checkpoints to production-grade persistent databases, and resolving the critical Context Window Overflow Problem using trimming, explicit deletion, and dynamic summarization.

1. The Core Mental Model: Stateless LLMs and Short-Term Memory

The Stateless Nature of LLMs

When you execute an LLM call (e.g., model.invoke(...)), the model evaluates the incoming list of tokens, generates an output, and immediately purges the execution context from its active memory.

If a user sends:

Turn 1: "Hi, my name is Nitish." $\rightarrow$ Model replies: "Hello Nitish, nice to meet you!"
Turn 2: "What is my name?" $\rightarrow$ Model replies: "I am an AI, I do not know your name."

Structure & Concept Flow

Turn 1:

User: "Hi, my name is Nitish" ---> [LLM] ---> "Hello Nitish!"

Context Purged Immediately

Turn 2:

User: "What is my name?" ---> [LLM] ---> "I do not know your name."

Because Turn 2 sends only the prompt "What is my name?", the model has no historical context to draw from.

The Conversation Buffer (Short-Term Memory)

To simulate continuous memory, the engineering harness must maintain a Conversation Buffer. Every subsequent prompt must concatenate and pass the entire conversational history along with the newest user message.

CODE

Turn 1:
User: "Hi, my name is Nitish" 
Buffer State: [User: "Hi, my name is Nitish"]
Invocation Payload: [User: "Hi, my name is Nitish"]
Response: "Hello Nitish!"
Buffer State Updated: [User: "Hi...", AI: "Hello Nitish!"]

Turn 2:
User: "What is my name?"
Buffer State Updated: [User: "Hi...", AI: "Hello...", User: "What is my name?"]
Invocation Payload: Entire Buffer State (3 messages)
Response: "Your name is Nitish!"

In LangGraph, this mechanism is natively integrated into graph workflows via Checkpointers and Threads.

2. LangGraph Memory Architecture: Checkpointers & Threads

LangGraph implements short-term memory by persisting graph state transitions across execution intervals. Two core primitives enable this architecture:

Checkpointer (BaseCheckpointSaver): A persistence adapter that intercepts execution at every Superstep (node transition boundary) and serializes the complete graph state (StateSnapshot).
Thread ID (thread_id): A unique partition key passed within the invocation configuration (RunnableConfig). It groups isolated conversational sessions under distinct execution threads.

Structure & Concept Flow

User Invocation (Message + thread_id: "thread-1")

LangGraph Engine

|-- 1. Read past StateSnapshot from Checkpointer for "thread-1"

|-- 2. Append new message to State["messages"]

|-- 3. Execute Graph Nodes (e.g., Call Model)

|-- 4. Write updated StateSnapshot back

to Checkpointer for "thread-1"

Return Output to User

3. Ephemeral In-Memory Implementation (`MemorySaver`)

The simplest way to maintain conversational history within a running session is using MemorySaver, which stores state transitions inside the local process memory (RAM).

Implementation Code

PYTHON

from dotenv import load_dotenv
from langchain_openai import ChatOpenAI
from langgraph.graph import StateGraph, START, END, MessagesState
from langgraph.checkpoint.memory import MemorySaver

load_dotenv()

# Initialize Model
model = ChatOpenAI(model="gpt-4o-mini")

# Define Graph Node
def call_model(state: MessagesState):
    response = model.invoke(state["messages"])
    return {"messages": [response]}

# Build Graph
builder = StateGraph(MessagesState)
builder.add_node("call_model", call_model)
builder.add_edge(START, "call_model")
builder.add_edge("call_model", END)

# Attach Checkpointer during Compilation
checkpointer = MemorySaver()
app = builder.compile(checkpointer=checkpointer)

Multi-Turn Execution Across Distinct Threads

When invoking the compiled graph, pass the thread_id inside the configurable dictionary:

PYTHON

# Session 1: Thread "chat-1"
config_1 = {"configurable": {"thread_id": "chat-1"}}

# Turn 1
app.invoke({"messages": [("user", "Hi, my name is Nitish.")]}, config=config_1)

# Turn 2
response = app.invoke({"messages": [("user", "What is my name?")]}, config=config_1)
print(response["messages"][-1].content)
# Output: "Your name is Nitish."

# Inspecting State
snapshot = app.get_state(config_1)
print(f"Total messages in thread 1: {len(snapshot.values['messages'])}")
# Output: 4 (User, AI, User, AI)

PYTHON

# Session 2: Thread "chat-2" (Isolated Memory Space)
config_2 = {"configurable": {"thread_id": "chat-2"}}

response_isolated = app.invoke({"messages": [("user", "What is my name?")]}, config=config_2)
print(response_isolated["messages"][-1].content)
# Output: "I am sorry, but I do not know your name as you haven't told me yet."

4. Production-Grade Persistent Memory with PostgreSQL

The Volatility Problem

While MemorySaver works for rapid local testing, it is fundamentally non-viable for production. If the application server restarts, crashes, or scales across distributed container instances:

The entire process RAM clears.
All stored checkpoints and conversational threads are lost permanently.
Calling app.get_state(config) post-restart yields an empty state.

To guarantee state survival across application lifecycles, state must be offloaded to a persistent, durable database. LangGraph natively supports PostgreSQL via PostgresSaver.

Structure & Concept Flow

Application Crash / Restart

+---> MemorySaver: RAM wipes out ---> Checkpoints Lost (State = Null)

+---> PostgresSaver: Checkpoints persisted on disk ---> State Restored

5. Setting Up Dockerized PostgreSQL for LangGraph

To avoid native OS-level dependency and service installation friction, deploy PostgreSQL inside a Docker container.

Step 1: Docker Compose Configuration (`docker-compose.yml`)

Create a local docker-compose.yml file configuring the PostgreSQL engine:

YAML

version: '3.8'

services:
  postgres:
    image: postgres:16-alpine
    container_name: langgraph_postgres
    restart: always
    environment:
      POSTGRES_USER: postgres_user
      POSTGRES_PASSWORD: postgres_password
      POSTGRES_DB: langgraph_db
    ports:
      - "5432:5432"
    volumes:
      - postgres_data:/var/lib/postgresql/data

volumes:
  postgres_data:

Step 2: Spin Up the Container

Run the daemonized Docker command in the root folder containing the compose file:

BASH

# Start PostgreSQL container in detached mode
docker compose up -d

# Verify that the container is actively healthy
docker ps

Step 3: Install Required Python Dependencies

Install the core dependencies alongside the official PostgreSQL checkpointer integration:

BASH

pip install langgraph langgraph-checkpoint-postgres psycopg[binary,pool] langchain-openai

6. Implementing Persistent Graph State with `PostgresSaver`

In production, interact with PostgresSaver using Python context managers (with statements) to maintain pooling connections safely. When configuring the database for the first time, run checkpointer.setup() to initialize the required checkpoint tables.

PYTHON

from dotenv import load_dotenv
from langchain_openai import ChatOpenAI
from langgraph.graph import StateGraph, START, END, MessagesState
from langgraph.checkpoint.postgres import PostgresSaver

load_dotenv()

# Database Connection URI
DB_URI = "postgresql://postgres_user:postgres_password@localhost:5432/langgraph_db?sslmode=disable"

# LLM Node Definition
model = ChatOpenAI(model="gpt-4o-mini")

def call_model(state: MessagesState):
    response = model.invoke(state["messages"])
    return {"messages": [response]}

# Initialize Graph
builder = StateGraph(MessagesState)
builder.add_node("call_model", call_model)
builder.add_edge(START, "call_model")
builder.add_edge("call_model", END)

# Context Manager for Connection Management
with PostgresSaver.from_conn_string(DB_URI) as checkpointer:
    # Auto-creates checkpoint schemas/tables on initial run
    checkpointer.setup()
    
    app = builder.compile(checkpointer=checkpointer)
    
    config = {"configurable": {"thread_id": "persistent-thread-1"}}
    
    # Send Interaction
    app.invoke({"messages": [("user", "Hi, my name is Nitish and I live in India.")]}, config=config)
    app.invoke({"messages": [("user", "What is my location?")]}, config=config)

Proving Persistence Across Process Termination

Even if you exit the Python interpreter, terminate the script, or reboot your physical system, the state persists inside PostgreSQL:

PYTHON

# Separate execution script / Post-restart session
from langgraph.checkpoint.postgres import PostgresSaver

DB_URI = "postgresql://postgres_user:postgres_password@localhost:5432/langgraph_db?sslmode=disable"

with PostgresSaver.from_conn_string(DB_URI) as checkpointer:
    # Notice: We are simply querying state without calling app.invoke()
    config = {"configurable": {"thread_id": "persistent-thread-1"}}
    saved_state = checkpointer.get(config)
    
    # State is retrieved intact directly from disk tables
    historical_messages = saved_state.checkpoint["channel_values"]["messages"]
    for msg in historical_messages:
        print(f"{msg.type.upper()}: {msg.content}")

7. The Context Window Overflow Problem

While appending every message to a state buffer guarantees infinite memory access in theory, it introduces a severe hardware and algorithmic barrier: the Context Window Overflow Problem.

The Mechanism of Failure

Every model has an upper limit on input/output tokens known as its Context Window ( $W_{\text{ctx}}$ ):

Let $T_{\text{in}}$ be total input tokens and $T_{\text{ctx}}$ be the context limit.
If $T_{\text{in}} \ge T_{\text{ctx}}$ , the invocation triggers an immediate API rate/limit error or causes model outputs to degrade drastically into hallucinations.

\text{Total Ingested Tokens } T = \sum_{i=1}^{k} \text{tokens}(m_i) + \text{tokens}(m_{\text{new}})

As conversations reach hundreds of turns ( $k \to \infty$ ), passing historical sequences through model.invoke() hits three scaling walls:

Context Limit Exceeded: Hard crash via provider errors.
Exponential Token Costs: You are billed on every single turn for re-reading ancient messages repeatedly.
Latency Explosion: Generation delay scales directly with payload length.

8. Mitigation Strategy 1: Message Trimming

Message Trimming discards older message history right before invoking the LLM, keeping the token total within a bounded threshold ( $T_{\text{max}}$ ).

Crucial Distinction: Trimming modifies the payload sent to the LLM invocation; it does not delete messages from your checkpointer or persistent graph state.

Structure & Concept Flow

Full State in Storage:

[Msg 1] -> [Msg 2] -> [Msg 3] -> [Msg 4] -> [Msg 5]

\__________________________/

Trimmed Payload Sent to LLM:

[Msg 3, Msg 4, Msg 5]

Implementation with `trim_messages`

LangChain provides the trim_messages utility to enforce maximum token boundaries.

PYTHON

from langchain_core.messages import trim_messages
from langchain_openai import ChatOpenAI
from langgraph.graph import StateGraph, START, END, MessagesState
from langgraph.checkpoint.memory import MemorySaver

model = ChatOpenAI(model="gpt-4o-mini")

def call_model(state: MessagesState):
    # Enforce an upper bound of 150 tokens before invocation
    selected_messages = trim_messages(
        state["messages"],
        max_tokens=150,
        strategy="last",          # Retain the most recent messages
        token_counter=model,       # Pass model instance to accurately tokenize
        include_system=True,      # Retain instructions if present
        allow_partial=False,      # Do not split individual messages mid-sentence
    )
    
    response = model.invoke(selected_messages)
    return {"messages": [response]}

builder = StateGraph(MessagesState)
builder.add_node("call_model", call_model)
builder.add_edge(START, "call_model")
builder.add_edge("call_model", END)

app = builder.compile(checkpointer=MemorySaver())

Behavioral Failure Mode of Trimming

Trimming relies on the assumption that only the last $N$ messages matter. If vital context—such as user credentials, baseline project requirements, or names—was established at Turn 1, trimming drops those tokens once the conversation grows long enough:

Structure & Concept Flow

Turn 1: User says name is Nitish (State tokens: 40) -> Model knows name.

Turn 2: Complex technical discussions (State tokens: 120) -> Model tracks context.

Turn 3: Detailed explanations (State tokens: 200) -> TRIMMING TRIGGERS.

Turn 4: "What is my name?"

Trimmed Payload: Only Turn 3 messages passed.

Model Output: "I'm sorry, I don't know your name."

9. Mitigation Strategy 2: Explicit Message Deletion

To purge invalid, expired, or unwanted state payloads permanently from the database checkpointer, LangGraph provides the RemoveMessage primitive.

Mechanism

Appending a RemoveMessage(id=target_message_id) to the messages channel instructs the checkpointer to evict the corresponding message from state.

PYTHON

from langchain_core.messages import RemoveMessage
from langgraph.graph import StateGraph, START, END, MessagesState

def clean_up_node(state: MessagesState):
    messages = state["messages"]
    
    # If messages exceed 10 turns, permanently delete the oldest 6
    if len(messages) > 10:
        messages_to_remove = messages[:6]
        return {"messages": [RemoveMessage(id=m.id) for m in messages_to_remove]}
    
    return {"messages": []}

This permanently purges early messages from both active memory and checkpointer tables, preventing unbounded state growth.

10. Mitigation Strategy 3: Dynamic Summarization with Selective Deletion

The most effective, production-standard solution to the Context Window Overflow Problem is Dynamic Summarization accompanied by Deletion.

Core Architecture

Instead of discarding historical messages completely (trimming) or purging them without a trace (simple deletion), summarization uses an auxiliary LLM call to condense expired dialogue into a high-density summary string.

Structure & Concept Flow

Step 1: Checkpoint contains 8 raw messages.

Step 2: Threshold exceeded (Length > 6).

Step 3: Extract older 6 messages -> Send to Model -> Produce condensed summary string.

Step 4: State updates:

- state["summary"] = updated_summary_string

- state["messages"] = Purge the 6 summarized messages using RemoveMessage

Step 5: Only the 2 most recent messages + the single summary string remain in state.

Structure & Concept Flow

Incoming Turn -> Chat Node (Prepends state['summary'] as SystemMessage)

Messages Count > 6?

/ \

[YES] / \ [NO]

v v

Summarize Node END

(Compress & Purge)

END

11. Complete Code Walkthrough: Summarization Workflow

Below is the complete implementation of a production-ready summarization and garbage-collection pipeline in LangGraph.

Step 1: Define Custom State Structure

Because standard MessagesState only tracks a list of messages, we define an extended state to hold the running summary string.

PYTHON

from typing import Annotated, Literal
from pydantic import BaseModel
from dotenv import load_dotenv

from langchain_openai import ChatOpenAI
from langchain_core.messages import SystemMessage, HumanMessage, RemoveMessage
from langgraph.graph import StateGraph, START, END, MessagesState
from langgraph.checkpoint.memory import MemorySaver

load_dotenv()
model = ChatOpenAI(model="gpt-4o-mini")

# Extend base state to include persistent summary storage
class CustomChatState(MessagesState):
    summary: str

Step 2: Implement the Conversational Node (`chat_node`)

The main chat node checks whether a running summary exists. If it does, the summary is injected as a SystemMessage at the start of the payload before invoking the model.

PYTHON

def chat_node(state: CustomChatState):
    summary = state.get("summary", "")
    
    # Prepend summary as SystemMessage if it exists
    if summary:
        system_message = SystemMessage(content=f"Summary of conversation so far: {summary}")
        messages_to_send = [system_message] + state["messages"]
    else:
        messages_to_send = state["messages"]
        
    response = model.invoke(messages_to_send)
    return {"messages": [response]}

Step 3: Implement Summarization with Explicit Deletion (`summarize_node`)

This node runs only when message depth exceeds the defined limit. It updates the running summary and purges old messages using RemoveMessage.

PYTHON

def summarize_node(state: CustomChatState):
    existing_summary = state.get("summary", "")
    messages = state["messages"]
    
    # Preserve the most recent 2 messages, summarize everything preceding them
    messages_to_summarize = messages[:-2]
    
    if existing_summary:
        prompt = (
            f"Existing conversation summary:\n{existing_summary}\n\n"
            f"Extend the summary by incorporating the new messages above:"
        )
    else:
        prompt = "Summarize the conversational interaction above concisely:"
        
    summary_payload = messages_to_summarize + [HumanMessage(content=prompt)]
    summary_response = model.invoke(summary_payload)
    new_summary = summary_response.content
    
    # Mark the older summarized messages for permanent state deletion
    delete_actions = [RemoveMessage(id=msg.id) for msg in messages_to_summarize]
    
    return {
        "summary": new_summary,
        "messages": delete_actions
    }

Step 4: Routing Logic and Graph Compilation

A conditional router routes to summarize_node only if total raw messages exceed 6.

PYTHON

def routing_condition(state: CustomChatState) -> Literal["summarize_node", "__end__"]:
    if len(state["messages"]) > 6:
        return "summarize_node"
    return END

# Construct the Workflow
workflow = StateGraph(CustomChatState)

workflow.add_node("chat_node", chat_node)
workflow.add_node("summarize_node", summarize_node)

workflow.add_edge(START, "chat_node")
workflow.add_conditional_edges("chat_node", routing_condition)
workflow.add_edge("summarize_node", END)

app = workflow.compile(checkpointer=MemorySaver())

Step 5: Execution Trace

PYTHON

config = {"configurable": {"thread_id": "math-session-1"}}

# Turns 1 to 3: Increment messages to 6 total (3 user, 3 AI)
app.invoke({"messages": [("user", "Hello, let's discuss quantum mechanics.")]}, config=config)
app.invoke({"messages": [("user", "Who is Max Planck?")]}, config=config)
app.invoke({"messages": [("user", "What constant did he invent?")]}, config=config)

# Current State Snapshot
state = app.get_state(config).values
print("Message Count:", len(state["messages"]))  # 6
print("Summary exists:", bool(state.get("summary"))) # False

# Turn 4: Pushes count to 8 -> Exceeds limit -> Triggers Summarization
app.invoke({"messages": [("user", "Explain Planck's constant simply.")]}, config=config)

post_state = app.get_state(config).values
print("Message Count:", len(post_state["messages"]))  # 2 (Only Turn 4 User & AI)
print("Summary content:", post_state["summary"])
# Output: "The user and AI discussed quantum mechanics, focusing on Max Planck and his discovery of Planck's constant..."

12. Architectural Comparison of Context Management Strategies

Feature / Metric	Naive Buffer (Default)	Message Trimming (`trim_messages`)	Explicit Deletion (`RemoveMessage`)	Dynamic Summarization
Storage Impact	Unbounded linear growth	Unbounded linear growth	Bounded (Constant $O(K)$ )	Bounded (Constant $O(K)$ )
Context Retention	100% until context crash	Lost permanently outside $N$ window	Lost permanently upon deletion	Retained via semantic compression
Token Cost / Turn	Scales linearly ( $O(N)$ )	Bounded by `max_tokens` ( $O(1)$ )	Bounded by queue size	Bounded (Recent + summary payload)
Implementation Complexity	Zero configuration	Very Low	Low	Moderate
LLM Overhead	Single API call	Single API call	Zero extra LLM calls	Dual LLM calls during compress cycle
Risk of Crash	Extremely High	None	None	None

13. Summary and Key Takeaways

LLMs are Stateless Primitives: Every interaction is completely isolated. Multi-turn dialogue requires an external conversation buffer that feeds past interactions back into the model.
Checkpointers Enable Graph Memory: LangGraph tracks state snapshots across supersteps using checkpointers (e.g., MemorySaver, PostgresSaver).
Threads Provide Isolation: Conversational spaces are partitioned by thread_id inside RunnableConfig.
Ephemeral vs. Persistent Checkpointing:
- MemorySaver stores state exclusively within volatile RAM—all history is lost on process restart.
- Production deployments require persistent stores like PostgresSaver running against managed or Dockerized PostgreSQL instances.
The Context Window Overflow Barrier: Continuously appending uncompressed interactions eventually exceeds the model's context ceiling ( $W_{\text{ctx}}$ ), causing API failures, high token costs, and slow response times.
Trimming Conceals Without Deleting: trim_messages cuts the input payload to a set token budget before making the model call, but it preserves raw messages intact within database storage.
State Cleanup via RemoveMessage: Passing RemoveMessage(id=...) into a messages channel permanently removes selected historical turns from the persistent store.
Summarization is the Production Standard: Dynamic summarization combines older turns into a condensed summary, uses RemoveMessage to purge the originals, and passes the running summary along with the newest messages to preserve vital context indefinitely.

How To Implement Short Term Memory Using LangGraph

Implementing Short-Term Memory in LangGraph

1. The Core Mental Model: Stateless LLMs and Short-Term Memory

The Stateless Nature of LLMs

The Conversation Buffer (Short-Term Memory)

2. LangGraph Memory Architecture: Checkpointers & Threads

3. Ephemeral In-Memory Implementation (MemorySaver)

Implementation Code

Multi-Turn Execution Across Distinct Threads

4. Production-Grade Persistent Memory with PostgreSQL

The Volatility Problem

5. Setting Up Dockerized PostgreSQL for LangGraph

Step 1: Docker Compose Configuration (docker-compose.yml)

Step 2: Spin Up the Container

Step 3: Install Required Python Dependencies

6. Implementing Persistent Graph State with PostgresSaver

Proving Persistence Across Process Termination

7. The Context Window Overflow Problem

The Mechanism of Failure

8. Mitigation Strategy 1: Message Trimming

Implementation with trim_messages

Behavioral Failure Mode of Trimming

9. Mitigation Strategy 2: Explicit Message Deletion

Mechanism

10. Mitigation Strategy 3: Dynamic Summarization with Selective Deletion

Core Architecture

11. Complete Code Walkthrough: Summarization Workflow

Step 1: Define Custom State Structure

Step 2: Implement the Conversational Node (chat_node)

Step 3: Implement Summarization with Explicit Deletion (summarize_node)

Step 4: Routing Logic and Graph Compilation

Step 5: Execution Trace

12. Architectural Comparison of Context Management Strategies

13. Summary and Key Takeaways

3. Ephemeral In-Memory Implementation (`MemorySaver`)

Step 1: Docker Compose Configuration (`docker-compose.yml`)

6. Implementing Persistent Graph State with `PostgresSaver`

Implementation with `trim_messages`

Step 2: Implement the Conversational Node (`chat_node`)

Step 3: Implement Summarization with Explicit Deletion (`summarize_node`)