LangGraph SQLite Checkpoint

Warnings

• gotcha SQLite's file-based nature limits its concurrency. For multi-user or high-concurrency applications (e.g., web services), direct use of `SQLiteSaver` can lead to database locking issues, performance bottlenecks, or even data corruption. Consider using a dedicated database like PostgreSQL (`langgraph-checkpoint-postgres`) or an external service for production environments.
  Fix: For high-concurrency applications, switch to `langgraph-checkpoint-postgres`, `langgraph-checkpoint-redis`, or a custom `BaseCheckpointSaver` implementation backed by a robust database.
• gotcha When initializing `SQLiteSaver.from_file(file_path)`, ensure the directory specified in `file_path` exists and is writable by the application. If the directory does not exist, an `OSError` or `sqlite3.OperationalError` might occur.
  Fix: Before calling `SQLiteSaver.from_file()`, ensure the parent directory of `file_path` exists, e.g., by using `os.makedirs(os.path.dirname(file_path), exist_ok=True)`.
• breaking The `langgraph` library underwent significant API changes with its 1.0 release. While `langgraph-checkpoint-sqlite` versions are generally compatible with `langgraph>=0.0.1` (which includes 1.x.x), users migrating from older pre-1.0 `langgraph` applications may find that the overall `StateGraph` and checkpointing interface has changed, requiring updates to their graph definitions and invocation patterns.
  Fix: Refer to the official LangGraph 1.0 migration guides and updated documentation for the new `StateGraph` and checkpointing API. Ensure your `langgraph` version is 1.0 or higher.

Install

• pip install langgraph-checkpoint-sqlite Install stable version

Imports

• SQLiteSaver

  from langgraph.checkpoint.sqlite import SQLiteSaver

Quickstart

This quickstart demonstrates how to set up `SQLiteSaver` with a basic LangGraph `StateGraph`. It creates a simple graph, initializes `SQLiteSaver` to a local file, and then compiles the graph with the checkpointer. Subsequent invocations with the same `thread_id` in the configuration will automatically load and resume the graph's state from the SQLite database.

import os
from typing import Annotated, TypedDict
import operator

from langgraph.graph import StateGraph, START
from langgraph.checkpoint.sqlite import SQLiteSaver

# Define a simple state for our graph
class AgentState(TypedDict):
    messages: Annotated[list, operator.add]
    turn: int

# Define a node function
def my_node(state: AgentState):
    print(f"Executing my_node, current turn: {state.get('turn', 0)}")
    return {"messages": [f"Hello from node in turn {state.get('turn', 0)}!"], "turn": state.get('turn', 0) + 1}

# Build the graph
workflow = StateGraph(AgentState)
workflow.add_node("step_one", my_node)
workflow.set_entry_point(START)
workflow.set_finish_point("step_one")

# Initialize SQLiteSaver
# Ensure the directory exists or create it. For this example, we'll use a local file.
sqlite_file = "./langgraph_checkpoints.sqlite"
memory = SQLiteSaver.from_file(sqlite_file)

# Compile the graph with the checkpointer
app = workflow.compile(checkpointer=memory)

# Invoke the graph with a thread_id to save state
config = {"configurable": {"thread_id": "my_first_thread"}}

print("\n--- First Invocation ---")
# The initial state passed here will be merged with any loaded state for 'my_first_thread'
initial_state = {"messages": ["User: Start conversation"], "turn": 0}
app.invoke(initial_state, config=config)

print("\n--- Second Invocation (resuming thread) ---")
# Invoke again with the same thread_id; it should load the previous state.
# We don't need to pass initial_state here to resume.
app.invoke(None, config=config)

print("\n--- Third Invocation (resuming thread) ---")
app.invoke(None, config=config)

# Clean up the SQLite file for demonstration purposes (optional)
# os.remove(sqlite_file)