LangGraph checkpointer using Google Colab + Google Drive + SQLite

langgraph-colab-sqlite-checkpoint

# Commented out IPython magic to ensure Python compatibility.
# %pip install -q langgraph
# %pip install -q langgraph-checkpoint-sqlite
# Demo notebook: https://colab.research.google.com/drive/16V9TpOsLMOuU655ryerni7Hc7mni6hcU?usp=sharing

import os

from google.colab import drive
DRIVE_MOUNT_POINT = "/content/drive"
drive.mount(DRIVE_MOUNT_POINT)

os.listdir("/content/drive")

YOUR_DRIVE="Shareddrives/<change this value>"

CHECKPOINT_FOLDER=os.path.join(DRIVE_MOUNT_POINT, YOUR_DRIVE, "my_app_checkpoints")
print(CHECKPOINT_FOLDER)

from langgraph.checkpoint.sqlite import SqliteSaver
import sqlite3
os.makedirs(CHECKPOINT_FOLDER, exist_ok=True)
conn = sqlite3.connect(os.path.join(CHECKPOINT_FOLDER, "checkpoints.sqlite"), check_same_thread=False)
memory = SqliteSaver(conn)

from langgraph.graph import StateGraph, START, END
from typing import Annotated
from typing_extensions import TypedDict
from operator import add

class State(TypedDict):
    foo: str
    bar: Annotated[list[str], add]

def node_a(state: State):
    return {"foo": "a", "bar": ["a"]}

def node_b(state: State):
    return {"foo": "b", "bar": ["b"]}


workflow = StateGraph(State)
workflow.add_node(node_a)
workflow.add_node(node_b)
workflow.add_edge(START, "node_a")
workflow.add_edge("node_a", "node_b")
workflow.add_edge("node_b", END)

checkpointer = memory
graph = workflow.compile(checkpointer=checkpointer)

config = {"configurable": {"thread_id": "1"}}
graph.invoke({"foo": ""}, config, checkpoint_during=True)

os.listdir(CHECKPOINT_FOLDER)