LangGraph Persistent Memory: Beyond MemorySaver

The Problem with LangGraph's Built-In Memory

LangGraph ships with checkpoint savers — InMemorySaver, SqliteSaver, PostgresSaver — that persist a graph's full state between invocations. They solve checkpointing (resuming a graph after a crash) and short-term continuity within a thread. But they weren't designed as memory systems.

When you ask a LangGraph agent "what did we work on last week?", a checkpoint saver has no way to answer: it stores structured JSON snapshots, not retrievable episodic memory. You can't search across threads. You can't weight recent interactions more heavily. You can't extract entities from conversation history and build a knowledge graph. And stale state from months ago is just as present as what happened yesterday.

The result is a ceiling. Agents built entirely on LangGraph's built-in persistence tend to degrade over time: the context window fills with irrelevant state, older decisions are forgotten, and multi-agent graphs have no shared memory layer.

What Checkpointing Misses

Checkpoint savers answer: "what was the exact state of this graph thread at this point in time?" That's the right answer for graph resumption. It's the wrong answer for agent memory.

Agent memory needs to answer: "what is relevant to know right now, given this query?" The two requirements diverge quickly:

The right architecture uses both: LangGraph's checkpoint saver for graph lifecycle management, and Dakera for semantic long-term memory. They operate at different layers and don't overlap.

Architecture: Two Memory Layers

Think of it as two separate concerns running in parallel:

┌─────────────────────────────────────────────────┐
│                 LangGraph StateGraph             │
│                                                  │
│  ┌──────────┐    ┌──────────┐    ┌──────────┐  │
│  │  recall  │───▶│  agent   │───▶│  store   │  │
│  │   node   │    │   node   │    │   node   │  │
│  └──────────┘    └──────────┘    └──────────┘  │
│       │                                │         │
│       └──────── Dakera MCP/SDK ────────┘         │
│                       │                          │
│   SqliteSaver / PostgresSaver (checkpoints)       │
└───────────────────────┼──────────────────────────┘
                        │
              ┌─────────▼─────────┐
              │   Dakera Server   │
              │  (self-hosted)    │
              │                  │
              │  • HNSW index    │
              │  • BM25 index    │
              │  • Knowledge     │
              │    graph         │
              │  • Decay engine  │
              └──────────────────┘

The graph's recall node queries Dakera at the start of each invocation, injecting relevant long-term context into the LangGraph state. The store node writes key outcomes back to Dakera at the end. The checkpoint saver still handles state snapshots for thread continuity — it just doesn't need to store everything.

Setup: Dakera + LangGraph

Start the Dakera server via Docker:

docker run -d --name dakera \
  -p 3300:3300 \
  -e DAKERA_API_KEY=dk-your-key \
  -v dakera-data:/data \
  ghcr.io/dakera-ai/dakera:latest

Install the Python packages:

pip install dakera langgraph langchain-anthropic

Create a shared Dakera client:

import os
from dakera import DakeraClient

dakera = DakeraClient(
    base_url=os.environ.get("DAKERA_URL", "http://localhost:3300"),
    api_key=os.environ.get("DAKERA_API_KEY", ""),
)

Basic Integration — Memory Nodes in StateGraph

Here's the core pattern: a TypedDict state that includes a memories field, populated by a recall node before the agent runs, and written back by a store node after.

from typing import TypedDict, Annotated, List
from langgraph.graph import StateGraph, END
from langgraph.checkpoint.sqlite import SqliteSaver
from langchain_anthropic import ChatAnthropic
from langchain_core.messages import HumanMessage, SystemMessage, AIMessage

class AgentState(TypedDict):
    messages: Annotated[List, "conversation messages"]
    memories: Annotated[List[str], "recalled long-term memories"]
    agent_id: str

llm = ChatAnthropic(model="claude-sonnet-5")

def recall_node(state: AgentState) -> dict:
    """Query Dakera for memories relevant to the latest user message."""
    messages = state["messages"]
    if not messages:
        return {"memories": []}

    # Use the last user message as the recall query
    last_human = next(
        (m.content for m in reversed(messages) if isinstance(m, HumanMessage)),
        None
    )
    if not last_human:
        return {"memories": []}

    results = dakera.search_memories(
        agent_id=state["agent_id"],
        query=last_human,
        top_k=5,
        min_score=0.3,
    )
    memories = [r["content"] for r in results]
    return {"memories": memories}

def agent_node(state: AgentState) -> dict:
    """Run the LLM with long-term memories injected into context."""
    system_parts = ["You are a helpful assistant with persistent memory."]
    if state["memories"]:
        memory_block = "\n".join(f"- {m}" for m in state["memories"])
        system_parts.append(f"\nRelevant memories from previous sessions:\n{memory_block}")

    response = llm.invoke([
        SystemMessage(content="\n".join(system_parts)),
        *state["messages"],
    ])
    return {"messages": state["messages"] + [response]}

def store_node(state: AgentState) -> dict:
    """Store the AI response and any key facts into Dakera."""
    messages = state["messages"]
    # Find the last AI message to persist
    last_ai = next(
        (m for m in reversed(messages) if isinstance(m, AIMessage)),
        None
    )
    if last_ai:
        last_human = next(
            (m.content for m in reversed(messages[:-1]) if isinstance(m, HumanMessage)),
            ""
        )
        dakera.store_memory(
            agent_id=state["agent_id"],
            content=f"Q: {last_human[:200]} A: {str(last_ai.content)[:400]}",
            importance=0.7,
        )
    return {}  # No state change — side-effect only

# Build the graph
builder = StateGraph(AgentState)
builder.add_node("recall", recall_node)
builder.add_node("agent", agent_node)
builder.add_node("store", store_node)

builder.set_entry_point("recall")
builder.add_edge("recall", "agent")
builder.add_edge("agent", "store")
builder.add_edge("store", END)

# Use SqliteSaver for thread checkpointing (short-term state)
memory = SqliteSaver.from_conn_string(":memory:")
graph = builder.compile(checkpointer=memory)

Invoke it with a thread ID and agent ID:

config = {"configurable": {"thread_id": "user-alice-session-1"}}

result = graph.invoke(
    {
        "messages": [HumanMessage(content="I prefer Python over TypeScript for backend work")],
        "memories": [],
        "agent_id": "assistant-alice",
    },
    config=config,
)

# Second invocation — Dakera will recall the preference
result2 = graph.invoke(
    {
        "messages": [HumanMessage(content="What language should I use for this new API?")],
        "memories": [],
        "agent_id": "assistant-alice",
    },
    config={"configurable": {"thread_id": "user-alice-session-2"}},  # New thread
)
print(result2["messages"][-1].content)
# "Based on your preference for Python for backend work, I'd recommend Python..."

Notice the second invocation uses a different thread ID — the SqliteSaver has no memory of session 1, but Dakera does. This is the gap checkpointing can't bridge.

Recall Node: Semantic Search at Graph Entry

The recall node is the most important piece. A naive implementation searches only on the latest message, but richer queries produce better results:

def recall_node_advanced(state: AgentState) -> dict:
    """Multi-strategy recall: semantic + recent + session context."""
    messages = state["messages"]
    agent_id = state["agent_id"]
    all_memories = []
    seen_ids = set()

    # 1. Semantic search on latest query
    if messages:
        last_human = next(
            (m.content for m in reversed(messages) if isinstance(m, HumanMessage)),
            None
        )
        if last_human:
            results = dakera.search_memories(
                agent_id=agent_id,
                query=last_human,
                top_k=5,
                min_score=0.35,
            )
            for r in results:
                if r["id"] not in seen_ids:
                    all_memories.append(r["content"])
                    seen_ids.add(r["id"])

    # 2. Pull the 3 most recent memories (continuity signal)
    recent = dakera.list_memories(
        agent_id=agent_id,
        limit=3,
        sort="created_at",
        sort_dir="desc",
    )
    for r in recent:
        if r["id"] not in seen_ids:
            all_memories.append(r["content"])
            seen_ids.add(r["id"])

    return {"memories": all_memories[:8]}

Cap the total to avoid filling the context window. Eight to twelve memories is typically the right balance for a 200K-token context model.

Store Node: Persisting Conversation Outcomes

Not every exchange is worth storing. A good store node filters for signal:

import re

# Phrases that signal storable facts
STORE_SIGNALS = [
    r"\bI prefer\b", r"\bmy \w+ is\b", r"\bI always\b",
    r"\bwe decided\b", r"\bthe architecture\b", r"\bthe deadline\b",
    r"\bremember that\b", r"\bdon't forget\b",
]

def should_store(text: str) -> bool:
    return any(re.search(p, text, re.IGNORECASE) for p in STORE_SIGNALS)

def store_node_smart(state: AgentState) -> dict:
    messages = state["messages"]
    agent_id = state["agent_id"]

    # Store user messages containing explicit facts or preferences
    for msg in messages:
        if isinstance(msg, HumanMessage) and should_store(msg.content):
            dakera.store_memory(
                agent_id=agent_id,
                content=msg.content,
                importance=0.85,
                metadata={"source": "user_explicit"},
            )

    # Store any AI message that contains a decision or recommendation
    last_ai = next(
        (m for m in reversed(messages) if isinstance(m, AIMessage)),
        None
    )
    if last_ai and should_store(str(last_ai.content)):
        dakera.store_memory(
            agent_id=agent_id,
            content=f"AI recommendation: {str(last_ai.content)[:500]}",
            importance=0.6,
            metadata={"source": "ai_response"},
        )

    return {}

Higher importance scores surface more readily in future recalls and decay more slowly. User-explicit facts should score 0.8–1.0; AI-generated summaries are typically 0.5–0.7.

Cross-Session Context in Practice

The value of this architecture becomes clear across a realistic multi-session workflow. Here's a coding assistant scenario spanning three days:

# Session 1 — User establishes context
invoke_graph("assistant-dev", "We're building a Rust CLI with clap 4 and tokio 1.38")
invoke_graph("assistant-dev", "We use anyhow for errors, not thiserror — team preference")
invoke_graph("assistant-dev", "The deploy target is a musl binary for Alpine Linux")

# Session 2 (next day) — New thread, Dakera recalls everything
result = invoke_graph(
    "assistant-dev",
    "I need to add a subcommand for exporting data as CSV"
)
# AI response will reference clap 4 syntax, anyhow, musl constraints
# without the user re-stating any of it

# Session 3 (two days later) — Querying decisions
result = invoke_graph(
    "assistant-dev",
    "What error handling approach are we using?"
)
# Recalls: "We use anyhow for errors, not thiserror — team preference"

Without Dakera, session 2 and session 3 start cold. The user re-explains the stack every time. With Dakera, the semantic recall surfaces the three stored facts in under 10ms.

Knowledge Graph: Entity Linking Across Runs

Dakera automatically extracts entities from stored memories and builds a traversable knowledge graph. For a coding assistant, this means entities like clap, tokio, anyhow, and relationships like uses, prefers, targets.

# After a few sessions, query the knowledge graph
entities = dakera.knowledge_graph(
    agent_id="assistant-dev",
    query="project dependencies",
    depth=2,
)

for node in entities["nodes"]:
    print(f"{node['name']} ({node['type']})")
# clap (library) — relationship: uses
# tokio (library) — relationship: uses
# anyhow (library) — relationship: prefers
# Alpine Linux (platform) — relationship: targets

# Graph traversal: what do we know about 'anyhow'?
related = dakera.graph_traverse(
    agent_id="assistant-dev",
    entity="anyhow",
    max_depth=2,
)
# Returns: team preference, error handling context, connected entities

Inject the knowledge graph into the LangGraph state alongside raw memories for richer context:

class AgentStateWithGraph(TypedDict):
    messages: Annotated[List, "messages"]
    memories: Annotated[List[str], "recalled memories"]
    entities: Annotated[List[str], "entity context from knowledge graph"]
    agent_id: str

def recall_node_with_graph(state: AgentStateWithGraph) -> dict:
    last_human = next(
        (m.content for m in reversed(state["messages"]) if isinstance(m, HumanMessage)),
        ""
    )
    memories_result = dakera.search_memories(
        agent_id=state["agent_id"],
        query=last_human,
        top_k=5,
    )
    graph_result = dakera.knowledge_graph(
        agent_id=state["agent_id"],
        query=last_human,
        depth=1,
    )
    entity_summaries = [
        f"{n['name']} ({n['type']}): {n.get('summary', '')}"
        for n in graph_result.get("nodes", [])[:6]
    ]
    return {
        "memories": [r["content"] for r in memories_result],
        "entities": entity_summaries,
    }

Memory Decay: Keeping Context Fresh

Without decay, memory stores accumulate stale facts indefinitely. A user's technology preferences from two years ago shouldn't have the same weight as last week's decision. Dakera's decay engine scores each memory based on recency and access frequency.

# Configure decay policy for this agent
dakera.set_memory_policy(
    agent_id="assistant-dev",
    policy={
        "decay_enabled": True,
        "half_life_days": 30,          # Score halves every 30 days
        "min_importance_to_decay": 0.3, # Don't decay high-value memories quickly
        "access_boost": 0.1,            # Each recall boosts score by 10%
    }
)

This means frequently-recalled facts (the user's core preferences, key architecture decisions) stay surface-level while old one-off exchanges decay gracefully. The semantic search automatically returns higher-scored memories first.

To exclude decayed memories from recall:

results = dakera.search_memories(
    agent_id="assistant-dev",
    query="error handling",
    top_k=5,
    min_score=0.25,  # Filter out very stale memories (score < 0.25)
)

Multi-Agent Memory Sharing

LangGraph multi-agent graphs — where a supervisor dispatches to specialized subgraphs — benefit from shared memory with per-agent namespaces:

# Supervisor graph and two specialized agents share a common namespace
SHARED_NS = "team-project-shared"
PLANNER_NS = "agent-planner"
REVIEWER_NS = "agent-reviewer"

def planner_recall(state):
    # Search both planner-private and shared namespaces
    private = dakera.search_memories(agent_id=PLANNER_NS, query=state["task"], top_k=3)
    shared = dakera.search_memories(agent_id=SHARED_NS, query=state["task"], top_k=3)
    return {"memories": [r["content"] for r in private + shared]}

def reviewer_recall(state):
    # Reviewer sees its own memory + shared context
    private = dakera.search_memories(agent_id=REVIEWER_NS, query=state["task"], top_k=3)
    shared = dakera.search_memories(agent_id=SHARED_NS, query=state["task"], top_k=3)
    return {"memories": [r["content"] for r in private + shared]}

def planner_store(state):
    # Key planning decisions go to shared namespace for reviewer to see
    last_ai = next((m for m in reversed(state["messages"]) if isinstance(m, AIMessage)), None)
    if last_ai:
        dakera.store_memory(
            agent_id=SHARED_NS,
            content=f"Planner decision: {str(last_ai.content)[:400]}",
            importance=0.8,
            metadata={"source_agent": "planner"},
        )
    return {}

This is the pattern that enables true multi-agent collaboration: the reviewer can see what the planner decided without being in the same graph invocation or thread.

Production Deployment

For production LangGraph + Dakera deployments:

Asynchronous Store Node

The store node is a side-effect and doesn't affect the response. Run it asynchronously to avoid adding latency to the graph's critical path:

import asyncio
from concurrent.futures import ThreadPoolExecutor

executor = ThreadPoolExecutor(max_workers=4)

def store_node_async(state: AgentState) -> dict:
    """Fire-and-forget store — doesn't block graph completion."""
    messages = state["messages"]
    agent_id = state["agent_id"]

    def _store():
        last_ai = next((m for m in reversed(messages) if isinstance(m, AIMessage)), None)
        if last_ai and should_store(str(last_ai.content)):
            dakera.store_memory(
                agent_id=agent_id,
                content=str(last_ai.content)[:500],
                importance=0.6,
            )

    executor.submit(_store)
    return {}

Using the MCP Interface

If you're deploying LangGraph with a Claude model via Claude Code or Cursor, you can skip the Python SDK entirely and use Dakera's MCP server. The agent calls dakera_recall and dakera_store as MCP tools directly — no recall/store nodes needed. See the MCP Server docs for configuration.

Connection Pooling

# Initialize once at module level, reuse across all graph invocations
dakera = DakeraClient(
    base_url=os.environ["DAKERA_URL"],
    api_key=os.environ["DAKERA_API_KEY"],
    timeout=5.0,      # Fast timeout — don't stall the graph on memory failure
    max_retries=1,    # One retry; don't queue up on a slow memory server
)

Graceful Degradation

def recall_node_safe(state: AgentState) -> dict:
    """Return empty memories on Dakera failure — graph continues."""
    try:
        results = dakera.search_memories(
            agent_id=state["agent_id"],
            query=state["messages"][-1].content if state["messages"] else "",
            top_k=5,
        )
        return {"memories": [r["content"] for r in results]}
    except Exception:
        return {"memories": []}  # Degrade gracefully, don't fail the graph

When to Use What

Use LangGraph's checkpoint savers for:

• Resuming a graph after a failure mid-run
• Human-in-the-loop interrupts and approvals
• Replaying a graph from a specific checkpoint
• Short-term thread continuity within a single conversation session

Use Dakera for:

• Cross-session memory (user preferences, past decisions, project context)
• Semantic search over past interactions ("what did we discuss about auth?")
• Multi-agent shared memory across different graph invocations
• Long-horizon agents that need to remember things weeks or months old
• Knowledge graphs of entities extracted from conversations
• Memory decay to keep context fresh as facts age

They're complementary. A production LangGraph application should typically use both: a checkpoint saver for graph lifecycle, and Dakera for the semantic memory layer that makes agents feel genuinely knowledgeable across time.

Related: LangChain integration · Multi-Agent Memory Patterns · Dakera Python SDK · Temporal Memory & Decay · Quick Start