Debug LangGraph Workflow Execution and View State Transitions in Open‑Notebook

You can debug LangGraph workflows in Open‑Notebook by inspecting checkpointed states via graph.get_state(), stepping through nodes manually, and querying the SQLite persistence layer to view every state transition.

Open‑Notebook is an open‑source knowledge management system that leverages LangGraph to power conversational AI workflows. When building complex multi‑step pipelines like the ask and chat graphs, knowing how to debug LangGraph workflow execution and view state transitions is critical for verifying conditional routing and ensuring data integrity between asynchronous nodes.

How LangGraph Structures Workflows in Open‑Notebook

StateGraph Compilation and Typed State

In open_notebook/graphs/ask.py, each workflow begins as a StateGraph that declares typed state structures using TypedDict models. These structures define the exact schema of data flowing between nodes. The graph is compiled into a callable object via agent_state.compile(), which enables asynchronous execution with automatic checkpointing.

The Ask Workflow Architecture

The ask workflow implements a complex routing pattern defined in open_notebook/graphs/ask.py:

• The agent node (implemented by call_model_with_messages) parses user input into a Strategy object containing search terms and instructions.
• Conditional edges route to trigger_queries, which creates Send objects for each search term, dynamically spawning parallel paths.
• The provide_answer node executes vector searches via vector_search and synthesizes responses using an LLM.
• write_final_answer formats the final output before reaching END.

This state‑machine architecture means you can pause execution after any node to inspect the exact values being passed downstream.

Inspecting State Transitions and Checkpointed Data

Retrieving the Full State Snapshot

LangGraph persists state after each node execution in a SQLite checkpoint. To inspect the complete state history for a given thread, use the get_state method on the compiled graph:

import asyncio
from langchain_core.runnables import RunnableConfig
from open_notebook.graphs.ask import graph

async def dump_state(thread_id: str):
    config = RunnableConfig(configurable={"thread_id": thread_id})
    # Returns a StateSnapshot containing values and next node

    state = await asyncio.to_thread(graph.get_state, config=config)
    print("Current values:", state.values)
    print("Next node to execute:", state.next)

The graph.get_state method returns a StateSnapshot with values (the current TypedDict) and next (the pending node name), allowing you to debug LangGraph workflow execution by viewing exact data at any transition point.

Querying Message Counts and Session Data

For quick diagnostics without manually parsing the checkpoint, use the utility in open_notebook/utils/graph_utils.py:

from open_notebook.utils.graph_utils import get_session_message_count

async def check_session(thread_id: str):
    count = await get_session_message_count(graph, thread_id)
    print(f"Messages stored for {thread_id}: {count}")

This helper demonstrates the standard pattern for querying SQLite checkpoints using the thread‑id scoping mechanism.

Step‑by‑Step Debugging Techniques

Manual Node Execution

To debug individual nodes without running the full graph, compile the graph without external checkpointers and invoke stepwise:

import asyncio
from langchain_core.runnables import RunnableConfig
from open_notebook.graphs.ask import agent_state

# In-memory compilation for isolated debugging

debug_graph = agent_state.compile()

async def step_through():
    cfg = RunnableConfig(configurable={"thread_id": "debug-1"})
    # Execute only the agent node

    state = await debug_graph.ainvoke(
        {"question": "How do conditional edges work?"},
        config=cfg
    )
    print("Agent output:", state)
    
    # Manually process subsequent sends if present

    for send in state.get("__next_edges__", []):
        sub_state = await debug_graph.ainvoke(send["payload"], config=cfg)
        print(f"After {send['node']}: {sub_state}")

asyncio.run(step_through())

This technique lets you pause execution after any node to verify intermediate state dictionaries before LangGraph routes to the next step.

Common Debugging Patterns

When tracing issues in open_notebook/graphs/ask.py:

• Missing results in provide_answer: Insert print statements before the LLM call (around lines 100‑110) to inspect raw vector search outputs.
• Unexpected model selection: Verify configurable["answer_model"] in your RunnableConfig; override with {"configurable": {"answer_model": "gpt-4"}} to force a specific model.
• Conditional edge failures: Log the list of Send objects inside trigger_queries to confirm the routing logic produced the expected destinations.

Configuration and Persistence Details

SQLite Checkpoint Storage

Open‑Notebook configures persistence in open_notebook/config.py via the LANGGRAPH_CHECKPOINT_FILE variable, which defaults to a SQLite database in the repository root. This SqliteSaver checkpointer enables:

• Cross‑API call persistence using thread_id scoping via RunnableConfig.
• Direct SQL inspection using any SQLite browser to view raw state rows.
• State pruning via graph.checkpointer.wipe(thread_id) for long‑running sessions.

The checkpoint file stores serialized StateSnapshot rows, making it possible to audit every state transition offline.

Summary

• Open‑Notebook defines LangGraph workflows in open_notebook/graphs/ask.py and open_notebook/graphs/chat.py using StateGraph with typed states.
• Use graph.get_state(config) to retrieve full checkpoint snapshots and debug LangGraph workflow execution.
• Step through nodes manually by compiling with agent_state.compile() and inspecting __next_edges__.
• Query session metrics via get_session_message_count() in open_notebook/utils/graph_utils.py.
• State persists to SQLite via LANGGRAPH_CHECKPOINT_FILE, enabling offline inspection of every transition.

Frequently Asked Questions

How do I view the intermediate state between nodes in a LangGraph workflow?

Call graph.get_state() with a RunnableConfig containing the target thread_id. This returns a StateSnapshot object with values (the current data) and next (the upcoming node), allowing you to inspect exactly what data is being passed between steps in open_notebook/graphs/ask.py.

Why is my conditional edge not triggering the expected next node?

Conditional edges in open_notebook/graphs/ask.py return Send objects that determine routing. Add logging inside the trigger_queries function to verify the list of Send objects being returned, and ensure your state contains the required keys (like search_terms) that the edge logic depends on.

Can I debug LangGraph workflows without using the SQLite checkpoint?

Yes. Compile the graph using agent_state.compile() without passing a checkpointer to create an in‑memory instance. This allows you to invoke ainvoke() repeatedly and inspect state dictionaries without persisting data to the LANGGRAPH_CHECKPOINT_FILE defined in open_notebook/config.py.

Where is conversation history stored when debugging the chat workflow?

The chat workflow in open_notebook/graphs/chat.py uses the same checkpoint mechanism as ask. Conversation messages are stored in the SQLite file specified by LANGGRAPH_CHECKPOINT_FILE, scoped to the thread_id in your RunnableConfig. Use get_session_message_count() to query the number of stored messages for a specific session.

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