# Use Stolon as LangGraph's checkpointer

> LangGraph's stock PostgresSaver and AsyncPostgresSaver (langgraph-checkpoint-postgres) run against Stolon unchanged. Point them at a branch, and fork every thread's checkpoints with one CREATE DATABASE.

Source: https://stolondb.com/guides/langgraph-checkpointer/ · Page title: Use Stolon as LangGraph's Postgres checkpointer

LangGraph's stock Postgres checkpointer runs unchanged. Point it at a branch, and fork every thread with one `CREATE DATABASE`.

*Run one saver process per checkpoint branch; give each worker process its own branch.*

## Install

```sh
pip install langgraph langgraph-checkpoint-postgres "psycopg[binary]"
```

## Make a branch for it

```sh
psql "postgresql://app:<password>@db.stolondb.com/acme.main" -c "CREATE DATABASE agents"
```

## Use the stock saver

```python
import operator
from typing import Annotated, TypedDict

from langgraph.checkpoint.postgres import PostgresSaver
from langgraph.graph import END, START, StateGraph

DB_URI = "postgresql://app:<password>@db.stolondb.com/acme.agents"  # the database name is the branch

class State(TypedDict):
    steps: Annotated[list, operator.add]

builder = StateGraph(State)
builder.add_node("plan", lambda state: {"steps": ["plan"]})
builder.add_node("act", lambda state: {"steps": ["act"]})
builder.add_edge(START, "plan")
builder.add_edge("plan", "act")
builder.add_edge("act", END)

with PostgresSaver.from_conn_string(DB_URI) as checkpointer:
    checkpointer.setup()  # creates LangGraph's own tables on this branch
    graph = builder.compile(checkpointer=checkpointer)
    config = {"configurable": {"thread_id": "thread-1"}}
    print(graph.invoke({"steps": []}, config))
    print(len(list(graph.get_state_history(config))), "checkpoints")
```

```text
{'steps': ['plan', 'act']}
4 checkpoints
```

`AsyncPostgresSaver` works the same way.

## Fork every thread at once

```python
import psycopg

with psycopg.connect("postgresql://app:<password>@db.stolondb.com/acme.agents", autocommit=True) as conn:
    conn.execute("CREATE DATABASE agents_try")

with PostgresSaver.from_conn_string("postgresql://app:<password>@db.stolondb.com/acme.agents_try") as cp:
    print("fork:", builder.compile(checkpointer=cp).invoke({"steps": ["retry"]}, config))

with PostgresSaver.from_conn_string(DB_URI) as cp:
    print("original:", builder.compile(checkpointer=cp).get_state(config).values)
```

```text
fork: {'steps': ['plan', 'act', 'retry', 'plan', 'act']}
original: {'steps': ['plan', 'act']}
```

Try a new prompt or tool across every thread on the copy, then `DROP DATABASE` or [`MERGE DATABASE`](https://stolondb.com/guides/review-agent-changes/).