mirror of
https://github.com/langchain-ai/langgraph.git
synced 2026-08-20 06:35:46 +02:00
2b1abc807b
## Summary - replace Python lint type-checking from mypy to ty across LangGraph packages - remove mypy config/cache wiring and mypy-only references - regenerate uv locks with ty 0.0.43 ## Verification - git diff --check - make lint_package && make lint_tests in libs/langgraph - make lint_package && make lint_tests in libs/checkpoint - make lint_package && make lint_tests in libs/checkpoint-sqlite - make lint_package && make lint_tests in libs/checkpoint-postgres - make lint_package && make lint_tests in libs/prebuilt - make lint_package && make lint_tests in libs/cli - make lint in libs/sdk-py - make lint in libs/checkpoint-conformance --------- Co-authored-by: Claude <noreply@anthropic.com> Co-authored-by: open-swe[bot] <215916821+open-swe[bot]@users.noreply.github.com>
LangGraph SQLite Checkpoint
Implementation of LangGraph CheckpointSaver that uses SQLite DB (both sync and async, via aiosqlite)
Security
Important
Set
LANGGRAPH_STRICT_MSGPACK=trueor pass an explicitallowed_msgpack_moduleslist when creating your checkpointer. This restricts checkpoint deserialization to known-safe types, preventing code execution if the database is compromised. See the langgraph-checkpoint README for details.
Usage
from langgraph.checkpoint.sqlite import SqliteSaver
write_config = {"configurable": {"thread_id": "1", "checkpoint_ns": ""}}
read_config = {"configurable": {"thread_id": "1"}}
with SqliteSaver.from_conn_string(":memory:") as checkpointer:
checkpoint = {
"v": 4,
"ts": "2024-07-31T20:14:19.804150+00:00",
"id": "1ef4f797-8335-6428-8001-8a1503f9b875",
"channel_values": {
"my_key": "meow",
"node": "node"
},
"channel_versions": {
"__start__": 2,
"my_key": 3,
"start:node": 3,
"node": 3
},
"versions_seen": {
"__input__": {},
"__start__": {
"__start__": 1
},
"node": {
"start:node": 2
}
},
}
# store checkpoint
checkpointer.put(write_config, checkpoint, {}, {})
# load checkpoint
checkpointer.get(read_config)
# list checkpoints
list(checkpointer.list(read_config))
Async
from langgraph.checkpoint.sqlite.aio import AsyncSqliteSaver
async with AsyncSqliteSaver.from_conn_string(":memory:") as checkpointer:
checkpoint = {
"v": 4,
"ts": "2024-07-31T20:14:19.804150+00:00",
"id": "1ef4f797-8335-6428-8001-8a1503f9b875",
"channel_values": {
"my_key": "meow",
"node": "node"
},
"channel_versions": {
"__start__": 2,
"my_key": 3,
"start:node": 3,
"node": 3
},
"versions_seen": {
"__input__": {},
"__start__": {
"__start__": 1
},
"node": {
"start:node": 2
}
},
}
# store checkpoint
await checkpointer.aput(write_config, checkpoint, {}, {})
# load checkpoint
await checkpointer.aget(read_config)
# list checkpoints
[c async for c in checkpointer.alist(read_config)]