Skip to content

langgraph-checkpoint-scylladb

Directory actions

More options

Directory actions

More options

Latest commit

 

History

History

langgraph-checkpoint-scylladb

Folders and files

README.md

LangGraph Checkpoint ScyllaDB

Implementation of a LangGraph CheckpointSaver that uses ScyllaDB.

Installation

pip install langgraph-checkpoint-scylladb

Usage

Sync

from cassandra.auth import PlainTextAuthProvider
from cassandra.cluster import Cluster, ExecutionProfile, EXEC_PROFILE_DEFAULT
from cassandra.policies import DCAwareRoundRobinPolicy
from langgraph.checkpoint.scylladb import ScyllaDBSaver

profile = ExecutionProfile(
    load_balancing_policy=DCAwareRoundRobinPolicy(local_dc="AWS_US_EAST_1"),
)
cluster = Cluster(
    contact_points=["node-0.example.scylla.cloud"],
    port=9042,
    auth_provider=PlainTextAuthProvider("scylla", "secret"),
    execution_profiles={EXEC_PROFILE_DEFAULT: profile},
)
session = cluster.connect()

checkpointer = ScyllaDBSaver(session, keyspace="langgraph")
checkpointer.setup()

write_config = {"configurable": {"thread_id": "1", "checkpoint_ns": ""}}
read_config  = {"configurable": {"thread_id": "1"}}

checkpointer.put(write_config, {...}, {}, {})
checkpointer.get(read_config)
list(checkpointer.list(read_config))

Async — from_conn_string

from langgraph.checkpoint.scylladb import ScyllaDBSaver

SCYLLADB_URI = "scylladb://scylla:secret@node-0.example.scylla.cloud:9042/langgraph?dc=AWS_US_EAST_1"

async with ScyllaDBSaver.from_conn_string(SCYLLADB_URI) as checkpointer:
    await checkpointer.aput(write_config, {...}, {}, {})
    await checkpointer.aget(read_config)
    [c async for c in checkpointer.alist(read_config)]

The connection string format is:

scylladb://<user>:<password>@<host>:<port>/<keyspace>?dc=<datacenter>

The dc query parameter is required for ScyllaDB Cloud (enables DC-aware load balancing).

TTL

Checkpoints can be automatically expired by passing ttl (seconds):

checkpointer = ScyllaDBSaver(session, keyspace="langgraph", ttl=3600)

# or via from_conn_string
async with ScyllaDBSaver.from_conn_string(SCYLLADB_URI, ttl=3600) as checkpointer:
    ...

Schema

Two tables are created in the target keyspace:

Table Partition key Clustering key
checkpoints thread_id checkpoint_ns ASC, checkpoint_id DESC
checkpoint_writes (thread_id, checkpoint_ns, checkpoint_id) task_id, idx

The schema is intentionally query-first: every access pattern hits a full partition key with no ALLOW FILTERING.

Additional Links