Adding Cross-Thread Memory to LangGraph: A Worked Example

The setup

We're going to take a small LangGraph agent and add cross-thread, multi-tenant memory to it. The agent is a personal coding assistant — it answers questions about a user's projects, remembers their preferences, and updates what it knows when those preferences change.

Without external memory, the agent has the Checkpointer (PostgresSaver in prod) — that's thread-state persistence. Open a new thread tomorrow, the agent has forgotten everything.

What we want: a new thread on day 7 should know the user uses Postgres, prefers TypeScript, deploys via ./deploy.sh, and switched from MySQL to CockroachDB last week — that last fact should be superseded, not stored alongside the old one.

Baseline: the LangGraph agent without memory

The starting point. A minimal StateGraph with one agent node, hooked up to a PostgresSaver checkpointer for thread-state persistence.

from langgraph.graph import StateGraph, START, END
from langgraph.checkpoint.postgres import PostgresSaver
from typing_extensions import TypedDict

class State(TypedDict):
    user_id: str
    messages: list[dict]
    answer: str

def agent_node(state: State) -> dict:
    # Call your LLM with state["messages"]
    # ...
    return {"answer": llm_call(state["messages"])}

graph = StateGraph(State)
graph.add_node("agent", agent_node)
graph.add_edge(START, "agent")
graph.add_edge("agent", END)

with PostgresSaver.from_conn_string(DB_URL) as checkpointer:
    app = graph.compile(checkpointer=checkpointer)
    result = app.invoke(
        {"user_id": "u_42", "messages": [{"role": "user", "content": "..."}]},
        config={"configurable": {"thread_id": "t_001"}},
    )

Inside thread t_001, the Checkpointer keeps state. Across threads it doesn't. That's the gap we're closing.

Option A: Build it on top of LangGraph Store

LangGraph ships InMemoryStore and PostgresStorefor cross-thread memory. They're primitives — key-value namespaces with optional vector indexing. To use them as a memory layer, you wire entity extraction, recall, and contradiction handling yourself.

Here's what a serious DIY version looks like. Three additional nodes (extract, save, recall), per-user namespacing, vector indexing for semantic recall.

from langgraph.store.postgres import PostgresStore
from openai import OpenAI
import uuid

openai = OpenAI()

def embed(text: str) -> list[float]:
    return openai.embeddings.create(
        model="text-embedding-3-small", input=text
    ).data[0].embedding

def extract_node(state: State, store: PostgresStore) -> dict:
    # Ask an LLM to pull facts out of the user's latest message.
    facts = llm_extract_facts(state["messages"][-1]["content"])
    return {"facts_to_save": facts}

def save_node(state: State, store: PostgresStore) -> dict:
    namespace = ("memories", state["user_id"])
    for fact in state.get("facts_to_save", []):
        # NOTE: no conflict resolution. New facts pile up alongside
        # old contradicting ones unless you write the dedup yourself.
        store.put(namespace, str(uuid.uuid4()), {
            "content": fact,
            "embedding": embed(fact),
        })
    return {}

def recall_node(state: State, store: PostgresStore) -> dict:
    namespace = ("memories", state["user_id"])
    query_emb = embed(state["messages"][-1]["content"])
    hits = store.search(namespace, query=query_emb, limit=5)
    return {"context": [h.value["content"] for h in hits]}

# Wire into the graph
graph = StateGraph(State)
graph.add_node("recall", recall_node)
graph.add_node("agent", agent_node)
graph.add_node("extract", extract_node)
graph.add_node("save", save_node)
graph.add_edge(START, "recall")
graph.add_edge("recall", "agent")
graph.add_edge("agent", "extract")
graph.add_edge("extract", "save")
graph.add_edge("save", END)

This works. It also has four problems you'll hit in production:

1. Extraction quality. The llm_extract_facts call needs a good prompt + structured output schema + few-shot examples. Building that to production grade is several engineering weeks of prompt iteration.
2. No conflict resolution. "User uses MySQL" gets saved on Monday. "User switched to CockroachDB" gets saved on Friday. Both now sit in the store, both come back from recall, the LLM has to figure it out at read time — and often picks wrong.
3. Embedding choice + cost. text-embedding-3-small is fine, but you're now paying OpenAI for every save AND every recall. At modest volume that's real money. Switching embedding models means re-embedding everything.
4. No browsable view.Your users can't see what the agent has learned about them. The store is opaque. Building a dashboard on top is its own project.

Each of these is solvable. The combined engineering cost is roughly a quarter of one engineer's time. If memory retrieval is your competitive edge, that's the right trade. If it's table-stakes infra for an agent product whose value is elsewhere, you probably want option B.

Option B: Wire Ricord alongside Checkpointer

Ricord is a hosted memory layer that runs alongside LangGraph's Checkpointer — it doesn't replace anything. Same agent_node, same Checkpointer for thread-state. Two thin nodes for recall and save; Ricord handles extraction, embeddings, namespacing, contradiction resolution, GDPR delete, the dashboard, and the cross-client MCP server.

import os, requests
from langgraph.graph import StateGraph, START, END
from langgraph.checkpoint.postgres import PostgresSaver

RICORD = "https://api.ricord.ai/v1"
HEADERS = {"Authorization": f"Bearer {os.environ['RICORD_API_KEY']}"}

def recall_node(state: State) -> dict:
    r = requests.get(
        f"{RICORD}/memories/recall",
        headers=HEADERS,
        params={
            "user_id": state["user_id"],
            "query": state["messages"][-1]["content"],
            "k": 5,
        },
    )
    return {"context": [m["content"] for m in r.json()["hits"]]}

def save_node(state: State) -> dict:
    # No extraction prompt to write — Ricord extracts server-side.
    # No deduplication code — Ricord handles supersedence at ingest.
    requests.post(
        f"{RICORD}/memories",
        headers=HEADERS,
        json={
            "user_id": state["user_id"],
            "content": state["messages"][-1]["content"],
        },
    )
    return {}

graph = StateGraph(State)
graph.add_node("recall", recall_node)
graph.add_node("agent", agent_node)
graph.add_node("save", save_node)
graph.add_edge(START, "recall")
graph.add_edge("recall", "agent")
graph.add_edge("agent", "save")
graph.add_edge("save", END)

with PostgresSaver.from_conn_string(DB_URL) as checkpointer:
    app = graph.compile(checkpointer=checkpointer)
    result = app.invoke(
        {"user_id": "u_42", "messages": [{"role": "user", "content": "..."}]},
        config={"configurable": {"thread_id": "t_001"}},
    )

The agent_node is unchanged. The Checkpointer still does its job. The two new nodes are 8 lines apiece. Everything that was an open engineering problem in option A is closed here.

The contradiction case, end-to-end

Let's walk through what happens when the user's facts change. Day 1, thread t_001:

User: "We use MySQL for the main app."
[save_node sends: user_id=u_42, content="We use MySQL for the main app."]
[Ricord extracts: database_choice(user_id=u_42, value="MySQL")]

Day 7, thread t_017 (different thread, same user):

User: "What database do I use?"
[recall_node returns: ["We use MySQL for the main app."]]
[agent_node answers: "MySQL"]
User: "We switched to CockroachDB last week."
[save_node sends: user_id=u_42, content="We switched to CockroachDB last week."]
[Ricord extracts: database_choice(user_id=u_42, value="CockroachDB")]
[Ricord notes: superseded the previous fact for database_choice(u_42)]

Day 14, thread t_034:

User: "What database am I using these days?"
[recall_node returns: ["We switched to CockroachDB last week."] only]
[agent_node answers: "CockroachDB"]

The old fact isn't recalled. It's still in the audit log (you can ask "what database did I used to use" and get a temporal answer), but the canonical recall returns only the current truth. Option A wouldn't do this without you writing the supersedence logic.

Multi-tenant by default

The user_idparameter is the whole multi-tenant story for option B. Every Ricord write and read is scoped to the user; there's no namespace prefix to manage, no DIY isolation code, no risk of one user's data leaking into another's recall. Option A's namespace = ("memories", state["user_id"])tuple does the same thing — provided you remember to use it in every Store call. Forget once and you have a cross-user data leak.

Bonus: the same memory works outside LangGraph

Because Ricord is an external service with an MCP server, the memory written by your LangGraph backend is reachable from Claude Desktop, Cursor, Codex, and any custom client that speaks the Model Context Protocol. Useful if your agent product has both an automated production deployment (the LangGraph backend) AND a human-facing surface where developers debug things by chatting with their AI. Install in any MCP client →

Option A's LangGraph Store has no equivalent. Memory inside the Store is reachable from LangGraph and nowhere else.

When to pick which

Honest framing:

• Option A (Store + DIY)if memory retrieval is genuinely your product's competitive edge, you have an ML engineer with cycles, and the extraction prompt is something you want to own forever.
• Option B (Ricord alongside Checkpointer) for everyone else. Builders shipping agent products where memory is necessary infra but not the differentiator. Teams that want a wiki view for their users out of the box. Anyone who'd rather spend a quarter on product than on memory plumbing.

Getting started

Drop in the option B snippet. Get an API key, set RICORD_API_KEY, point the recall/save nodes at the API. Existing Checkpointer logic doesn't change.

pip install requests
# Get a key at https://ricord.ai/login?signup=true
export RICORD_API_KEY=rc_live_...