Implementing Human-in-the-Loop Approval Workflows using LangGraph and SQLite State Management

PromptCube Expert 4/29/2026 486 views 11 likes 2 min read

State management is the biggest hurdle when moving from a simple "prompt-response" chatbot to a production agent that can actually handle long-running business processes. If you've tried building an approval flow—where an AI proposes a change but a human must click "Approve" before it hits the DB—you know that standard memory doesn't cut it. You need a persistent checkpoint that survives server restarts and allows the graph to "sleep" until an external signal wakes it up.

Implementing Human-in-the-Loop Approval Workflows using LangGraph and SQLite State Management

LangGraph handles this via Checkpointer. I've been using the SqliteSaver for local development because it's zero-config and lets me inspect the state in a DB browser if the agent goes off the rails.

The core trick is the interrupt_before parameter in the compile method. This tells the graph to halt execution right before a specific node, saving the current state to SQLite and yielding control back to the user.

Here is the skeletal setup for a "Budget Approval" agent:

from langgraph.graph import StateGraph, END
from langgraph.checkpoint.sqlite import SqliteSaver
from typing import TypedDict

class AgentState(TypedDict):
    proposal: str
    approved: bool
    status: str

def propose_spend(state: AgentState):
    # AI logic to generate proposal
    return {"proposal": "Buy 10x H100 GPUs", "status": "pending"}

def execute_purchase(state: AgentState):
    # Logic to actually hit an API
    return {"status": "completed"}

# Build graph
workflow = StateGraph(AgentState)
workflow.add_node("propose", propose_spend)
workflow.add_node("execute", execute_purchase)

workflow.set_entry_point("propose")
workflow.add_edge("propose", "execute")
workflow.add_edge("execute", END)

# This is the magic part: interrupt before the execution node
memory = SqliteSaver.from_conn_string(":memory:") 
app = workflow.compile(checkpointer=memory, interrupt_before=["execute"])

To actually run this, you need a thread_id. This is how SQLite differentiates between User A's approval flow and User B's.

config = {"configurable": {"thread_id": "user_123"}}

# First run: stops at 'execute'
app.invoke({"status": "start"}, config) 

# The graph is now suspended. The state is saved in SQLite.
# To resume after a human clicks 'Approve' in your UI:
app.update_state(config, {"approved": True}, as_node="execute")
app.invoke(None, config) # Passing None tells it to resume from the checkpoint

A few gotchas I hit during implementation:

The as_node trap: When using update_state, if you don't specify as_node, the graph might get confused about where the state update came from, potentially triggering the wrong edge. Always explicitly state which node is "providing" the update.

State Bloat: SQLite is great, but if you're passing massive JSON blobs through your state, your checkpoint table will explode in size. Keep your AgentState lean. Store the actual heavy documents in S3 or a proper DB and only pass the doc_id through LangGraph.

Concurrency: While SQLite handles this fine for a few users, if you scale to a multi-node cluster, you'll need to swap SqliteSaver for PostgresSaver. The API is almost identical, so the migration is painless.

The productivity gain here is massive because you stop writing "if/else" spaghetti code to track if a user has responded to a prompt. You treat the entire conversation as a state machine that can be paused and resumed across different HTTP requests.

All Replies (0)

No replies yet — be the first!

Write a Reply

Markdown supported