AI Session Memory Management: State Design, Reducers, and Checkpointing

Bug Example: Parallel Search Overwrites

A graph with a fan‑out node that launches three parallel search nodes returns only one node's output because the default write behavior overwrites the search_results field. The last node to finish replaces earlier results.

# Problematic State definition
class SearchState(TypedDict):
    query: str
    search_results: list[str]  # ← issue here

LangGraph solves this with a Reducer that defines how to merge concurrent writes.

from typing import TypedDict, Annotated
import operator

class SearchState(TypedDict):
    query: str
    # Use operator.add to append instead of overwrite
    search_results: Annotated[list[str], operator.add]

After adding the annotation, each parallel node appends its result to the same list, eliminating the data‑loss bug.

Reducer Mechanism

A Reducer is a function that tells the framework how to combine multiple writes to the same field. Besides operator.add, custom reducers can be defined:

def keep_latest(existing, new):
    """Keep only the newest value"""
    return new

def merge_dicts(existing, new):
    """Merge dictionaries, new values overwrite old ones"""
    return {**existing, **new}

def deduplicate(existing, new):
    """Append then deduplicate"""
    return list(set(existing + new))

class MyState(TypedDict):
    status: Annotated[str, keep_latest]
    metadata: Annotated[dict, merge_dicts]
    visited_urls: Annotated[list[str], deduplicate]

Each field must have a clear merge strategy; otherwise concurrent writes silently overwrite each other.

State Design Questions

Who can write which fields? Restrict nodes to only the fields they own (e.g., search nodes write search_results but not draft).

Overwrite or append? Sequential nodes usually overwrite; parallel nodes need an appending reducer.

What belongs in State? Only data that must be shared across nodes; temporary variables stay local.

Encode these constraints in node functions by returning only the fields a node manages:

def search_node(state: SearchState) -> dict:
    # Return only the field this node manages
    results = do_search(state["query"])
    return {"search_results": results}

Short‑term Memory and Checkpointing

State lives only for the duration of a single graph execution. When executions are long, interrupted, or need to persist across sessions, a persistent State is required.

Checkpointing stores the State after each node in a database (e.g., SQLite). The thread_id acts as a session identifier; using the same thread_id resumes from the last checkpoint.

from langgraph.graph import StateGraph, END
from langgraph.checkpoint.sqlite import SqliteSaver
from typing import TypedDict

class AnalysisState(TypedDict):
    topic: str
    search_results: list[str]
    analysis: str
    draft: str
    review_score: int

graph = StateGraph(AnalysisState)
# ... add nodes and edges ...
checkpointer = SqliteSaver.from_conn_string("checkpoints.db")
app = graph.compile(checkpointer=checkpointer)

config = {"configurable": {"thread_id": "analysis-task-001"}}
# First run (may be interrupted)
result = app.invoke({"topic": "AI code assistant market"}, config=config)
# Resume later
result = app.invoke(None, config=config)  # None means continue

Checkpointing also enables Time Travel —listing historical checkpoints, inspecting their States, and rolling back to any point for debugging:

# List checkpoints
history = list(app.get_state_history(config))
for checkpoint in history:
    print(f"Node: {checkpoint.metadata.get('source')}")
    print(f"State: {checkpoint.values}")
    print("---")

# Roll back to a specific checkpoint
target = history[3]  # 4th checkpoint
app.update_state(config, target.values)
# Re‑run from that point
result = app.invoke(None, config=config)

Thread‑level Memory for Multi‑User Isolation

Checkpointing isolates a single task, but in multi‑user scenarios each user must have an independent State. LangGraph uses thread_id to separate users:

# User A
config_a = {"configurable": {"thread_id": "user-a-task-001"}}
result_a = app.invoke({"topic": "User A query"}, config=config_a)

# User B
config_b = {"configurable": {"thread_id": "user-b-task-001"}}
result_b = app.invoke({"topic": "User B query"}, config=config_b)

When shared information (e.g., user preferences) is needed across a user's multiple threads, an InMemoryStore with a namespace provides cross‑thread memory:

from langgraph.store.memory import InMemoryStore
store = InMemoryStore()
# Store preference
store.put(namespace=("prefs", "user-a"), key="reading_style", value={"language": "Python", "style": "functional"})

# Retrieve in a node
def write_node(state, config, store):
    user_id = config["configurable"].get("user_id", "default")
    prefs = store.get(namespace=("prefs", user_id), key="reading_style")
    style = prefs.value.get("style", "concise") if prefs else "concise"
    # ... use style ...

Long‑term Memory with Vector Stores

Agents that should become smarter over time need a permanent knowledge base. This is achieved with a vector database (e.g., Chroma) that stores embeddings of past outputs and retrieves the most relevant pieces on demand.

import chromadb, uuid
from langchain_openai import OpenAIEmbeddings

client = chromadb.Client()
collection = client.create_collection("agent_knowledge")
embeddings = OpenAIEmbeddings()

def store_knowledge(content: str, metadata: dict = None):
    """Store a piece of knowledge in the vector store"""
    embedding = embeddings.embed_query(content)
    collection.add(
        documents=[content],
        embeddings=[embedding],
        metadatas=[metadata or {}],
        ids=[str(uuid.uuid4())]
    )

def retrieve_knowledge(query: str, n_results: int = 3) -> list[str]:
    """Retrieve the most semantically similar pieces of knowledge"""
    query_emb = embeddings.embed_query(query)
    results = collection.query(query_embeddings=[query_emb], n_results=n_results)
    return results["documents"][0]

In a node, the agent first pulls relevant historical briefs, appends them to the current context, and finally stores the new brief for future reuse.

Four Layers of Memory

The memory hierarchy in a LangGraph application consists of:

Graph‑internal State – lives for a single execution; implemented with TypedDict and optional reducers.

Checkpointing – persists State across interruptions; implemented with SqliteSaver or PostgresSaver.

Thread‑level Memory – isolates State per user or per logical thread; implemented with InMemoryStore and a namespace.

Long‑term Memory – permanent knowledge base accessed via vector stores such as Chroma or Pinecone.

Projects can start with the simplest layer and add additional layers only when the problem demands them.

Complete Example: Daily Market Briefing Generator

The example combines all layers: parallel searches with a list‑appending reducer, checkpointing for crash‑safe runs, thread‑level storage of user reading preferences, and long‑term storage of published briefs.

from langgraph.graph import StateGraph, END
from langgraph.checkpoint.sqlite import SqliteSaver
from langgraph.store.memory import InMemoryStore
from typing import TypedDict, Annotated
import operator

class BriefingState(TypedDict):
    date: str
    topic: str
    raw_sources: Annotated[list[str], operator.add]
    analysis: str
    draft: str
    review_score: int
    final_briefing: str

# Node definitions (search, analyze, write, review, publish, routing) omitted for brevity
# ... build graph, add edges, set entry point, compile with checkpointing and store ...
checkpointer = SqliteSaver.from_conn_string("briefings.db")
store = InMemoryStore()
app = builder.compile(checkpointer=checkpointer, store=store)

config = {"configurable": {"thread_id": "briefing-2026-08-21", "user_id": "user-001"}}
result = app.invoke({"date": "2026-08-21", "topic": "AI industry trends", "raw_sources": []}, config=config)

Key takeaways from the example: raw_sources uses operator.add to merge parallel search results.

SQLite checkpointing enables safe resume after interruptions.

Thread‑level store keeps user preferences isolated per user.

Vector‑store functions store_knowledge / retrieve_knowledge accumulate and reuse historical briefs.

Practical Principles

Keep State minimal. Only include data that must be shared across nodes.

Define reducers for every parallel write. Forgetting this leads to silent data‑loss bugs.

Add Checkpointing early. The cost is low, but the benefit of avoiding full re‑runs is high.

Limit long‑term retrieval size. Typically 3‑5 results keep the context window manageable.

Use Time Travel for debugging. Jump directly to a failing node, adjust State, and re‑execute.

Conclusion

State in LangGraph is a global shared memory, not a simple pass‑through of the previous node's output. Properly designing who writes what, choosing overwrite versus append, and limiting State contents are essential. By layering short‑term State, Checkpointing, thread‑level memory, and long‑term vector stores, developers can build robust, scalable AI agents that persist knowledge across sessions and users.