AsyncNeo4jSaver(
self,
driver: AsyncDriver,
database: str | None = None
)| Name | Type | Description |
|---|---|---|
driver* | AsyncDriver | An async Neo4j Driver instance. |
database | str | None | Default: NoneOptional database name (defaults to Neo4j default). |
| Name | Type |
|---|---|
| driver | AsyncDriver |
| database | str | None |
Asynchronous Neo4j checkpoint saver for LangGraph.
This class implements the BaseCheckpointSaver interface using Neo4j as the persistence backend with async support and a proper graph model. It supports storing checkpoints, channel states, and pending writes using relationships for efficient traversal.
Graph Model:
(:Thread)-[:HAS_CHECKPOINT]->(:Checkpoint)-[:PREVIOUS]->(:Checkpoint) (:Checkpoint)-[:HAS_CHANNEL]->(:ChannelState) (:Checkpoint)-[:HAS_WRITE]->(:PendingWrite)
Example:
Using from_conn_string (recommended)
async with await AsyncNeo4jSaver.from_conn_string( ... uri="bolt://localhost:7687", ... user="neo4j", ... password="password" ... ) as checkpointer: ... await checkpointer.setup() # Create indexes (run once) ... graph = builder.compile(checkpointer=checkpointer) ... result = await graph.ainvoke({"messages": [...]}, config)
Using existing async driver
driver = AsyncGraphDatabase.driver(uri, auth=(user, password)) checkpointer = AsyncNeo4jSaver(driver) await checkpointer.setup()