LangGraph with Checkpointing – A Must for Production-Ready Graphs

If you're already building projects using LangGraph, then you must know about checkpointing — a powerful feature for production systems that gives your entire pipeline the ability to persist and restore graph state. In simpler terms, checkpointing allows your graph to pause and resume exactly from where it left off.

Why Does Checkpointing Exist?

Let’s first understand the problem and what’s lacking in a basic LangGraph project.

When you're implementing the business logic of a graph, it eventually gets integrated into your application. Each user triggers their own instance of graph execution, and the state transitions from one node to another depending on your logic.

Now here’s the tricky part — it’s entirely possible that a node is waiting for a human response (e.g., payment confirmation or OTP input). This wait time is indefinite.

So what do you do? You need to save the user’s graph instance — that is, the current state — and restore it later so that execution resumes exactly from where it was paused.

This is where Checkpointing steps in.

Checkpointing solves the problem of saving the current state of the graph. Since the state is stored in databases, LangGraph supports databases like PostgreSQL, MongoDB, and more to store and restore graph state.

How to Integrate Checkpointing in LangGraph?

We'll use 3 files:

• graph.py: Graph logic (states, nodes, edges)

• main.py: Executes the graph with a checkpointer

• docker-compose.yml: To run MongoDB using Docker

Note: I'm using the ChatGPT API key here, but you can swap in Gemini if you'd like. Do not forgot to add .env file.

graph.py

# Imports 
from typing import Annotated
from typing_extensions import TypedDict
from langgraph.graph.message import add_messages
from langchain.chat_models import init_chat_model  # for using the LLM
from langgraph.graph import StateGraph, START, END
from langchain_core.tools import tool  # Adding the tool in our Graph
from langgraph.types import interrupt
from langgraph.prebuilt import ToolNode, tools_condition  # creating the tool node , edge

@tool()
def human_assistance_tool(query: str):
    """Request assistance from a human."""
    human_response = interrupt({ "query": query })  # Graph will exit out after saving data in DB
    return human_response["data"]  # resume with the data

tools = [human_assistance_tool]

# llm integration 
llm = init_chat_model(model_provider="openai", model="gpt-4.1")
llm_with_tools = llm.bind_tools(tools=tools)  # Ingesting tool in our LLM 

# Building the Graph 
class State(TypedDict):
    messages: Annotated[list, add_messages]

def chatbot(state: State):
    message = llm_with_tools.invoke(state["messages"])
    assert len(message.tool_calls) <= 1  # check normal call or tool call
    return {"messages": [message]}

tool_node = ToolNode(tools=tools)

graph_builder = StateGraph(State)

graph_builder.add_node("chatbot", chatbot)
graph_builder.add_node("tools", tool_node)

graph_builder.add_edge(START, "chatbot")

graph_builder.add_conditional_edges(
    "chatbot",
    tools_condition,
)
graph_builder.add_edge("tools", "chatbot") 
graph_builder.add_edge("chatbot", END)

# Option 1: Without memory or checkpointing (default)
graph = graph_builder.compile()

# Option 2: Creates a new graph with checkpointing
def create_graph_checkpoint(checkpointer):
    return graph_builder.compile(checkpointer=checkpointer)

---

main.py

from graph import create_graph_checkpoint
from dotenv import load_dotenv
from langgraph.checkpoint.mongodb import MongoDBSaver

load_dotenv()

MONGODB_URI = "mongodb://admin:admin@localhost:27017"
config = {"configurable": {"thread_id": "1"}}  # Help you to uniquely identify each user graph instance

def init():
    with MongoDBSaver.from_conn_string(MONGODB_URI) as checkpointer:  # we will stream on graph
        graph_with_mongo = create_graph_checkpoint(checkpointer=checkpointer)

        while True:
            user_input = input("> ")
            for event in graph_with_mongo.stream(
                {"messages": [{"role": "user", "content": user_input}]},
                config,
                stream_mode="values"
            ):
                if "messages" in event:
                    event["messages"][-1].pretty_print()

init()

---

docker-compose.yml

services:
  mongodb:
    image: mongo
    restart: always
    ports:
      - "27017:27017"
    environment:
      - MONGO_INITDB_ROOT_USERNAME=admin
      - MONGO_INITDB_ROOT_PASSWORD=admin
    volumes:
      - mongodb_data:/data/db

volumes:
  mongodb_data:

---

Final Thoughts

With this setup, your graph can now pause execution (e.g., when waiting for a human response), persist the state in MongoDB, and resume later without restarting the entire graph. This is especially useful in long-running workflows, production pipelines, or human-in-the-loop systems.

Now your LangGraph is production-grade, fault-tolerant, and much more reliable!