Skip to content

Checkpoints

You can compile any LangGraph workflow with a CheckPointer to give your agent "memory" by persisting its state. This permits things like:

  • Remembering things across multiple interactions
  • Interrupting to wait for user input
  • Resiliance for long-running, error-prone agents
  • Time travel retry and branch from a previous checkpoint

Checkpoint

Bases: TypedDict

State snapshot at a given point in time.

Source code in langgraph/checkpoint/base.py 
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
class Checkpoint(TypedDict):
    """State snapshot at a given point in time."""

    v: int
    """The version of the checkpoint format. Currently 1."""
    ts: str
    """The timestamp of the checkpoint in ISO 8601 format."""
    channel_values: dict[str, Any]
    """The values of the channels at the time of the checkpoint.

    Mapping from channel name to channel snapshot value.
    """
    channel_versions: defaultdict[str, int]
    """The versions of the channels at the time of the checkpoint.

    The keys are channel names and the values are the logical time step
    at which the channel was last updated.
    """
    versions_seen: defaultdict[str, defaultdict[str, int]]
    """Map from node ID to map from channel name to version seen.

    This keeps track of the versions of the channels that each node has seen.

    Used to determine which nodes to execute next.
    """

v: int instance-attribute

The version of the checkpoint format. Currently 1.

ts: str instance-attribute

The timestamp of the checkpoint in ISO 8601 format.

channel_values: dict[str, Any] instance-attribute

The values of the channels at the time of the checkpoint.

Mapping from channel name to channel snapshot value.

channel_versions: defaultdict[str, int] instance-attribute

The versions of the channels at the time of the checkpoint.

The keys are channel names and the values are the logical time step at which the channel was last updated.

versions_seen: defaultdict[str, defaultdict[str, int]] instance-attribute

Map from node ID to map from channel name to version seen.

This keeps track of the versions of the channels that each node has seen.

Used to determine which nodes to execute next.

BaseCheckpointSaver

Bases: ABC

Source code in langgraph/checkpoint/base.py 
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
173
174
175
176
177
class BaseCheckpointSaver(ABC):
    serde: SerializerProtocol = JsonPlusSerializer()

    def __init__(
        self,
        *,
        serde: Optional[SerializerProtocol] = None,
    ) -> None:
        self.serde = serde or self.serde

    @property
    def config_specs(self) -> list[ConfigurableFieldSpec]:
        return [CheckpointThreadId, CheckpointThreadTs]

    def get(self, config: RunnableConfig) -> Optional[Checkpoint]:
        if value := self.get_tuple(config):
            return value.checkpoint

    def get_tuple(self, config: RunnableConfig) -> Optional[CheckpointTuple]:
        raise NotImplementedError

    def list(
        self,
        config: RunnableConfig,
        *,
        before: Optional[RunnableConfig] = None,
        limit: Optional[int] = None,
    ) -> Iterator[CheckpointTuple]:
        raise NotImplementedError

    def put(
        self,
        config: RunnableConfig,
        checkpoint: Checkpoint,
        metadata: CheckpointMetadata,
    ) -> RunnableConfig:
        raise NotImplementedError

    async def aget(self, config: RunnableConfig) -> Optional[Checkpoint]:
        if value := await self.aget_tuple(config):
            return value.checkpoint

    async def aget_tuple(self, config: RunnableConfig) -> Optional[CheckpointTuple]:
        raise NotImplementedError

    def alist(
        self,
        config: RunnableConfig,
        *,
        before: Optional[RunnableConfig] = None,
        limit: Optional[int] = None,
    ) -> AsyncIterator[CheckpointTuple]:
        raise NotImplementedError
        yield

    async def aput(
        self,
        config: RunnableConfig,
        checkpoint: Checkpoint,
        metadata: CheckpointMetadata,
    ) -> RunnableConfig:
        raise NotImplementedError

SerializerProtocol

Bases: Protocol

Protocol for serialization and deserialization of objects.

  • dumps: Serialize an object to bytes.
  • loads: Deserialize an object from bytes.

Valid implementations include the pickle, json and orjson modules.

Source code in langgraph/serde/base.py 
 4
 5
 6
 7
 8
 9
10
11
12
13
14
15
16
17
class SerializerProtocol(Protocol):
    """Protocol for serialization and deserialization of objects.

    - `dumps`: Serialize an object to bytes.
    - `loads`: Deserialize an object from bytes.

    Valid implementations include the `pickle`, `json` and `orjson` modules.
    """

    def dumps(self, obj: Any) -> bytes:
        ...

    def loads(self, data: bytes) -> Any:
        ...

Implementations

LangGraph also natively provides the following checkpoint implementations.

MemorySaver

Bases: BaseCheckpointSaver

An in-memory checkpoint saver.

This checkpoint saver stores checkpoints in memory using a defaultdict.

Note

Since checkpoints are saved in memory, they will be lost when the program exits. Only use this saver for debugging or testing purposes.

Parameters:

  • serde (Optional[SerializerProtocol], default: None ) –

    The serializer to use for serializing and deserializing checkpoints. Defaults to None.

Examples:

    import asyncio

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

    builder = StateGraph(int)
    builder.add_node("add_one", lambda x: x + 1)
    builder.set_entry_point("add_one")
    builder.set_finish_point("add_one")

    memory = MemorySaver()
    graph = builder.compile(checkpointer=memory)
    coro = graph.ainvoke(1, {"configurable": {"thread_id": "thread-1"}})
    asyncio.run(coro)  # Output: 2
Source code in langgraph/checkpoint/memory.py 
 16
 17
 18
 19
 20
 21
 22
 23
 24
 25
 26
 27
 28
 29
 30
 31
 32
 33
 34
 35
 36
 37
 38
 39
 40
 41
 42
 43
 44
 45
 46
 47
 48
 49
 50
 51
 52
 53
 54
 55
 56
 57
 58
 59
 60
 61
 62
 63
 64
 65
 66
 67
 68
 69
 70
 71
 72
 73
 74
 75
 76
 77
 78
 79
 80
 81
 82
 83
 84
 85
 86
 87
 88
 89
 90
 91
 92
 93
 94
 95
 96
 97
 98
 99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
173
174
175
176
177
178
179
180
181
182
183
184
185
186
187
188
189
190
191
192
193
194
195
196
197
198
199
class MemorySaver(BaseCheckpointSaver):
    """An in-memory checkpoint saver.

    This checkpoint saver stores checkpoints in memory using a defaultdict.

    Note:
        Since checkpoints are saved in memory, they will be lost when the program exits.
        Only use this saver for debugging or testing purposes.

    Args:
        serde (Optional[SerializerProtocol]): The serializer to use for serializing and deserializing checkpoints. Defaults to None.

    Examples:

            import asyncio

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

            builder = StateGraph(int)
            builder.add_node("add_one", lambda x: x + 1)
            builder.set_entry_point("add_one")
            builder.set_finish_point("add_one")

            memory = MemorySaver()
            graph = builder.compile(checkpointer=memory)
            coro = graph.ainvoke(1, {"configurable": {"thread_id": "thread-1"}})
            asyncio.run(coro)  # Output: 2
    """

    storage: defaultdict[str, dict[str, tuple[bytes, bytes]]]

    def __init__(
        self,
        *,
        serde: Optional[SerializerProtocol] = None,
    ) -> None:
        super().__init__(serde=serde)
        self.storage = defaultdict(dict)

    def get_tuple(self, config: RunnableConfig) -> Optional[CheckpointTuple]:
        """Get a checkpoint tuple from the in-memory storage.

        This method retrieves a checkpoint tuple from the in-memory storage based on the
        provided config. If the config contains a "thread_ts" key, the checkpoint with
        the matching thread ID and timestamp is retrieved. Otherwise, the latest checkpoint
        for the given thread ID is retrieved.

        Args:
            config (RunnableConfig): The config to use for retrieving the checkpoint.

        Returns:
            Optional[CheckpointTuple]: The retrieved checkpoint tuple, or None if no matching checkpoint was found.
        """
        thread_id = config["configurable"]["thread_id"]
        if ts := config["configurable"].get("thread_ts"):
            if saved := self.storage[thread_id].get(ts):
                checkpoint, metadata = saved
                return CheckpointTuple(
                    config=config,
                    checkpoint=self.serde.loads(checkpoint),
                    metadata=self.serde.loads(metadata),
                )
        else:
            if checkpoints := self.storage[thread_id]:
                ts = max(checkpoints.keys())
                checkpoint, metadata = checkpoints[ts]
                return CheckpointTuple(
                    config={"configurable": {"thread_id": thread_id, "thread_ts": ts}},
                    checkpoint=self.serde.loads(checkpoint),
                    metadata=self.serde.loads(metadata),
                )

    def list(
        self,
        config: RunnableConfig,
        *,
        before: Optional[RunnableConfig] = None,
        limit: Optional[int] = None,
    ) -> Iterator[CheckpointTuple]:
        """List checkpoints from the in-memory storage.

        This method retrieves a list of checkpoint tuples from the in-memory storage based
        on the provided config. The checkpoints are ordered by timestamp in descending order.

        Args:
            config (RunnableConfig): The config to use for listing the checkpoints.
            before (Optional[RunnableConfig]): If provided, only checkpoints before the specified timestamp are returned. Defaults to None.
            limit (Optional[int]): The maximum number of checkpoints to return. Defaults to None.

        Yields:
            Iterator[CheckpointTuple]: An iterator of checkpoint tuples.
        """
        thread_id = config["configurable"]["thread_id"]
        for ts, (checkpoint, metadata) in self.storage[thread_id].items():
            if before and ts >= before["configurable"]["thread_ts"]:
                continue
            if limit is not None and limit <= 0:
                break
            limit -= 1
            yield CheckpointTuple(
                config={"configurable": {"thread_id": thread_id, "thread_ts": ts}},
                checkpoint=self.serde.loads(checkpoint),
                metadata=self.serde.loads(metadata),
            )

    def put(
        self,
        config: RunnableConfig,
        checkpoint: Checkpoint,
        metadata: CheckpointMetadata,
    ) -> RunnableConfig:
        """Save a checkpoint to the in-memory storage.

        This method saves a checkpoint to the in-memory storage. The checkpoint is associated
        with the provided config.

        Args:
            config (RunnableConfig): The config to associate with the checkpoint.
            checkpoint (Checkpoint): The checkpoint to save.

        Returns:
            RunnableConfig: The updated config containing the saved checkpoint's timestamp.
        """
        self.storage[config["configurable"]["thread_id"]].update(
            {
                checkpoint["ts"]: (
                    self.serde.dumps(checkpoint),
                    self.serde.dumps(metadata),
                )
            }
        )
        return {
            "configurable": {
                "thread_id": config["configurable"]["thread_id"],
                "thread_ts": checkpoint["ts"],
            }
        }

    async def aget_tuple(self, config: RunnableConfig) -> Optional[CheckpointTuple]:
        """Asynchronous version of get_tuple.

        This method is an asynchronous wrapper around get_tuple that runs the synchronous
        method in a separate thread using asyncio.

        Args:
            config (RunnableConfig): The config to use for retrieving the checkpoint.

        Returns:
            Optional[CheckpointTuple]: The retrieved checkpoint tuple, or None if no matching checkpoint was found.
        """
        return await asyncio.get_running_loop().run_in_executor(
            None, self.get_tuple, config
        )

    async def alist(self, config: RunnableConfig) -> AsyncIterator[CheckpointTuple]:
        """Asynchronous version of list.

        This method is an asynchronous wrapper around list that runs the synchronous
        method in a separate thread using asyncio.

        Args:
            config (RunnableConfig): The config to use for listing the checkpoints.

        Yields:
            AsyncIterator[CheckpointTuple]: An asynchronous iterator of checkpoint tuples.
        """
        loop = asyncio.get_running_loop()
        iter = loop.run_in_executor(None, self.list, config)
        while True:
            try:
                yield await loop.run_in_executor(None, next, iter)
            except StopIteration:
                return

    async def aput(
        self,
        config: RunnableConfig,
        checkpoint: Checkpoint,
        metadata: CheckpointMetadata,
    ) -> RunnableConfig:
        return await asyncio.get_running_loop().run_in_executor(
            None, self.put, config, checkpoint, metadata
        )

get_tuple(config)

Get a checkpoint tuple from the in-memory storage.

This method retrieves a checkpoint tuple from the in-memory storage based on the provided config. If the config contains a "thread_ts" key, the checkpoint with the matching thread ID and timestamp is retrieved. Otherwise, the latest checkpoint for the given thread ID is retrieved.

Parameters:

  • config (RunnableConfig) –

    The config to use for retrieving the checkpoint.

Returns:

  • Optional[CheckpointTuple] –

    Optional[CheckpointTuple]: The retrieved checkpoint tuple, or None if no matching checkpoint was found.

Source code in langgraph/checkpoint/memory.py 
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
def get_tuple(self, config: RunnableConfig) -> Optional[CheckpointTuple]:
    """Get a checkpoint tuple from the in-memory storage.

    This method retrieves a checkpoint tuple from the in-memory storage based on the
    provided config. If the config contains a "thread_ts" key, the checkpoint with
    the matching thread ID and timestamp is retrieved. Otherwise, the latest checkpoint
    for the given thread ID is retrieved.

    Args:
        config (RunnableConfig): The config to use for retrieving the checkpoint.

    Returns:
        Optional[CheckpointTuple]: The retrieved checkpoint tuple, or None if no matching checkpoint was found.
    """
    thread_id = config["configurable"]["thread_id"]
    if ts := config["configurable"].get("thread_ts"):
        if saved := self.storage[thread_id].get(ts):
            checkpoint, metadata = saved
            return CheckpointTuple(
                config=config,
                checkpoint=self.serde.loads(checkpoint),
                metadata=self.serde.loads(metadata),
            )
    else:
        if checkpoints := self.storage[thread_id]:
            ts = max(checkpoints.keys())
            checkpoint, metadata = checkpoints[ts]
            return CheckpointTuple(
                config={"configurable": {"thread_id": thread_id, "thread_ts": ts}},
                checkpoint=self.serde.loads(checkpoint),
                metadata=self.serde.loads(metadata),
            )

list(config, *, before=None, limit=None)

List checkpoints from the in-memory storage.

This method retrieves a list of checkpoint tuples from the in-memory storage based on the provided config. The checkpoints are ordered by timestamp in descending order.

Parameters:

  • config (RunnableConfig) –

    The config to use for listing the checkpoints.

  • before (Optional[RunnableConfig], default: None ) –

    If provided, only checkpoints before the specified timestamp are returned. Defaults to None.

  • limit (Optional[int], default: None ) –

    The maximum number of checkpoints to return. Defaults to None.

Yields:

  • CheckpointTuple –

    Iterator[CheckpointTuple]: An iterator of checkpoint tuples.

Source code in langgraph/checkpoint/memory.py 
 89
 90
 91
 92
 93
 94
 95
 96
 97
 98
 99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
def list(
    self,
    config: RunnableConfig,
    *,
    before: Optional[RunnableConfig] = None,
    limit: Optional[int] = None,
) -> Iterator[CheckpointTuple]:
    """List checkpoints from the in-memory storage.

    This method retrieves a list of checkpoint tuples from the in-memory storage based
    on the provided config. The checkpoints are ordered by timestamp in descending order.

    Args:
        config (RunnableConfig): The config to use for listing the checkpoints.
        before (Optional[RunnableConfig]): If provided, only checkpoints before the specified timestamp are returned. Defaults to None.
        limit (Optional[int]): The maximum number of checkpoints to return. Defaults to None.

    Yields:
        Iterator[CheckpointTuple]: An iterator of checkpoint tuples.
    """
    thread_id = config["configurable"]["thread_id"]
    for ts, (checkpoint, metadata) in self.storage[thread_id].items():
        if before and ts >= before["configurable"]["thread_ts"]:
            continue
        if limit is not None and limit <= 0:
            break
        limit -= 1
        yield CheckpointTuple(
            config={"configurable": {"thread_id": thread_id, "thread_ts": ts}},
            checkpoint=self.serde.loads(checkpoint),
            metadata=self.serde.loads(metadata),
        )

put(config, checkpoint, metadata)

Save a checkpoint to the in-memory storage.

This method saves a checkpoint to the in-memory storage. The checkpoint is associated with the provided config.

Parameters:

  • config (RunnableConfig) –

    The config to associate with the checkpoint.

  • checkpoint (Checkpoint) –

    The checkpoint to save.

Returns:

  • RunnableConfig ( RunnableConfig ) –

    The updated config containing the saved checkpoint's timestamp.

Source code in langgraph/checkpoint/memory.py 
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
def put(
    self,
    config: RunnableConfig,
    checkpoint: Checkpoint,
    metadata: CheckpointMetadata,
) -> RunnableConfig:
    """Save a checkpoint to the in-memory storage.

    This method saves a checkpoint to the in-memory storage. The checkpoint is associated
    with the provided config.

    Args:
        config (RunnableConfig): The config to associate with the checkpoint.
        checkpoint (Checkpoint): The checkpoint to save.

    Returns:
        RunnableConfig: The updated config containing the saved checkpoint's timestamp.
    """
    self.storage[config["configurable"]["thread_id"]].update(
        {
            checkpoint["ts"]: (
                self.serde.dumps(checkpoint),
                self.serde.dumps(metadata),
            )
        }
    )
    return {
        "configurable": {
            "thread_id": config["configurable"]["thread_id"],
            "thread_ts": checkpoint["ts"],
        }
    }

aget_tuple(config) async

Asynchronous version of get_tuple.

This method is an asynchronous wrapper around get_tuple that runs the synchronous method in a separate thread using asyncio.

Parameters:

  • config (RunnableConfig) –

    The config to use for retrieving the checkpoint.

Returns:

  • Optional[CheckpointTuple] –

    Optional[CheckpointTuple]: The retrieved checkpoint tuple, or None if no matching checkpoint was found.

Source code in langgraph/checkpoint/memory.py 
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
async def aget_tuple(self, config: RunnableConfig) -> Optional[CheckpointTuple]:
    """Asynchronous version of get_tuple.

    This method is an asynchronous wrapper around get_tuple that runs the synchronous
    method in a separate thread using asyncio.

    Args:
        config (RunnableConfig): The config to use for retrieving the checkpoint.

    Returns:
        Optional[CheckpointTuple]: The retrieved checkpoint tuple, or None if no matching checkpoint was found.
    """
    return await asyncio.get_running_loop().run_in_executor(
        None, self.get_tuple, config
    )

alist(config) async

Asynchronous version of list.

This method is an asynchronous wrapper around list that runs the synchronous method in a separate thread using asyncio.

Parameters:

  • config (RunnableConfig) –

    The config to use for listing the checkpoints.

Yields:

  • AsyncIterator[CheckpointTuple] –

    AsyncIterator[CheckpointTuple]: An asynchronous iterator of checkpoint tuples.

Source code in langgraph/checkpoint/memory.py 
171
172
173
174
175
176
177
178
179
180
181
182
183
184
185
186
187
188
189
async def alist(self, config: RunnableConfig) -> AsyncIterator[CheckpointTuple]:
    """Asynchronous version of list.

    This method is an asynchronous wrapper around list that runs the synchronous
    method in a separate thread using asyncio.

    Args:
        config (RunnableConfig): The config to use for listing the checkpoints.

    Yields:
        AsyncIterator[CheckpointTuple]: An asynchronous iterator of checkpoint tuples.
    """
    loop = asyncio.get_running_loop()
    iter = loop.run_in_executor(None, self.list, config)
    while True:
        try:
            yield await loop.run_in_executor(None, next, iter)
        except StopIteration:
            return

AsyncSqliteSaver

Bases: BaseCheckpointSaver, AbstractAsyncContextManager

An asynchronous checkpoint saver that stores checkpoints in a SQLite database.

Tip

Requires the aiosqlite package. Install it with pip install aiosqlite.

Note

While this class does support asynchronous checkpointing, it is not recommended for production workloads, due to limitations in SQLite's write performance. For production workloads, consider using a more robust database like PostgreSQL.

Parameters:

  • conn (Connection) –

    The asynchronous SQLite database connection.

  • serde (Optional[SerializerProtocol], default: None ) –

    The serializer to use for serializing and deserializing checkpoints. Defaults to JsonPlusSerializerCompat.

Examples:

Usage within a StateGraph:

    import asyncio
    import aiosqlite

    from langgraph.checkpoint.aiosqlite import AsyncSqliteSaver
    from langgraph.graph import StateGraph

    builder = StateGraph(int)
    builder.add_node("add_one", lambda x: x + 1)
    builder.set_entry_point("add_one")
    builder.set_finish_point("add_one")

    memory = AsyncSqliteSaver.from_conn_string("checkpoints.sqlite")
    graph = builder.compile(checkpointer=memory)
    coro = graph.ainvoke(1, {"configurable": {"thread_id": "thread-1"}})
    asyncio.run(coro)  # Output: 2


Raw usage:

    import asyncio
    import aiosqlite
    from langgraph.checkpoint.aiosqlite import AsyncSqliteSaver


    async def main():
        async with aiosqlite.connect("checkpoints.db") as conn:
            saver = AsyncSqliteSaver(conn)
            config = {"configurable": {"thread_id": "1"}}
            checkpoint = {"ts": "2023-05-03T10:00:00Z", "data": {"key": "value"}}
            saved_config = await saver.aput(config, checkpoint)
            print(
                saved_config
            )  # Output: {"configurable": {"thread_id": "1", "thread_ts": "2023-05-03T10:00:00Z"}}


    asyncio.run(main())
Source code in langgraph/checkpoint/aiosqlite.py 
 20
 21
 22
 23
 24
 25
 26
 27
 28
 29
 30
 31
 32
 33
 34
 35
 36
 37
 38
 39
 40
 41
 42
 43
 44
 45
 46
 47
 48
 49
 50
 51
 52
 53
 54
 55
 56
 57
 58
 59
 60
 61
 62
 63
 64
 65
 66
 67
 68
 69
 70
 71
 72
 73
 74
 75
 76
 77
 78
 79
 80
 81
 82
 83
 84
 85
 86
 87
 88
 89
 90
 91
 92
 93
 94
 95
 96
 97
 98
 99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
173
174
175
176
177
178
179
180
181
182
183
184
185
186
187
188
189
190
191
192
193
194
195
196
197
198
199
200
201
202
203
204
205
206
207
208
209
210
211
212
213
214
215
216
217
218
219
220
221
222
223
224
225
226
227
228
229
230
231
232
233
234
235
236
237
238
239
240
241
242
243
244
245
246
247
248
249
250
251
252
253
254
255
256
257
258
259
260
261
262
263
264
265
266
267
268
269
270
271
272
273
274
275
276
277
278
279
280
281
282
283
284
285
286
287
288
289
290
291
292
293
class AsyncSqliteSaver(BaseCheckpointSaver, AbstractAsyncContextManager):
    """An asynchronous checkpoint saver that stores checkpoints in a SQLite database.

    Tip:
        Requires the [aiosqlite](https://pypi.org/project/aiosqlite/) package.
        Install it with `pip install aiosqlite`.

    Note:
        While this class does support asynchronous checkpointing, it is not recommended
        for production workloads, due to limitations in SQLite's write performance. For
        production workloads, consider using a more robust database like PostgreSQL.

    Args:
        conn (aiosqlite.Connection): The asynchronous SQLite database connection.
        serde (Optional[SerializerProtocol]): The serializer to use for serializing and deserializing checkpoints. Defaults to JsonPlusSerializerCompat.

    Examples:

        Usage within a StateGraph:

            import asyncio
            import aiosqlite

            from langgraph.checkpoint.aiosqlite import AsyncSqliteSaver
            from langgraph.graph import StateGraph

            builder = StateGraph(int)
            builder.add_node("add_one", lambda x: x + 1)
            builder.set_entry_point("add_one")
            builder.set_finish_point("add_one")

            memory = AsyncSqliteSaver.from_conn_string("checkpoints.sqlite")
            graph = builder.compile(checkpointer=memory)
            coro = graph.ainvoke(1, {"configurable": {"thread_id": "thread-1"}})
            asyncio.run(coro)  # Output: 2


        Raw usage:

            import asyncio
            import aiosqlite
            from langgraph.checkpoint.aiosqlite import AsyncSqliteSaver


            async def main():
                async with aiosqlite.connect("checkpoints.db") as conn:
                    saver = AsyncSqliteSaver(conn)
                    config = {"configurable": {"thread_id": "1"}}
                    checkpoint = {"ts": "2023-05-03T10:00:00Z", "data": {"key": "value"}}
                    saved_config = await saver.aput(config, checkpoint)
                    print(
                        saved_config
                    )  # Output: {"configurable": {"thread_id": "1", "thread_ts": "2023-05-03T10:00:00Z"}}


            asyncio.run(main())
    """

    serde = JsonPlusSerializerCompat()

    conn: aiosqlite.Connection
    lock: asyncio.Lock
    is_setup: bool

    def __init__(
        self,
        conn: aiosqlite.Connection,
        *,
        serde: Optional[SerializerProtocol] = None,
    ):
        super().__init__(serde=serde)
        self.conn = conn
        self.lock = asyncio.Lock()
        self.is_setup = False

    @classmethod
    def from_conn_string(cls, conn_string: str) -> "AsyncSqliteSaver":
        """Create a new AsyncSqliteSaver instance from a connection string.

        Args:
            conn_string (str): The SQLite connection string.

        Returns:
            AsyncSqliteSaver: A new AsyncSqliteSaver instance.
        """
        return AsyncSqliteSaver(conn=aiosqlite.connect(conn_string))

    async def __aenter__(self) -> Self:
        return self

    async def __aexit__(
        self,
        __exc_type: Optional[type[BaseException]],
        __exc_value: Optional[BaseException],
        __traceback: Optional[TracebackType],
    ) -> Optional[bool]:
        if self.is_setup:
            return await self.conn.close()

    async def setup(self) -> None:
        """Set up the checkpoint database asynchronously.

        This method creates the necessary tables in the SQLite database if they don't
        already exist. It is called automatically when needed and should not be called
        directly by the user.
        """
        async with self.lock:
            if self.is_setup:
                return
            if not self.conn.is_alive():
                await self.conn
            async with self.conn.executescript(
                """
                CREATE TABLE IF NOT EXISTS checkpoints (
                    thread_id TEXT NOT NULL,
                    thread_ts TEXT NOT NULL,
                    parent_ts TEXT,
                    checkpoint BLOB,
                    metadata BLOB,
                    PRIMARY KEY (thread_id, thread_ts)
                );
                """
            ):
                await self.conn.commit()

            self.is_setup = True

    async def aget_tuple(self, config: RunnableConfig) -> Optional[CheckpointTuple]:
        """Get a checkpoint tuple from the database asynchronously.

        This method retrieves a checkpoint tuple from the SQLite database based on the
        provided config. If the config contains a "thread_ts" key, the checkpoint with
        the matching thread ID and timestamp is retrieved. Otherwise, the latest checkpoint
        for the given thread ID is retrieved.

        Args:
            config (RunnableConfig): The config to use for retrieving the checkpoint.

        Returns:
            Optional[CheckpointTuple]: The retrieved checkpoint tuple, or None if no matching checkpoint was found.
        """
        await self.setup()
        if config["configurable"].get("thread_ts"):
            async with self.conn.execute(
                "SELECT checkpoint, parent_ts, metadata FROM checkpoints WHERE thread_id = ? AND thread_ts = ?",
                (
                    str(config["configurable"]["thread_id"]),
                    str(config["configurable"]["thread_ts"]),
                ),
            ) as cursor:
                if value := await cursor.fetchone():
                    return CheckpointTuple(
                        config,
                        self.serde.loads(value[0]),
                        self.serde.loads(value[2]) if value[2] is not None else {},
                        {
                            "configurable": {
                                "thread_id": config["configurable"]["thread_id"],
                                "thread_ts": value[1],
                            }
                        }
                        if value[1]
                        else None,
                    )
        else:
            async with self.conn.execute(
                "SELECT thread_id, thread_ts, parent_ts, checkpoint, metadata FROM checkpoints WHERE thread_id = ? ORDER BY thread_ts DESC LIMIT 1",
                (str(config["configurable"]["thread_id"]),),
            ) as cursor:
                if value := await cursor.fetchone():
                    return CheckpointTuple(
                        {
                            "configurable": {
                                "thread_id": value[0],
                                "thread_ts": value[1],
                            }
                        },
                        self.serde.loads(value[3]),
                        self.serde.loads(value[4]) if value[4] is not None else {},
                        {
                            "configurable": {
                                "thread_id": value[0],
                                "thread_ts": value[2],
                            }
                        }
                        if value[2]
                        else None,
                    )

    async def alist(
        self,
        config: RunnableConfig,
        *,
        before: Optional[RunnableConfig] = None,
        limit: Optional[int] = None,
    ) -> AsyncIterator[CheckpointTuple]:
        """List checkpoints from the database asynchronously.

        This method retrieves a list of checkpoint tuples from the SQLite database based
        on the provided config. The checkpoints are ordered by timestamp in descending order.

        Args:
            config (RunnableConfig): The config to use for listing the checkpoints.
            before (Optional[RunnableConfig]): If provided, only checkpoints before the specified timestamp are returned. Defaults to None.
            limit (Optional[int]): The maximum number of checkpoints to return. Defaults to None.

        Yields:
            AsyncIterator[CheckpointTuple]: An asynchronous iterator of checkpoint tuples.
        """
        await self.setup()
        query = (
            "SELECT thread_id, thread_ts, parent_ts, checkpoint, metadata FROM checkpoints WHERE thread_id = ? ORDER BY thread_ts DESC"
            if before is None
            else "SELECT thread_id, thread_ts, parent_ts, checkpoint, metadata FROM checkpoints WHERE thread_id = ? AND thread_ts < ? ORDER BY thread_ts DESC"
        )
        if limit:
            query += f" LIMIT {limit}"
        async with self.conn.execute(
            query,
            (
                (str(config["configurable"]["thread_id"]),)
                if before is None
                else (
                    str(config["configurable"]["thread_id"]),
                    str(before["configurable"]["thread_ts"]),
                )
            ),
        ) as cursor:
            async for thread_id, thread_ts, parent_ts, value, metadata in cursor:
                yield CheckpointTuple(
                    {"configurable": {"thread_id": thread_id, "thread_ts": thread_ts}},
                    self.serde.loads(value),
                    self.serde.loads(metadata) if metadata is not None else {},
                    {"configurable": {"thread_id": thread_id, "thread_ts": parent_ts}}
                    if parent_ts
                    else None,
                )

    async def aput(
        self,
        config: RunnableConfig,
        checkpoint: Checkpoint,
        metadata: CheckpointMetadata,
    ) -> RunnableConfig:
        """Save a checkpoint to the database asynchronously.

        This method saves a checkpoint to the SQLite database. The checkpoint is associated
        with the provided config and its parent config (if any).

        Args:
            config (RunnableConfig): The config to associate with the checkpoint.
            checkpoint (Checkpoint): The checkpoint to save.

        Returns:
            RunnableConfig: The updated config containing the saved checkpoint's timestamp.
        """
        await self.setup()
        async with self.conn.execute(
            "INSERT OR REPLACE INTO checkpoints (thread_id, thread_ts, parent_ts, checkpoint, metadata) VALUES (?, ?, ?, ?, ?)",
            (
                str(config["configurable"]["thread_id"]),
                checkpoint["ts"],
                config["configurable"].get("thread_ts"),
                self.serde.dumps(checkpoint),
                self.serde.dumps(metadata),
            ),
        ):
            await self.conn.commit()
        return {
            "configurable": {
                "thread_id": config["configurable"]["thread_id"],
                "thread_ts": checkpoint["ts"],
            }
        }

from_conn_string(conn_string) classmethod

Create a new AsyncSqliteSaver instance from a connection string.

Parameters:

  • conn_string (str) –

    The SQLite connection string.

Returns:

  • AsyncSqliteSaver ( AsyncSqliteSaver ) –

    A new AsyncSqliteSaver instance.

Source code in langgraph/checkpoint/aiosqlite.py 
 95
 96
 97
 98
 99
100
101
102
103
104
105
@classmethod
def from_conn_string(cls, conn_string: str) -> "AsyncSqliteSaver":
    """Create a new AsyncSqliteSaver instance from a connection string.

    Args:
        conn_string (str): The SQLite connection string.

    Returns:
        AsyncSqliteSaver: A new AsyncSqliteSaver instance.
    """
    return AsyncSqliteSaver(conn=aiosqlite.connect(conn_string))

setup() async

Set up the checkpoint database asynchronously.

This method creates the necessary tables in the SQLite database if they don't already exist. It is called automatically when needed and should not be called directly by the user.

Source code in langgraph/checkpoint/aiosqlite.py 
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
async def setup(self) -> None:
    """Set up the checkpoint database asynchronously.

    This method creates the necessary tables in the SQLite database if they don't
    already exist. It is called automatically when needed and should not be called
    directly by the user.
    """
    async with self.lock:
        if self.is_setup:
            return
        if not self.conn.is_alive():
            await self.conn
        async with self.conn.executescript(
            """
            CREATE TABLE IF NOT EXISTS checkpoints (
                thread_id TEXT NOT NULL,
                thread_ts TEXT NOT NULL,
                parent_ts TEXT,
                checkpoint BLOB,
                metadata BLOB,
                PRIMARY KEY (thread_id, thread_ts)
            );
            """
        ):
            await self.conn.commit()

        self.is_setup = True

aget_tuple(config) async

Get a checkpoint tuple from the database asynchronously.

This method retrieves a checkpoint tuple from the SQLite database based on the provided config. If the config contains a "thread_ts" key, the checkpoint with the matching thread ID and timestamp is retrieved. Otherwise, the latest checkpoint for the given thread ID is retrieved.

Parameters:

  • config (RunnableConfig) –

    The config to use for retrieving the checkpoint.

Returns:

  • Optional[CheckpointTuple] –

    Optional[CheckpointTuple]: The retrieved checkpoint tuple, or None if no matching checkpoint was found.

Source code in langgraph/checkpoint/aiosqlite.py 
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
173
174
175
176
177
178
179
180
181
182
183
184
185
186
187
188
189
190
191
192
193
194
195
196
197
198
199
200
201
202
203
204
205
206
207
async def aget_tuple(self, config: RunnableConfig) -> Optional[CheckpointTuple]:
    """Get a checkpoint tuple from the database asynchronously.

    This method retrieves a checkpoint tuple from the SQLite database based on the
    provided config. If the config contains a "thread_ts" key, the checkpoint with
    the matching thread ID and timestamp is retrieved. Otherwise, the latest checkpoint
    for the given thread ID is retrieved.

    Args:
        config (RunnableConfig): The config to use for retrieving the checkpoint.

    Returns:
        Optional[CheckpointTuple]: The retrieved checkpoint tuple, or None if no matching checkpoint was found.
    """
    await self.setup()
    if config["configurable"].get("thread_ts"):
        async with self.conn.execute(
            "SELECT checkpoint, parent_ts, metadata FROM checkpoints WHERE thread_id = ? AND thread_ts = ?",
            (
                str(config["configurable"]["thread_id"]),
                str(config["configurable"]["thread_ts"]),
            ),
        ) as cursor:
            if value := await cursor.fetchone():
                return CheckpointTuple(
                    config,
                    self.serde.loads(value[0]),
                    self.serde.loads(value[2]) if value[2] is not None else {},
                    {
                        "configurable": {
                            "thread_id": config["configurable"]["thread_id"],
                            "thread_ts": value[1],
                        }
                    }
                    if value[1]
                    else None,
                )
    else:
        async with self.conn.execute(
            "SELECT thread_id, thread_ts, parent_ts, checkpoint, metadata FROM checkpoints WHERE thread_id = ? ORDER BY thread_ts DESC LIMIT 1",
            (str(config["configurable"]["thread_id"]),),
        ) as cursor:
            if value := await cursor.fetchone():
                return CheckpointTuple(
                    {
                        "configurable": {
                            "thread_id": value[0],
                            "thread_ts": value[1],
                        }
                    },
                    self.serde.loads(value[3]),
                    self.serde.loads(value[4]) if value[4] is not None else {},
                    {
                        "configurable": {
                            "thread_id": value[0],
                            "thread_ts": value[2],
                        }
                    }
                    if value[2]
                    else None,
                )

alist(config, *, before=None, limit=None) async

List checkpoints from the database asynchronously.

This method retrieves a list of checkpoint tuples from the SQLite database based on the provided config. The checkpoints are ordered by timestamp in descending order.

Parameters:

  • config (RunnableConfig) –

    The config to use for listing the checkpoints.

  • before (Optional[RunnableConfig], default: None ) –

    If provided, only checkpoints before the specified timestamp are returned. Defaults to None.

  • limit (Optional[int], default: None ) –

    The maximum number of checkpoints to return. Defaults to None.

Yields:

  • AsyncIterator[CheckpointTuple] –

    AsyncIterator[CheckpointTuple]: An asynchronous iterator of checkpoint tuples.

Source code in langgraph/checkpoint/aiosqlite.py 
209
210
211
212
213
214
215
216
217
218
219
220
221
222
223
224
225
226
227
228
229
230
231
232
233
234
235
236
237
238
239
240
241
242
243
244
245
246
247
248
249
250
251
252
253
254
255
256
async def alist(
    self,
    config: RunnableConfig,
    *,
    before: Optional[RunnableConfig] = None,
    limit: Optional[int] = None,
) -> AsyncIterator[CheckpointTuple]:
    """List checkpoints from the database asynchronously.

    This method retrieves a list of checkpoint tuples from the SQLite database based
    on the provided config. The checkpoints are ordered by timestamp in descending order.

    Args:
        config (RunnableConfig): The config to use for listing the checkpoints.
        before (Optional[RunnableConfig]): If provided, only checkpoints before the specified timestamp are returned. Defaults to None.
        limit (Optional[int]): The maximum number of checkpoints to return. Defaults to None.

    Yields:
        AsyncIterator[CheckpointTuple]: An asynchronous iterator of checkpoint tuples.
    """
    await self.setup()
    query = (
        "SELECT thread_id, thread_ts, parent_ts, checkpoint, metadata FROM checkpoints WHERE thread_id = ? ORDER BY thread_ts DESC"
        if before is None
        else "SELECT thread_id, thread_ts, parent_ts, checkpoint, metadata FROM checkpoints WHERE thread_id = ? AND thread_ts < ? ORDER BY thread_ts DESC"
    )
    if limit:
        query += f" LIMIT {limit}"
    async with self.conn.execute(
        query,
        (
            (str(config["configurable"]["thread_id"]),)
            if before is None
            else (
                str(config["configurable"]["thread_id"]),
                str(before["configurable"]["thread_ts"]),
            )
        ),
    ) as cursor:
        async for thread_id, thread_ts, parent_ts, value, metadata in cursor:
            yield CheckpointTuple(
                {"configurable": {"thread_id": thread_id, "thread_ts": thread_ts}},
                self.serde.loads(value),
                self.serde.loads(metadata) if metadata is not None else {},
                {"configurable": {"thread_id": thread_id, "thread_ts": parent_ts}}
                if parent_ts
                else None,
            )

aput(config, checkpoint, metadata) async

Save a checkpoint to the database asynchronously.

This method saves a checkpoint to the SQLite database. The checkpoint is associated with the provided config and its parent config (if any).

Parameters:

  • config (RunnableConfig) –

    The config to associate with the checkpoint.

  • checkpoint (Checkpoint) –

    The checkpoint to save.

Returns:

  • RunnableConfig ( RunnableConfig ) –

    The updated config containing the saved checkpoint's timestamp.

Source code in langgraph/checkpoint/aiosqlite.py 
258
259
260
261
262
263
264
265
266
267
268
269
270
271
272
273
274
275
276
277
278
279
280
281
282
283
284
285
286
287
288
289
290
291
292
293
async def aput(
    self,
    config: RunnableConfig,
    checkpoint: Checkpoint,
    metadata: CheckpointMetadata,
) -> RunnableConfig:
    """Save a checkpoint to the database asynchronously.

    This method saves a checkpoint to the SQLite database. The checkpoint is associated
    with the provided config and its parent config (if any).

    Args:
        config (RunnableConfig): The config to associate with the checkpoint.
        checkpoint (Checkpoint): The checkpoint to save.

    Returns:
        RunnableConfig: The updated config containing the saved checkpoint's timestamp.
    """
    await self.setup()
    async with self.conn.execute(
        "INSERT OR REPLACE INTO checkpoints (thread_id, thread_ts, parent_ts, checkpoint, metadata) VALUES (?, ?, ?, ?, ?)",
        (
            str(config["configurable"]["thread_id"]),
            checkpoint["ts"],
            config["configurable"].get("thread_ts"),
            self.serde.dumps(checkpoint),
            self.serde.dumps(metadata),
        ),
    ):
        await self.conn.commit()
    return {
        "configurable": {
            "thread_id": config["configurable"]["thread_id"],
            "thread_ts": checkpoint["ts"],
        }
    }

SqliteSaver

Bases: BaseCheckpointSaver, AbstractContextManager

A checkpoint saver that stores checkpoints in a SQLite database.

Note

This class is meant for lightweight, synchronous use cases (demos and small projects) and does not scale to multiple threads. For a similar sqlite saver with async support, consider using AsyncSqliteSaver.

Parameters:

  • conn (Connection) –

    The SQLite database connection.

  • serde (Optional[SerializerProtocol], default: None ) –

    The serializer to use for serializing and deserializing checkpoints. Defaults to JsonPlusSerializerCompat.

Examples:

import sqlite3

from langgraph.checkpoint.sqlite import SqliteSaver
from langgraph.graph import StateGraph

builder = StateGraph(int)
builder.add_node("add_one", lambda x: x + 1)
builder.set_entry_point("add_one")
builder.set_finish_point("add_one")
conn = sqlite3.connect("checkpoints.sqlite")
memory = SqliteSaver(conn)
graph = builder.compile(checkpointer=memory)

config = {"configurable": {"thread_id": "1"}}
# checkpoint = {"ts": "2023-05-03T10:00:00Z", "data": {"key": "value"}}
result = graph.invoke(3, config)
graph.get_state(config)
# Output: StateSnapshot(values=4, next=(), config={'configurable': {'thread_id': '1', 'thread_ts': '2024-05-04T06:32:42.235444+00:00'}}, parent_config=None)
Source code in langgraph/checkpoint/sqlite.py 
 51
 52
 53
 54
 55
 56
 57
 58
 59
 60
 61
 62
 63
 64
 65
 66
 67
 68
 69
 70
 71
 72
 73
 74
 75
 76
 77
 78
 79
 80
 81
 82
 83
 84
 85
 86
 87
 88
 89
 90
 91
 92
 93
 94
 95
 96
 97
 98
 99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
173
174
175
176
177
178
179
180
181
182
183
184
185
186
187
188
189
190
191
192
193
194
195
196
197
198
199
200
201
202
203
204
205
206
207
208
209
210
211
212
213
214
215
216
217
218
219
220
221
222
223
224
225
226
227
228
229
230
231
232
233
234
235
236
237
238
239
240
241
242
243
244
245
246
247
248
249
250
251
252
253
254
255
256
257
258
259
260
261
262
263
264
265
266
267
268
269
270
271
272
273
274
275
276
277
278
279
280
281
282
283
284
285
286
287
288
289
290
291
292
293
294
295
296
297
298
299
300
301
302
303
304
305
306
307
308
309
310
311
312
313
314
315
316
317
318
319
320
321
322
323
324
325
326
327
328
329
330
331
332
333
334
335
336
337
338
339
340
341
342
343
344
345
346
347
348
349
350
351
352
353
354
355
356
357
358
359
360
361
362
363
364
365
366
367
368
369
370
371
372
373
374
375
376
377
378
379
380
381
382
383
384
class SqliteSaver(BaseCheckpointSaver, AbstractContextManager):
    """A checkpoint saver that stores checkpoints in a SQLite database.

    Note:
        This class is meant for lightweight, synchronous use cases
        (demos and small projects) and does not
        scale to multiple threads.
        For a similar sqlite saver with `async` support,
        consider using AsyncSqliteSaver.

    Args:
        conn (sqlite3.Connection): The SQLite database connection.
        serde (Optional[SerializerProtocol]): The serializer to use for serializing and deserializing checkpoints. Defaults to JsonPlusSerializerCompat.

    Examples:

        import sqlite3

        from langgraph.checkpoint.sqlite import SqliteSaver
        from langgraph.graph import StateGraph

        builder = StateGraph(int)
        builder.add_node("add_one", lambda x: x + 1)
        builder.set_entry_point("add_one")
        builder.set_finish_point("add_one")
        conn = sqlite3.connect("checkpoints.sqlite")
        memory = SqliteSaver(conn)
        graph = builder.compile(checkpointer=memory)

        config = {"configurable": {"thread_id": "1"}}
        # checkpoint = {"ts": "2023-05-03T10:00:00Z", "data": {"key": "value"}}
        result = graph.invoke(3, config)
        graph.get_state(config)
        # Output: StateSnapshot(values=4, next=(), config={'configurable': {'thread_id': '1', 'thread_ts': '2024-05-04T06:32:42.235444+00:00'}}, parent_config=None)
    """  # noqa

    serde = JsonPlusSerializerCompat()

    conn: sqlite3.Connection
    is_setup: bool

    def __init__(
        self,
        conn: sqlite3.Connection,
        *,
        serde: Optional[SerializerProtocol] = None,
    ) -> None:
        super().__init__(serde=serde)
        self.conn = conn
        self.is_setup = False
        self.lock = threading.Lock()

    @classmethod
    def from_conn_string(cls, conn_string: str) -> "SqliteSaver":
        """Create a new SqliteSaver instance from a connection string.

        Args:
            conn_string (str): The SQLite connection string.

        Returns:
            SqliteSaver: A new SqliteSaver instance.

        Examples:

            In memory:

                memory = SqliteSaver.from_conn_string(":memory:")

            To disk:

                memory = SqliteSaver.from_conn_string("checkpoints.sqlite")
        """
        return SqliteSaver(
            conn=sqlite3.connect(
                conn_string,
                # https://ricardoanderegg.com/posts/python-sqlite-thread-safety/
                check_same_thread=False,
            )
        )

    def __enter__(self) -> Self:
        return self

    def __exit__(
        self,
        __exc_type: Optional[type[BaseException]],
        __exc_value: Optional[BaseException],
        __traceback: Optional[TracebackType],
    ) -> Optional[bool]:
        return self.conn.close()

    def setup(self) -> None:
        """Set up the checkpoint database.

        This method creates the necessary tables in the SQLite database if they don't
        already exist. It is called automatically when needed and should not be called
        directly by the user.
        """
        if self.is_setup:
            return

        self.conn.executescript(
            """
            CREATE TABLE IF NOT EXISTS checkpoints (
                thread_id TEXT NOT NULL,
                thread_ts TEXT NOT NULL,
                parent_ts TEXT,
                checkpoint BLOB,
                metadata BLOB,
                PRIMARY KEY (thread_id, thread_ts)
            );
            """
        )

        self.is_setup = True

    @contextmanager
    def cursor(self, transaction: bool = True) -> Iterator[sqlite3.Cursor]:
        """Get a cursor for the SQLite database.

        This method returns a cursor for the SQLite database. It is used internally
        by the SqliteSaver and should not be called directly by the user.

        Args:
            transaction (bool): Whether to commit the transaction when the cursor is closed. Defaults to True.

        Yields:
            sqlite3.Cursor: A cursor for the SQLite database.
        """
        self.setup()
        cur = self.conn.cursor()
        try:
            yield cur
        finally:
            if transaction:
                self.conn.commit()
            cur.close()

    def get_tuple(self, config: RunnableConfig) -> Optional[CheckpointTuple]:
        """Get a checkpoint tuple from the database.

        This method retrieves a checkpoint tuple from the SQLite database based on the
        provided config. If the config contains a "thread_ts" key, the checkpoint with
        the matching thread ID and timestamp is retrieved. Otherwise, the latest checkpoint
        for the given thread ID is retrieved.

        Args:
            config (RunnableConfig): The config to use for retrieving the checkpoint.

        Returns:
            Optional[CheckpointTuple]: The retrieved checkpoint tuple, or None if no matching checkpoint was found.

        Examples:

            Basic:

                config = {"configurable": {"thread_id": "1"}}
                checkpoint_tuple = memory.get_tuple(config)
                print(checkpoint_tuple)  # Output: CheckpointTuple(...)

            With timestamp:

                config = {
                    "configurable": {
                        "thread_id": "1",
                        "thread_ts": "2024-05-04T06:32:42.235444+00:00",
                    }
                }
                checkpoint_tuple = memory.get_tuple(config)
                print(checkpoint_tuple)  # Output: CheckpointTuple(...)
        """  # noqa
        with self.cursor(transaction=False) as cur:
            if config["configurable"].get("thread_ts"):
                cur.execute(
                    "SELECT checkpoint, parent_ts, metadata FROM checkpoints WHERE thread_id = ? AND thread_ts = ?",
                    (
                        str(config["configurable"]["thread_id"]),
                        str(config["configurable"]["thread_ts"]),
                    ),
                )
                if value := cur.fetchone():
                    return CheckpointTuple(
                        config,
                        self.serde.loads(value[0]),
                        self.serde.loads(value[2]) if value[2] is not None else {},
                        (
                            {
                                "configurable": {
                                    "thread_id": config["configurable"]["thread_id"],
                                    "thread_ts": value[1],
                                }
                            }
                            if value[1]
                            else None
                        ),
                    )
            else:
                cur.execute(
                    "SELECT thread_id, thread_ts, parent_ts, checkpoint, metadata FROM checkpoints WHERE thread_id = ? ORDER BY thread_ts DESC LIMIT 1",
                    (str(config["configurable"]["thread_id"]),),
                )
                if value := cur.fetchone():
                    return CheckpointTuple(
                        {
                            "configurable": {
                                "thread_id": value[0],
                                "thread_ts": value[1],
                            }
                        },
                        self.serde.loads(value[3]),
                        self.serde.loads(value[4]) if value[4] is not None else {},
                        (
                            {
                                "configurable": {
                                    "thread_id": value[0],
                                    "thread_ts": value[2],
                                }
                            }
                            if value[2]
                            else None
                        ),
                    )

    def list(
        self,
        config: RunnableConfig,
        *,
        before: Optional[RunnableConfig] = None,
        limit: Optional[int] = None,
    ) -> Iterator[CheckpointTuple]:
        """List checkpoints from the database.

        This method retrieves a list of checkpoint tuples from the SQLite database based
        on the provided config. The checkpoints are ordered by timestamp in descending order.

        Args:
            config (RunnableConfig): The config to use for listing the checkpoints.
            before (Optional[RunnableConfig]): If provided, only checkpoints before the specified timestamp are returned. Defaults to None.
            limit (Optional[int]): The maximum number of checkpoints to return. Defaults to None.

        Yields:
            Iterator[CheckpointTuple]: An iterator of checkpoint tuples.

        Examples:
                config = {"configurable": {"thread_id": "1"}}
                checkpoints = list(memory.list(config, limit=2))
                print(checkpoints)  # Output: [CheckpointTuple(...), CheckpointTuple(...)]

                config = {"configurable": {"thread_id": "1"}}
                before = {"configurable": {"thread_ts": "2024-05-04T06:32:42.235444+00:00"}}
                checkpoints = list(memory.list(config, before=before))
                print(checkpoints)  # Output: [CheckpointTuple(...), ...]
        """
        query = (
            "SELECT thread_id, thread_ts, parent_ts, checkpoint, metadata FROM checkpoints WHERE thread_id = ? ORDER BY thread_ts DESC"
            if before is None
            else "SELECT thread_id, thread_ts, parent_ts, checkpoint, metadata FROM checkpoints WHERE thread_id = ? AND thread_ts < ? ORDER BY thread_ts DESC"
        )
        if limit:
            query += f" LIMIT {limit}"
        with self.cursor(transaction=False) as cur:
            cur.execute(
                query,
                (
                    (str(config["configurable"]["thread_id"]),)
                    if before is None
                    else (
                        str(config["configurable"]["thread_id"]),
                        before["configurable"]["thread_ts"],
                    )
                ),
            )
            for thread_id, thread_ts, parent_ts, value, metadata in cur:
                yield CheckpointTuple(
                    {"configurable": {"thread_id": thread_id, "thread_ts": thread_ts}},
                    self.serde.loads(value),
                    self.serde.loads(metadata) if metadata is not None else {},
                    (
                        {
                            "configurable": {
                                "thread_id": thread_id,
                                "thread_ts": parent_ts,
                            }
                        }
                        if parent_ts
                        else None
                    ),
                )

    def put(
        self,
        config: RunnableConfig,
        checkpoint: Checkpoint,
        metadata: CheckpointMetadata,
    ) -> RunnableConfig:
        """Save a checkpoint to the database.

        This method saves a checkpoint to the SQLite database. The checkpoint is associated
        with the provided config and its parent config (if any).

        Args:
            config (RunnableConfig): The config to associate with the checkpoint.
            checkpoint (Checkpoint): The checkpoint to save.
            metadata (Optional[dict[str, Any]]): Additional metadata to save with the checkpoint. Defaults to None.

        Returns:
            RunnableConfig: The updated config containing the saved checkpoint's timestamp.

        Examples:

                config = {"configurable": {"thread_id": "1"}}
                checkpoint = {"ts": "2024-05-04T06:32:42.235444+00:00", "data": {"key": "value"}}
                saved_config = memory.put(config, checkpoint)
                print(
                    saved_config
                )  # Output: {"configurable": {"thread_id": "1", "thread_ts": 2024-05-04T06:32:42.235444+00:00"}}
        """
        with self.lock, self.cursor() as cur:
            cur.execute(
                "INSERT OR REPLACE INTO checkpoints (thread_id, thread_ts, parent_ts, checkpoint, metadata) VALUES (?, ?, ?, ?, ?)",
                (
                    str(config["configurable"]["thread_id"]),
                    checkpoint["ts"],
                    config["configurable"].get("thread_ts"),
                    self.serde.dumps(checkpoint),
                    self.serde.dumps(metadata),
                ),
            )
        return {
            "configurable": {
                "thread_id": config["configurable"]["thread_id"],
                "thread_ts": checkpoint["ts"],
            }
        }

from_conn_string(conn_string) classmethod

Create a new SqliteSaver instance from a connection string.

Parameters:

  • conn_string (str) –

    The SQLite connection string.

Returns:

  • SqliteSaver ( SqliteSaver ) –

    A new SqliteSaver instance.

Examples:

In memory:

    memory = SqliteSaver.from_conn_string(":memory:")

To disk:

    memory = SqliteSaver.from_conn_string("checkpoints.sqlite")
Source code in langgraph/checkpoint/sqlite.py 
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
@classmethod
def from_conn_string(cls, conn_string: str) -> "SqliteSaver":
    """Create a new SqliteSaver instance from a connection string.

    Args:
        conn_string (str): The SQLite connection string.

    Returns:
        SqliteSaver: A new SqliteSaver instance.

    Examples:

        In memory:

            memory = SqliteSaver.from_conn_string(":memory:")

        To disk:

            memory = SqliteSaver.from_conn_string("checkpoints.sqlite")
    """
    return SqliteSaver(
        conn=sqlite3.connect(
            conn_string,
            # https://ricardoanderegg.com/posts/python-sqlite-thread-safety/
            check_same_thread=False,
        )
    )

setup()

Set up the checkpoint database.

This method creates the necessary tables in the SQLite database if they don't already exist. It is called automatically when needed and should not be called directly by the user.

Source code in langgraph/checkpoint/sqlite.py 
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
def setup(self) -> None:
    """Set up the checkpoint database.

    This method creates the necessary tables in the SQLite database if they don't
    already exist. It is called automatically when needed and should not be called
    directly by the user.
    """
    if self.is_setup:
        return

    self.conn.executescript(
        """
        CREATE TABLE IF NOT EXISTS checkpoints (
            thread_id TEXT NOT NULL,
            thread_ts TEXT NOT NULL,
            parent_ts TEXT,
            checkpoint BLOB,
            metadata BLOB,
            PRIMARY KEY (thread_id, thread_ts)
        );
        """
    )

    self.is_setup = True

cursor(transaction=True)

Get a cursor for the SQLite database.

This method returns a cursor for the SQLite database. It is used internally by the SqliteSaver and should not be called directly by the user.

Parameters:

  • transaction (bool, default: True ) –

    Whether to commit the transaction when the cursor is closed. Defaults to True.

Yields:

  • Cursor –

    sqlite3.Cursor: A cursor for the SQLite database.

Source code in langgraph/checkpoint/sqlite.py 
167
168
169
170
171
172
173
174
175
176
177
178
179
180
181
182
183
184
185
186
187
@contextmanager
def cursor(self, transaction: bool = True) -> Iterator[sqlite3.Cursor]:
    """Get a cursor for the SQLite database.

    This method returns a cursor for the SQLite database. It is used internally
    by the SqliteSaver and should not be called directly by the user.

    Args:
        transaction (bool): Whether to commit the transaction when the cursor is closed. Defaults to True.

    Yields:
        sqlite3.Cursor: A cursor for the SQLite database.
    """
    self.setup()
    cur = self.conn.cursor()
    try:
        yield cur
    finally:
        if transaction:
            self.conn.commit()
        cur.close()

get_tuple(config)

Get a checkpoint tuple from the database.

This method retrieves a checkpoint tuple from the SQLite database based on the provided config. If the config contains a "thread_ts" key, the checkpoint with the matching thread ID and timestamp is retrieved. Otherwise, the latest checkpoint for the given thread ID is retrieved.

Parameters:

  • config (RunnableConfig) –

    The config to use for retrieving the checkpoint.

Returns:

  • Optional[CheckpointTuple] –

    Optional[CheckpointTuple]: The retrieved checkpoint tuple, or None if no matching checkpoint was found.

Examples:

Basic:

    config = {"configurable": {"thread_id": "1"}}
    checkpoint_tuple = memory.get_tuple(config)
    print(checkpoint_tuple)  # Output: CheckpointTuple(...)

With timestamp:

    config = {
        "configurable": {
            "thread_id": "1",
            "thread_ts": "2024-05-04T06:32:42.235444+00:00",
        }
    }
    checkpoint_tuple = memory.get_tuple(config)
    print(checkpoint_tuple)  # Output: CheckpointTuple(...)
Source code in langgraph/checkpoint/sqlite.py 
189
190
191
192
193
194
195
196
197
198
199
200
201
202
203
204
205
206
207
208
209
210
211
212
213
214
215
216
217
218
219
220
221
222
223
224
225
226
227
228
229
230
231
232
233
234
235
236
237
238
239
240
241
242
243
244
245
246
247
248
249
250
251
252
253
254
255
256
257
258
259
260
261
262
263
264
265
266
267
268
269
270
271
272
def get_tuple(self, config: RunnableConfig) -> Optional[CheckpointTuple]:
    """Get a checkpoint tuple from the database.

    This method retrieves a checkpoint tuple from the SQLite database based on the
    provided config. If the config contains a "thread_ts" key, the checkpoint with
    the matching thread ID and timestamp is retrieved. Otherwise, the latest checkpoint
    for the given thread ID is retrieved.

    Args:
        config (RunnableConfig): The config to use for retrieving the checkpoint.

    Returns:
        Optional[CheckpointTuple]: The retrieved checkpoint tuple, or None if no matching checkpoint was found.

    Examples:

        Basic:

            config = {"configurable": {"thread_id": "1"}}
            checkpoint_tuple = memory.get_tuple(config)
            print(checkpoint_tuple)  # Output: CheckpointTuple(...)

        With timestamp:

            config = {
                "configurable": {
                    "thread_id": "1",
                    "thread_ts": "2024-05-04T06:32:42.235444+00:00",
                }
            }
            checkpoint_tuple = memory.get_tuple(config)
            print(checkpoint_tuple)  # Output: CheckpointTuple(...)
    """  # noqa
    with self.cursor(transaction=False) as cur:
        if config["configurable"].get("thread_ts"):
            cur.execute(
                "SELECT checkpoint, parent_ts, metadata FROM checkpoints WHERE thread_id = ? AND thread_ts = ?",
                (
                    str(config["configurable"]["thread_id"]),
                    str(config["configurable"]["thread_ts"]),
                ),
            )
            if value := cur.fetchone():
                return CheckpointTuple(
                    config,
                    self.serde.loads(value[0]),
                    self.serde.loads(value[2]) if value[2] is not None else {},
                    (
                        {
                            "configurable": {
                                "thread_id": config["configurable"]["thread_id"],
                                "thread_ts": value[1],
                            }
                        }
                        if value[1]
                        else None
                    ),
                )
        else:
            cur.execute(
                "SELECT thread_id, thread_ts, parent_ts, checkpoint, metadata FROM checkpoints WHERE thread_id = ? ORDER BY thread_ts DESC LIMIT 1",
                (str(config["configurable"]["thread_id"]),),
            )
            if value := cur.fetchone():
                return CheckpointTuple(
                    {
                        "configurable": {
                            "thread_id": value[0],
                            "thread_ts": value[1],
                        }
                    },
                    self.serde.loads(value[3]),
                    self.serde.loads(value[4]) if value[4] is not None else {},
                    (
                        {
                            "configurable": {
                                "thread_id": value[0],
                                "thread_ts": value[2],
                            }
                        }
                        if value[2]
                        else None
                    ),
                )

list(config, *, before=None, limit=None)

List checkpoints from the database.

This method retrieves a list of checkpoint tuples from the SQLite database based on the provided config. The checkpoints are ordered by timestamp in descending order.

Parameters:

  • config (RunnableConfig) –

    The config to use for listing the checkpoints.

  • before (Optional[RunnableConfig], default: None ) –

    If provided, only checkpoints before the specified timestamp are returned. Defaults to None.

  • limit (Optional[int], default: None ) –

    The maximum number of checkpoints to return. Defaults to None.

Yields:

  • CheckpointTuple –

    Iterator[CheckpointTuple]: An iterator of checkpoint tuples.

Examples:

config = {"configurable": {"thread_id": "1"}} checkpoints = list(memory.list(config, limit=2)) print(checkpoints) # Output: [CheckpointTuple(...), CheckpointTuple(...)]

config = {"configurable": {"thread_id": "1"}} before = {"configurable": {"thread_ts": "2024-05-04T06:32:42.235444+00:00"}} checkpoints = list(memory.list(config, before=before)) print(checkpoints) # Output: [CheckpointTuple(...), ...]

Source code in langgraph/checkpoint/sqlite.py 
274
275
276
277
278
279
280
281
282
283
284
285
286
287
288
289
290
291
292
293
294
295
296
297
298
299
300
301
302
303
304
305
306
307
308
309
310
311
312
313
314
315
316
317
318
319
320
321
322
323
324
325
326
327
328
329
330
331
332
333
334
335
336
337
338
def list(
    self,
    config: RunnableConfig,
    *,
    before: Optional[RunnableConfig] = None,
    limit: Optional[int] = None,
) -> Iterator[CheckpointTuple]:
    """List checkpoints from the database.

    This method retrieves a list of checkpoint tuples from the SQLite database based
    on the provided config. The checkpoints are ordered by timestamp in descending order.

    Args:
        config (RunnableConfig): The config to use for listing the checkpoints.
        before (Optional[RunnableConfig]): If provided, only checkpoints before the specified timestamp are returned. Defaults to None.
        limit (Optional[int]): The maximum number of checkpoints to return. Defaults to None.

    Yields:
        Iterator[CheckpointTuple]: An iterator of checkpoint tuples.

    Examples:
            config = {"configurable": {"thread_id": "1"}}
            checkpoints = list(memory.list(config, limit=2))
            print(checkpoints)  # Output: [CheckpointTuple(...), CheckpointTuple(...)]

            config = {"configurable": {"thread_id": "1"}}
            before = {"configurable": {"thread_ts": "2024-05-04T06:32:42.235444+00:00"}}
            checkpoints = list(memory.list(config, before=before))
            print(checkpoints)  # Output: [CheckpointTuple(...), ...]
    """
    query = (
        "SELECT thread_id, thread_ts, parent_ts, checkpoint, metadata FROM checkpoints WHERE thread_id = ? ORDER BY thread_ts DESC"
        if before is None
        else "SELECT thread_id, thread_ts, parent_ts, checkpoint, metadata FROM checkpoints WHERE thread_id = ? AND thread_ts < ? ORDER BY thread_ts DESC"
    )
    if limit:
        query += f" LIMIT {limit}"
    with self.cursor(transaction=False) as cur:
        cur.execute(
            query,
            (
                (str(config["configurable"]["thread_id"]),)
                if before is None
                else (
                    str(config["configurable"]["thread_id"]),
                    before["configurable"]["thread_ts"],
                )
            ),
        )
        for thread_id, thread_ts, parent_ts, value, metadata in cur:
            yield CheckpointTuple(
                {"configurable": {"thread_id": thread_id, "thread_ts": thread_ts}},
                self.serde.loads(value),
                self.serde.loads(metadata) if metadata is not None else {},
                (
                    {
                        "configurable": {
                            "thread_id": thread_id,
                            "thread_ts": parent_ts,
                        }
                    }
                    if parent_ts
                    else None
                ),
            )

put(config, checkpoint, metadata)

Save a checkpoint to the database.

This method saves a checkpoint to the SQLite database. The checkpoint is associated with the provided config and its parent config (if any).

Parameters:

  • config (RunnableConfig) –

    The config to associate with the checkpoint.

  • checkpoint (Checkpoint) –

    The checkpoint to save.

  • metadata (Optional[dict[str, Any]]) –

    Additional metadata to save with the checkpoint. Defaults to None.

Returns:

  • RunnableConfig ( RunnableConfig ) –

    The updated config containing the saved checkpoint's timestamp.

Examples:

    config = {"configurable": {"thread_id": "1"}}
    checkpoint = {"ts": "2024-05-04T06:32:42.235444+00:00", "data": {"key": "value"}}
    saved_config = memory.put(config, checkpoint)
    print(
        saved_config
    )  # Output: {"configurable": {"thread_id": "1", "thread_ts": 2024-05-04T06:32:42.235444+00:00"}}
Source code in langgraph/checkpoint/sqlite.py 
340
341
342
343
344
345
346
347
348
349
350
351
352
353
354
355
356
357
358
359
360
361
362
363
364
365
366
367
368
369
370
371
372
373
374
375
376
377
378
379
380
381
382
383
384
def put(
    self,
    config: RunnableConfig,
    checkpoint: Checkpoint,
    metadata: CheckpointMetadata,
) -> RunnableConfig:
    """Save a checkpoint to the database.

    This method saves a checkpoint to the SQLite database. The checkpoint is associated
    with the provided config and its parent config (if any).

    Args:
        config (RunnableConfig): The config to associate with the checkpoint.
        checkpoint (Checkpoint): The checkpoint to save.
        metadata (Optional[dict[str, Any]]): Additional metadata to save with the checkpoint. Defaults to None.

    Returns:
        RunnableConfig: The updated config containing the saved checkpoint's timestamp.

    Examples:

            config = {"configurable": {"thread_id": "1"}}
            checkpoint = {"ts": "2024-05-04T06:32:42.235444+00:00", "data": {"key": "value"}}
            saved_config = memory.put(config, checkpoint)
            print(
                saved_config
            )  # Output: {"configurable": {"thread_id": "1", "thread_ts": 2024-05-04T06:32:42.235444+00:00"}}
    """
    with self.lock, self.cursor() as cur:
        cur.execute(
            "INSERT OR REPLACE INTO checkpoints (thread_id, thread_ts, parent_ts, checkpoint, metadata) VALUES (?, ?, ?, ?, ?)",
            (
                str(config["configurable"]["thread_id"]),
                checkpoint["ts"],
                config["configurable"].get("thread_ts"),
                self.serde.dumps(checkpoint),
                self.serde.dumps(metadata),
            ),
        )
    return {
        "configurable": {
            "thread_id": config["configurable"]["thread_id"],
            "thread_ts": checkpoint["ts"],
        }
    }