Human-in-the-Loop AI Agents: Implementing Approval Breakpoints in LangGraph

Fully autonomous AI agents are powerful, but deploying them in production without safety boundaries is dangerous. An autonomous agent equipped with database write permissions, payment gateway APIs, or shell execution tools can easily hallucinate bad arguments, wipe customer data, or send unintended external emails.

To deploy agents safely in enterprise settings, software architects implement Human-in-the-Loop (HITL) controls. In a human-in-the-loop pattern, the agent executes routine planning and read-only lookups autonomously, but automatically pauses execution—creating a persistent state checkpoint—whenever it reaches a high-risk tool call. It waits for a human operator to review, approve, edit, or reject the planned action before resuming.

In this comprehensive production guide, you will learn how to implement robust Human-in-the-Loop approval breakpoints in Python using LangGraph and persistent SQLite checkpoints.

1. System Architecture: Persistent State Breakpoints

Unlike stateless LLM chains, LangGraph models agent workflows as cyclical state graphs backed by a persistent checkpointer. When a graph execution reaches an interrupted node, LangGraph persists the entire conversation memory and execution stack to storage, exiting cleanly without keeping an expensive Python process in memory:

2. Environment Setup & Dependencies

Create an isolated virtual environment and install the required dependencies:

python3 -m venv .venv
source .venv/bin/activate
pip install "langgraph>=0.2.0" "langchain-core>=0.3.0" "pydantic>=2.7.0"

We utilize LangGraph for cyclical graph state machines and its native MemorySaver / SqliteSaver checkpoint engines.

3. Defining the State Graph & Tools

Let us define an agent that can handle two types of actions:
1. Safe Read-Only Tool: lookup_customer_account (runs autonomously without human intervention).
2. High-Risk Mutating Tool: execute_customer_refund (requires explicit human approval before money is transferred).

import sqlite3
from typing import TypedDict, Annotated, List, Dict, Any, Union
from pydantic import BaseModel, Field

# --- State Definition ---
class AgentState(TypedDict):
    messages: List[Dict[str, Any]]
    action_type: str
    target_account: str
    refund_amount: float
    human_approved: bool
    final_output: str

# --- Mock Tool Implementations ---
def lookup_customer_account(account_id: str) -> Dict[str, Any]:
    """Safe read-only lookup: autonomous."""
    return {
        "account_id": account_id,
        "name": "Acme Corp Enterprise",
        "current_balance_usd": 12500.00,
        "status": "Active"
    }

def execute_customer_refund(account_id: str, amount_usd: float) -> Dict[str, Any]:
    """Sensitive write action: REQUIRES HUMAN APPROVAL."""
    return {
        "status": "SUCCESS",
        "transaction_id": "TXN-994821",
        "account_id": account_id,
        "refunded_usd": amount_usd
    }

4. Building the Graph Nodes

Now, we create the nodes: a planning node, a human approval checkpoint node, and an execution node:

def planner_node(state: AgentState) -> Dict[str, Any]:
    """Simulates LLM analyzing user request and deciding on tool invocation."""
    user_request = state["messages"][-1]["content"]
    print(f"\n[*] Planner evaluating: '{user_request}'")

    # In production, an LLM extracts these structured entities
    if "refund" in user_request.lower():
        return {
            "action_type": "REFUND",
            "target_account": "CUST-10492",
            "refund_amount": 450.00,
            "messages": state["messages"] + [{"role": "assistant", "content": "Proposed Action: Refund $450.00 to CUST-10492."}]
        }
    else:
        info = lookup_customer_account("CUST-10492")
        return {
            "action_type": "READ",
            "final_output": f"Account details retrieved: {info}",
            "messages": state["messages"] + [{"role": "assistant", "content": str(info)}]
        }

def route_action(state: AgentState) -> str:
    """Conditional router: Routes sensitive actions to approval checkpoint."""
    if state["action_type"] == "REFUND":
        return "human_approval_checkpoint"
    return "finalize"

def human_approval_checkpoint(state: AgentState) -> Dict[str, Any]:
    """
    Breakpoint Node: Execution will halt BEFORE this node executes.
    When resumed, this node processes the human verdict.
    """
    print("\n[!] Checkpoint reached: Processing human decision...")
    if not state.get("human_approved", False):
        return {
            "final_output": "Action cancelled: Human operator rejected the refund request.",
            "messages": state["messages"] + [{"role": "system", "content": "Refund rejected by operator."}]
        }

    # If approved, proceed to execute sensitive tool
    result = execute_customer_refund(
        account_id=state["target_account"],
        amount_usd=state["refund_amount"]
    )
    return {
        "final_output": f"Transaction successful! {result}",
        "messages": state["messages"] + [{"role": "system", "content": f"Executed refund: {result}"}]
    }

def finalize(state: AgentState) -> Dict[str, Any]:
    """Final response generation node."""
    print(f"[+] Completing workflow. Final summary:\n    {state.get('final_output')}")
    return state

5. Compiling the Graph with Breakpoint Interruption

We compile the graph using MemorySaver (or SqliteSaver) and declare interrupt_before=["human_approval_checkpoint"]:

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

# 1. Initialize StateGraph
builder = StateGraph(AgentState)

# 2. Add Nodes
builder.add_node("planner", planner_node)
builder.add_node("human_approval_checkpoint", human_approval_checkpoint)
builder.add_node("finalize", finalize)

# 3. Add Edges & Conditional Routing
builder.add_edge(START, "planner")
builder.add_conditional_edges("planner", route_action, {
    "human_approval_checkpoint": "human_approval_checkpoint",
    "finalize": "finalize"
})
builder.add_edge("human_approval_checkpoint", "finalize")
builder.add_edge("finalize", END)

# 4. Checkpointer & Interruption Configuration
memory = MemorySaver()
app = builder.compile(
    checkpointer=memory,
    interrupt_before=["human_approval_checkpoint"]
)

6. End-to-End Execution: Halting, Inspecting, and Resuming

Now let us execute the workflow. Notice how the agent automatically stops before money is refunded, allows state inspection, and resumes cleanly:

def run_demonstration():
    # Every conversation session has a unique thread_id
    config = {"configurable": {"thread_id": "session-customer-support-01"}}

    # 1. Initial User Prompt
    initial_input = {
        "messages": [{"role": "user", "content": "Please process a refund of $450 to customer CUST-10492 for defective hardware."}],
        "action_type": "",
        "target_account": "",
        "refund_amount": 0.0,
        "human_approved": False,
        "final_output": ""
    }

    print("=== STEP 1: INVOKING WORKFLOW ===")
    events = app.invoke(initial_input, config=config)

    # 2. Inspecting Paused State
    current_state = app.get_state(config)
    print("\n=== STEP 2: INSPECTING INTERRUPTED STATE ===")
    print(f"Graph Next Node to Execute: {current_state.next}")
    print(f"Target Account: {current_state.values.get('target_account')}")
    print(f"Requested Refund: ${current_state.values.get('refund_amount')}")

    # 3. Human Operator Decision (Can approve, reject, or modify values!)
    print("\n=== STEP 3: HUMAN OPERATOR REVIEWS PROPOSAL ===")
    print("Operator action: APPROVED (with adjusted discount of $400.00)")

    # Operator modifies state to adjust amount and mark approved
    app.update_state(
        config,
        {"human_approved": True, "refund_amount": 400.00},
        as_node="planner"
    )

    # 4. Resuming the Workflow
    print("\n=== STEP 4: RESUMING GRAPH EXECUTION ===")
    # Passing None resumes execution from the saved checkpoint!
    resumed_events = app.invoke(None, config=config)

    final_state = app.get_state(config)
    print(f"\n[+] Execution Finished: {final_state.values.get('final_output')}")

if __name__ == "__main__":
    run_demonstration()

Execution Output:

=== STEP 1: INVOKING WORKFLOW ===
[*] Planner evaluating: 'Please process a refund of $450 to customer CUST-10492 for defective hardware.'

=== STEP 2: INSPECTING INTERRUPTED STATE ===
Graph Next Node to Execute: ('human_approval_checkpoint',)
Target Account: CUST-10492
Requested Refund: $450.0

=== STEP 3: HUMAN OPERATOR REVIEWS PROPOSAL ===
Operator action: APPROVED (with adjusted discount of $400.00)

=== STEP 4: RESUMING GRAPH EXECUTION ===
[!] Checkpoint reached: Processing human decision...
[+] Completing workflow. Final summary:
    Transaction successful! {'status': 'SUCCESS', 'transaction_id': 'TXN-994821', 'account_id': 'CUST-10492', 'refunded_usd': 400.0}

[+] Execution Finished: Transaction successful! {'status': 'SUCCESS', 'transaction_id': 'TXN-994821', 'account_id': 'CUST-10492', 'refunded_usd': 400.0}

7. Production Best Practices for Enterprise HITL

1. Use Persistent Storage: In production, replace MemorySaver() with SqliteSaver or PostgresSaver. In-memory checkpointers will lose paused approvals if the server restarts.
2. State Editing (app.update_state): Take advantage of LangGraph’s state manipulation. Operators should not just have a binary “Yes/No” button; allow them to correct customer email addresses, update SQL query limits, or modify refund amounts directly before resuming.
3. Audit Trails & Security: Log operator identity (reviewed_by: "admin@company.com") and timestamps directly into the checkpoint state to maintain strict compliance records.
4. Defensive Fallbacks: For error handling on the tools themselves, implement tool calling retry loops to catch transient network failures automatically.

8. Frequently Asked Questions (FAQ)

What is the difference between interrupt_before and interrupt_after?

interrupt_before pauses execution before a specified node runs, which is ideal for approving dangerous write actions like refunds or deletes. interrupt_after pauses execution after a node finishes, which is ideal for having a human review or polish model-generated email drafts before they are dispatched.

Can LangGraph pause a workflow for days while waiting for an approval?

Yes. Because state is saved to a persistent database checkpoint (such as SQLite or PostgreSQL), the agent process consumes zero CPU or memory while waiting. When the human approves via a web dashboard days later, the workflow resumes instantly using its thread_id.

How does this integrate with asynchronous web frameworks like FastAPI?

In FastAPI, expose a webhook endpoint (POST /api/agents/{thread_id}/approve). When invoked, the endpoint calls app.update_state(config, {"human_approved": True}) followed by app.ainvoke(None, config=config) to stream the resumed response back to the client.