Implementation:Langchain ai Langgraph InMemorySaver Init

Property	Value
API	`InMemorySaver.__init__(self, *, serde=None, factory=defaultdict)`
Type	API Doc
Workflow	Human_in_the_Loop_Agent
Pipeline Stage	Configuration
Repository	Langchain_ai_Langgraph
Source	`libs/checkpoint/langgraph/checkpoint/memory/__init__.py:L83-98` (class at L31)

Overview

InMemorySaver is an in-memory checkpoint saver that stores graph execution state in Python dictionaries. It extends BaseCheckpointSaver[str], AbstractContextManager, and AbstractAsyncContextManager, providing both synchronous and asynchronous checkpoint operations. It is intended for debugging and testing only; for production, PostgresSaver / AsyncPostgresSaver from langgraph-checkpoint-postgres is recommended.

Description

The InMemorySaver class maintains three primary data structures:

storage: A nested defaultdict mapping thread_id -> checkpoint_ns -> checkpoint_id -> (checkpoint, metadata, parent_id). This holds serialized checkpoint data.
writes: A defaultdict mapping (thread_id, checkpoint_ns, checkpoint_id) -> {(task_id, write_idx): (task_id, channel, value, task_path)}. This tracks pending writes from node executions.
blobs: A dictionary mapping (thread_id, checkpoint_ns, channel, version) -> (type, bytes). This stores serialized channel values keyed by version.

On initialization, the constructor sets up these three data structures and optionally manages their lifecycle through an ExitStack if a custom factory (such as PersistentDict) is provided.

Usage

The most common usage pattern is to instantiate InMemorySaver with default arguments and pass it to StateGraph.compile():

from langgraph.checkpoint.memory import InMemorySaver
from langgraph.graph import StateGraph

memory = InMemorySaver()
graph = builder.compile(checkpointer=memory)

Code Reference

Source Location

Item	Path
File	`libs/checkpoint/langgraph/checkpoint/memory/__init__.py`
Class	Line 31
`__init__`	Lines 83-98

Signature

class InMemorySaver(
    BaseCheckpointSaver[str], AbstractContextManager, AbstractAsyncContextManager
):
    def __init__(
        self,
        *,
        serde: SerializerProtocol | None = None,
        factory: type[defaultdict] = defaultdict,
    ) -> None:

Import

from langgraph.checkpoint.memory import InMemorySaver

I/O Contract

Parameter	Type	Default	Description
`serde`	None	`None`	The serializer to use for serializing and deserializing checkpoints. If `None`, the default serializer from `BaseCheckpointSaver` is used.
`factory`	`type[defaultdict]`	`defaultdict`	The factory class for creating the storage dictionaries. Defaults to `defaultdict`. Can be set to `PersistentDict` for disk-backed storage.

Returns: None (constructor)

Side effects:

Initializes self.storage as a nested defaultdict
Initializes self.writes as a defaultdict(dict)
Initializes self.blobs using the factory
Creates an ExitStack and optionally enters the context managers if a custom factory is provided

Usage Examples

Basic initialization for testing

from langgraph.checkpoint.memory import InMemorySaver
from langgraph.graph import StateGraph

# Create in-memory checkpointer
memory = InMemorySaver()

# Build and compile a graph with the checkpointer
builder = StateGraph(int)
builder.add_node("add_one", lambda x: x + 1)
builder.set_entry_point("add_one")
builder.set_finish_point("add_one")
graph = builder.compile(checkpointer=memory)

# Invoke with a thread_id for state persistence
result = graph.invoke(1, {"configurable": {"thread_id": "thread-1"}})

Using as a context manager

from langgraph.checkpoint.memory import InMemorySaver

with InMemorySaver() as memory:
    graph = builder.compile(checkpointer=memory)
    graph.invoke(input_data, {"configurable": {"thread_id": "t1"}})

Async usage

import asyncio
from langgraph.checkpoint.memory import InMemorySaver

async def main():
    async with InMemorySaver() as memory:
        graph = builder.compile(checkpointer=memory)
        result = await graph.ainvoke(1, {"configurable": {"thread_id": "thread-1"}})

asyncio.run(main())

Related Pages

Page Connections

Double-click a node to navigate. Hold to expand connections.

Principle

Implementation

Heuristic

Environment