checkpoint: add DuckDB checkpointer (#2145)

.github/workflows/ci.yml

@@ -31,6 +31,7 @@ jobs:
             "libs/cli",
             "libs/checkpoint",
             "libs/checkpoint-sqlite",
+            "libs/checkpoint-duckdb",
             "libs/checkpoint-postgres",
             "libs/scheduler-kafka",
           ]
@@ -47,6 +48,7 @@ jobs:
             "libs/cli",
             "libs/checkpoint",
             "libs/checkpoint-sqlite",
+            "libs/checkpoint-duckdb",
             "libs/checkpoint-postgres"
           ]
     uses: ./.github/workflows/_test.yml

libs/checkpoint-duckdb/Makefile

@@ -0,0 +1,35 @@
+.PHONY: test test_watch lint format
+
+######################
+# TESTING AND COVERAGE
+######################
+
+test:
+	poetry run pytest tests
+
+test_watch:
+	poetry run ptw .
+
+######################
+# LINTING AND FORMATTING
+######################
+
+# Define a variable for Python and notebook files.
+PYTHON_FILES=.
+MYPY_CACHE=.mypy_cache
+lint format: PYTHON_FILES=.
+lint_diff format_diff: PYTHON_FILES=$(shell git diff --name-only --relative --diff-filter=d main . | grep -E '\.py$$|\.ipynb$$')
+lint_package: PYTHON_FILES=langgraph
+lint_tests: PYTHON_FILES=tests
+lint_tests: MYPY_CACHE=.mypy_cache_test
+
+lint lint_diff lint_package lint_tests:
+	poetry run ruff check .
+	[ "$(PYTHON_FILES)" = "" ] || poetry run ruff format $(PYTHON_FILES) --diff
+	[ "$(PYTHON_FILES)" = "" ] || poetry run ruff check --select I $(PYTHON_FILES)
+	[ "$(PYTHON_FILES)" = "" ] || mkdir -p $(MYPY_CACHE)
+	[ "$(PYTHON_FILES)" = "" ] || poetry run mypy $(PYTHON_FILES) --cache-dir $(MYPY_CACHE)
+
+format format_diff:
+	poetry run ruff format $(PYTHON_FILES)
+	poetry run ruff check --select I --fix $(PYTHON_FILES)

libs/checkpoint-duckdb/README.md

@@ -0,0 +1,95 @@
+# LangGraph Checkpoint DuckDB
+
+Implementation of LangGraph CheckpointSaver that uses DuckDB.
+
+## Usage
+
+> [!IMPORTANT]
+> When using DuckDB checkpointers for the first time, make sure to call `.setup()` method on them to create required tables. See example below.
+
+```python
+from langgraph.checkpoint.duckdb import DuckDBSaver
+
+write_config = {"configurable": {"thread_id": "1", "checkpoint_ns": ""}}
+read_config = {"configurable": {"thread_id": "1"}}
+
+with DuckDBSaver.from_conn_string(":memory:") as checkpointer:
+    # call .setup() the first time you're using the checkpointer
+    checkpointer.setup()
+    checkpoint = {
+        "v": 1,
+        "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
+            }
+        },
+        "pending_sends": [],
+    }
+
+    # store checkpoint
+    checkpointer.put(write_config, checkpoint, {}, {})
+
+    # load checkpoint
+    checkpointer.get(read_config)
+
+    # list checkpoints
+    list(checkpointer.list(read_config))
+```
+
+### Async
+
+```python
+from langgraph.checkpoint.duckdb.aio import AsyncDuckDBSaver
+
+async with AsyncDuckDBSaver.from_conn_string(":memory:") as checkpointer:
+    checkpoint = {
+        "v": 1,
+        "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
+            }
+        },
+        "pending_sends": [],
+    }
+
+    # 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)]
+```