pip install statebase langgraph langchain-openai

from statebase import StateBase
from langgraph.graph import StateGraph, START, END

sb = StateBase(api_key="your-key")

class StateBaseCheckpointer:
    """Minimal durable checkpointer backed by StateBase sessions."""

    def __init__(self, session_id):
        self.session_id = session_id

    def get(self, thread_id):
        state = sb.sessions.get(session_id=self.session_id).state
        return state.get("graph", {}).get(thread_id)

    def put(self, thread_id, graph_state):
        state = sb.sessions.get(session_id=self.session_id).state
        graph = state.get("graph", {})
        graph[thread_id] = graph_state
        sb.sessions.update_state(
            session_id=self.session_id,
            state={"graph": graph},
            reasoning=f"LangGraph checkpoint for {thread_id}"
        )
        return graph_state


# Build a simple graph
def research(state):
    return {"research": f"Research for: {state['topic']}"}


def summarize(state):
    return {"summary": f"Summary of {state['research']}"}


builder = StateGraph(dict)
builder.add_node("research", research)
builder.add_node("summarize", summarize)
builder.add_edge(START, "research")
builder.add_edge("research", "summarize")
builder.add_edge("summarize", END)

graph = builder.compile(checkpointer=StateBaseCheckpointer("langgraph-prod"))

config = {"configurable": {"thread_id": "thread-1"}}

# First run — crashes are recoverable at any node
for event in graph.stream({"topic": "durable execution"}, config):
    print(event)

# Resume after a crash: LangGraph replays from the last StateBase checkpoint
for event in graph.stream(None, config):
    print(event)

turns = sb.sessions.list_turns(session_id="langgraph-prod")
for turn in turns:
    print(turn.reasoning)