langgraph-checkpoint-sqlite

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)