How to persist LangGraph thread state with checkpoints

Graph executions often span several interactions rather than finish in one call. LangGraph checkpointers attach saved graph-state snapshots to a thread ID, allowing a later invocation in that thread to continue from the latest checkpoint.

The in-memory saver is appropriate for local development and small examples because it needs no external service. Its checkpoints disappear when the Python process stops, so production applications that must survive restarts should use a persistent saver such as SQLite or PostgreSQL.

Each invocation must include a stable thread_id in the graph configuration. Reuse it for related calls and choose another value for independent work; get_state() can then read the latest checkpoint for either thread.

1. Create thread_state.py with the event state schema and counter node.

   thread_state.py

   from operator import add
   from typing import Annotated
    
   from langgraph.checkpoint.memory import InMemorySaver
   from langgraph.graph import START, StateGraph
   from typing_extensions import NotRequired, TypedDict
    
    
   class ThreadState(TypedDict):
       events: Annotated[list[str], add]
       event_count: NotRequired[int]
    
    
   def count_events(state: ThreadState) -> dict[str, int]:
       return {"event_count": len(state["events"])}

   The add reducer appends new events values to the list already stored for the active thread.

2. Add the graph definition after the count_events() function.

   builder = StateGraph(ThreadState)
   builder.add_node("count_events", count_events)
   builder.add_edge(START, "count_events")
   graph = builder.compile(checkpointer=InMemorySaver())

   InMemorySaver keeps checkpoints only for the lifetime of this Python process; persistent checkpointers retain state across application restarts.

3. Add the two thread configurations and checkpoint checks after the graph assignment.

   customer_thread = {"configurable": {"thread_id": "customer-42"}}
   support_thread = {"configurable": {"thread_id": "support-7"}}
    
   graph.invoke({"events": ["cart created"]}, customer_thread)
   graph.invoke({"events": ["payment received"]}, customer_thread)
   graph.invoke({"events": ["ticket opened"]}, support_thread)
    
   customer_state = graph.get_state(customer_thread).values
   support_state = graph.get_state(support_thread).values
    
   assert customer_state["events"] == ["cart created", "payment received"]
   assert customer_state["event_count"] == 2
   assert support_state["events"] == ["ticket opened"]
   assert support_state["event_count"] == 1
    
   print(
       f"customer-42: {customer_state['events']} "
       f"({customer_state['event_count']} events)"
   )
   print(
       f"support-7: {support_state['events']} "
       f"({support_state['event_count']} event)"
   )

   The sample thread IDs represent stable application identifiers such as a conversation ID or workflow-run ID.

4. Run thread_state.py to verify state continuation and thread isolation.

   $ python3 thread_state.py
   customer-42: ['cart created', 'payment received'] (2 events)
   support-7: ['ticket opened'] (1 event)

   The assertions stop the script before it prints output if the first thread loses an event or the second thread receives state from the first.