LangGraph MemorySaver & SqliteSaver Setup Errors

import uuid
from langgraph.checkpoint.memory import MemorySaver
from langgraph.graph import StateGraph, START, END, MessagesState


def responder(state):
    reply = llm.invoke(state["messages"])
    return {"messages": [reply]}


graph = StateGraph(MessagesState)
graph.add_node("responder", responder)
graph.add_edge(START, "responder")
graph.add_edge("responder", END)

checkpointer = MemorySaver()
app = graph.compile(checkpointer=checkpointer)   # <-- attached at compile


# ✅ Always pass thread_id in config
thread_id = str(uuid.uuid4())               # or a user_id, session_id, etc.
config = {"configurable": {"thread_id": thread_id}}

result = app.invoke(
    {"messages": [HumanMessage(content="Hello")]},
    config,                                  # <-- required
)

# Same thread_id continues the conversation
result = app.invoke(
    {"messages": [HumanMessage(content="What did I just say?")]},
    config,                                  # ← same config, reads prior state
)


# ❌ WRONG — missing config, raises ValueError
# result = app.invoke({"messages": [...]})


# ✅ FastAPI pattern — thread_id from URL param
from fastapi import FastAPI
api = FastAPI()

@api.post("/chat/{session_id}")
async def chat(session_id: str, message: str):
    config = {"configurable": {"thread_id": session_id}}
    return app.invoke(
        {"messages": [HumanMessage(content=message)]},
        config,
    )

from pathlib import Path
from langgraph.checkpoint.sqlite import SqliteSaver
from langgraph.graph import StateGraph, START, END, MessagesState


# ✅ from_conn_string — auto-runs setup(), returns a context manager
DB_PATH = Path("/var/data/langgraph_checkpoints.db").as_posix()

with SqliteSaver.from_conn_string(DB_PATH) as checkpointer:
    graph = StateGraph(MessagesState)
    graph.add_node("responder", responder)
    graph.add_edge(START, "responder")
    graph.add_edge("responder", END)
    app = graph.compile(checkpointer=checkpointer)

    config = {"configurable": {"thread_id": "user-42"}}
    result = app.invoke({"messages": [HumanMessage(content="Hi")]}, config)


# ✅ For long-lived apps (FastAPI), open once at startup and keep the conn
import sqlite3
from contextlib import asynccontextmanager
from fastapi import FastAPI

CHECKPOINT_DB = "/var/data/langgraph.db"

@asynccontextmanager
async def lifespan(app: FastAPI):
    conn = sqlite3.connect(CHECKPOINT_DB, check_same_thread=False)
    checkpointer = SqliteSaver(conn)
    checkpointer.setup()                  # explicit setup for the raw constructor
    app.state.graph = build_graph(checkpointer)
    yield
    conn.close()

api = FastAPI(lifespan=lifespan)


# ❌ WRONG — using SqliteSaver constructor directly without setup()
# conn = sqlite3.connect("data.db")
# checkpointer = SqliteSaver(conn)         # no setup() call
# app = graph.compile(checkpointer=checkpointer)
# # First invoke raises sqlite3.OperationalError: no such table


# ✅ In-memory SQLite — perfect for tests
with SqliteSaver.from_conn_string(":memory:") as checkpointer:
    app = build_graph(checkpointer)
    # ... run tests ...
    # DB is gone when the `with` block exits — no cleanup needed

# ✅ Async SQLite — for FastAPI, aiohttp, or any asyncio app
from langgraph.checkpoint.sqlite.aio import AsyncSqliteSaver
from langgraph.graph import StateGraph, START, END, MessagesState


async def responder(state):
    reply = await llm.ainvoke(state["messages"])
    return {"messages": [reply]}


# from_conn_string is an ASYNC context manager here
async def run_chat():
    async with AsyncSqliteSaver.from_conn_string("/var/data/chat.db") as checkpointer:
        graph = StateGraph(MessagesState)
        graph.add_node("responder", responder)
        graph.add_edge(START, "responder")
        graph.add_edge("responder", END)
        app = graph.compile(checkpointer=checkpointer)

        config = {"configurable": {"thread_id": "user-42"}}
        result = await app.ainvoke(
            {"messages": [HumanMessage(content="Hello")]},
            config,
        )
        return result


# ✅ FastAPI lifespan pattern for long-lived app
from contextlib import asynccontextmanager
import aiosqlite
from fastapi import FastAPI

@asynccontextmanager
async def lifespan(app: FastAPI):
    conn = await aiosqlite.connect("/var/data/langgraph.db")
    checkpointer = AsyncSqliteSaver(conn)
    await checkpointer.setup()             # await, not call
    app.state.graph = build_graph(checkpointer)
    yield
    await conn.close()

api = FastAPI(lifespan=lifespan)


# ❌ Common mistake — sync saver in async handler
# from langgraph.checkpoint.sqlite import SqliteSaver
# @api.post("/chat")
# async def chat(msg: str):
#     conn = sqlite3.connect("data.db")
#     saver = SqliteSaver(conn); saver.setup()
#     app = build_graph(saver)
#     return await app.ainvoke(...)          # RuntimeError: sync checkpointer