How the Open Notebook Chat Workflow Persists Message History

The Open Notebook chat workflow persists message history using a LangGraph checkpoint backed by SQLite, automatically storing conversation state after each model invocation.

The lfnovo/open-notebook repository implements a durable chat system that retains conversation context across sessions without requiring an external database server. By leveraging LangGraph's checkpointing capabilities, the chat workflow persists message history in a local SQLite file, enabling seamless continuity for multi-turn conversations. This architecture ensures that every interaction is durably stored and can be reconstructed on demand by loading the graph's state.

Architecture Overview

The persistence mechanism relies on three core components working together: a state definition that marks messages for accumulation, a SQLite-based checkpoint saver, and asynchronous API wrappers that handle the synchronous storage layer. When a user sends a message, the system updates the conversational state, triggers the LLM through the call_model_with_messages node, and automatically commits the new message pair to the checkpoint file.

Defining Conversational State in ThreadState

In open_notebook/graphs/chat.py, the ThreadState class defines the schema for conversation data. The messages field is annotated with add_messages, instructing LangGraph to append new messages rather than replace the existing list.

# open_notebook/graphs/chat.py

class ThreadState:
    messages: Annotated[list, add_messages]
    # ... other fields

This field holds a list of LangChain message objects such as HumanMessage and AIMessage. The annotation ensures that when the agent node returns new messages, they are merged with the existing history rather than overwriting it.

Configuring the SQLite Checkpoint

The system uses a SqliteSaver to provide durable storage. During application startup, the code establishes a connection to the SQLite file defined in open_notebook/config.py via the LANGGRAPH_CHECKPOINT_FILE constant.

# open_notebook/config.py

LANGGRAPH_CHECKPOINT_FILE = f"{sqlite_folder}/checkpoints.sqlite"

# open_notebook/graphs/chat.py

conn = sqlite3.connect(LANGGRAPH_CHECKPOINT_FILE, check_same_thread=False)
memory = SqliteSaver(conn)
graph = agent_state.compile(checkpointer=memory)

Passing the SqliteSaver instance as the checkpointer parameter in StateGraph.compile() enables automatic persistence. After each node execution, LangGraph writes the updated state to the SQLite database without requiring explicit save calls in the business logic.

Message Persistence Flow

The chat graph contains a single node named "agent" that executes the call_model_with_messages function. This node receives the current ThreadState, appends the user's input as a HumanMessage, invokes the LLM, and returns the assistant's response as an AIMessage.

Because the graph was compiled with the SqliteSaver checkpointer, the updated messages list is automatically written to the checkpoint file immediately after the node completes execution. This atomic write ensures that even if the application crashes mid-conversation, the history up to the last completed turn remains intact in the SQLite file.

Retrieving Chat History from Checkpoints

Retrieving conversation history requires accessing the stored state via graph.get_state(). Since SqliteSaver does not support asynchronous operations, the API layer in api/routers/chat.py wraps these calls in asyncio.to_thread() to prevent blocking the FastAPI event loop.

# api/routers/chat.py

thread_state = await asyncio.to_thread(
    chat_graph.get_state,
    config=RunnableConfig(configurable={"thread_id": full_session_id}),
)
messages = thread_state.values.get("messages", [])

The RunnableConfig must include the thread_id used during the initial conversation to locate the correct checkpoint snapshot. The returned state contains the full messages list, which the router translates into the public ChatMessage model for API responses used by endpoints such as get_session and execute_chat.

Optimized Session Queries with Message Counts

For operations that only need metadata rather than the full conversation payload, open_notebook/utils/graph_utils.py provides the get_session_message_count helper. This function performs the same thread-based get_state call but returns only len(state.values["messages"]), reducing memory overhead when listing sessions.

# open_notebook/utils/graph_utils.py

def get_session_message_count(graph, session_id: str) -> int:
    state = graph.get_state(
        config=RunnableConfig(configurable={"thread_id": session_id})
    )
    return len(state.values["messages"])

The FastAPI endpoints use this utility to display message counts in session lists without deserializing the entire chat history.

Practical Code Examples

Starting a Chat Session with Automatic Persistence

The following pattern demonstrates how the API initiates a conversation while ensuring the history is preserved:

import asyncio
from open_notebook.graphs.chat import graph as chat_graph
from langchain_core.runnables import RunnableConfig
from langchain_core.messages import HumanMessage

async def start_chat(session_id: str, user_text: str, model: str | None = None):
    # Load existing state (or start with an empty dict)

    state = await asyncio.to_thread(
        chat_graph.get_state,
        config=RunnableConfig(configurable={"thread_id": session_id}),
    )
    values = state.values if state else {}
    values.setdefault("messages", []).append(HumanMessage(content=user_text))

    # Run the graph – the new AI reply is stored in the checkpoint

    result = chat_graph.invoke(
        input=values,
        config=RunnableConfig(
            configurable={"thread_id": session_id, "model_id": model}
        ),
    )
    return result["messages"]          # contains both HumanMessage and AIMessage

Fetching the Full History for a Session

To retrieve all messages for an existing session:

import asyncio
from open_notebook.graphs.chat import graph as chat_graph
from langchain_core.runnables import RunnableConfig

async def get_history(session_id: str):
    state = await asyncio.to_thread(
        chat_graph.get_state,
        config=RunnableConfig(configurable={"thread_id": session_id}),
    )
    return state.values.get("messages", [])

Accessing History via the FastAPI Endpoint

You can also retrieve persisted messages using the HTTP API:

curl -X GET "http://localhost:5055/chat/sessions/abc123" \
     -H "Accept: application/json"

The server builds the response from the checkpoint using the logic shown above.

Summary

• ThreadState in open_notebook/graphs/chat.py defines the conversation schema using add_messages to accumulate history across turns.
• SqliteSaver provides durable storage to the file specified by LANGGRAPH_CHECKPOINT_FILE in open_notebook/config.py.
• Automatic persistence occurs after each node execution when the graph is compiled with the checkpointer parameter.
• Asynchronous retrieval in api/routers/chat.py requires wrapping graph.get_state() in asyncio.to_thread() due to the synchronous nature of SQLite operations.
• Optimized counting via get_session_message_count in open_notebook/utils/graph_utils.py enables lightweight session metadata queries.

Frequently Asked Questions

Where is the chat history physically stored?

The chat history is stored in a SQLite database file defined by the LANGGRAPH_CHECKPOINT_FILE constant in open_notebook/config.py. By default, this file resides at {sqlite_folder}/checkpoints.sqlite and contains the serialized checkpoint data managed by LangGraph's SqliteSaver.

Does the chat history persist across application restarts?

Yes, because the SqliteSaver writes state to a physical SQLite file on disk, all conversation history survives application restarts. When the application restarts and reconnects to the same checkpoint file, calling graph.get_state() with the original thread_id returns the complete message history from the previous session.

Why does the API use asyncio.to_thread() when accessing checkpoints?

LangGraph's SqliteSaver implementation uses synchronous SQLite operations that block the calling thread. Since the FastAPI application runs on an async event loop, the API layer wraps get_state() and invoke() calls in asyncio.to_thread() to offload the blocking I/O to a separate thread pool, maintaining responsiveness for concurrent requests.

How can I check the number of messages in a session without loading the full conversation?

Use the get_session_message_count helper function located in open_notebook/utils/graph_utils.py. This function calls graph.get_state() and returns the length of the messages list, providing an efficient way to display message counts in session listings without deserializing the entire conversation payload.

{
  "mcpServers": {
    "instagit": {
      "command": "npx",
      "args": ["-y", "instagit@latest"]
    }
  }
}