Guide
Use Stolon as LangGraph's 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
pip install langgraph langgraph-checkpoint-postgres "psycopg[binary]"
Make a branch for it
psql "postgresql://app:<password>@db.stolondb.com/acme.main" -c "CREATE DATABASE agents"
Use the stock saver
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")
{'steps': ['plan', 'act']}
4 checkpoints
AsyncPostgresSaver works the same way.
Fork every thread at once
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)
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.