langgraph-persistence

Durable graph execution with thread-scoped checkpoints, state history, and cross-thread long-term memory. Two memory types: class State(TypedDict): messages: Annotated[list, operator.add] def add_message(state: State) -> dict: return {"messages": ["Bot response"]} checkpointer = InMemorySaver() graph = ( StateGraph(State) .add_node("respond", add_message) .add_edge(START, "respond") .add_edge("respond", END) .compile(checkpointer=checkpointer) # Pass at compile time ) config = {"configurable": {"thread_id": "conversation-1"}} result1 = graph.invoke({"messages": ["Hello"]}, config) print(len(result1["messages"])) # 2 result2 = graph.invoke({"messages": ["How are you?"]}, config) print(len(result2["messages"])) # 4 (previous + new) with PostgresSaver.from_conn_string(os.environ["DATABASE_URL"]) as checkpointer: graph = builder.compile(checkpointer=checkpointer) graph.invoke({"messages": ["Hi from Alice"]}, alice_config) graph.invoke({"messages": ["Hi from Bob"]}, bob_config) result = graph.invoke({"messages": ["start"]}, config) states = list(graph.get_state_history(config)) past = states[-2] result = graph.invoke(None, past.config) # None = resume from checkpoint fork_config = graph.update_state(past.config, {"messages": ["edited"]}) result = graph.invoke(None, fork_config) graph.update_state(config, {"data": "manually_updated"}) result = graph.invoke(None, config) When...