Human-in-the-Loop Workflows with LangGraph

This is the fourth post in our LangGraph fundamentals series, building on conditional edges and the agent-loop concepts from earlier in this series.

We flagged in our LangChain series that a tool taking a consequential action — a refund, an account change — needs safeguards beyond just trusting an agent's judgment. Pausing for actual human approval before a consequential step proceeds is one of the most direct safeguards available, and it needs real infrastructure: a graph has to genuinely stop, potentially for hours or days, and later resume in exactly the state it left off in. That's what LangGraph's checkpointing provides.

Interrupts: pausing a graph deliberately

from langgraph.graph import StateGraph, END
from langgraph.checkpoint.memory import MemorySaver
 
class RefundState(TypedDict):
    request_amount: float
    customer_id: str
    approved: bool
    result: str
 
def evaluate_refund(state: RefundState) -> RefundState:
    if state["request_amount"] > 500:
        state["approved"] = False  # requires human review — set explicitly, not assumed
    else:
        state["approved"] = True
    return state
 
def process_refund(state: RefundState) -> RefundState:
    state["result"] = f"Refund of ${state['request_amount']} processed for {state['customer_id']}"
    return state
 
graph = StateGraph(RefundState)
graph.add_node("evaluate_refund", evaluate_refund)
graph.add_node("process_refund", process_refund)
graph.set_entry_point("evaluate_refund")
graph.add_edge("evaluate_refund", "process_refund")
graph.add_edge("process_refund", END)
 
checkpointer = MemorySaver()
app = graph.compile(checkpointer=checkpointer, interrupt_before=["process_refund"])

interrupt_before=["process_refund"] tells the graph to stop before that specific node runs, no matter what any conditional logic decided — the graph reaches that point and genuinely pauses, returning control to your application rather than continuing automatically.

Resuming after a real decision

config = {"configurable": {"thread_id": "refund-request-4471"}}
 
result = app.invoke(
    {"request_amount": 750, "customer_id": "cust-123", "approved": False, "result": ""},
    config=config,
)
print("Paused. Current state:", app.get_state(config).values)

Paused. Current state: {'request_amount': 750, 'customer_id': 'cust-123', 'approved': False, 'result': ''}

Execution has genuinely stopped here — not just slowed down, actually halted, with its exact state persisted via the checkpointer and addressable later through thread_id. A human reviewer — hours or days later, in an entirely separate process, possibly on an entirely different machine — can approve and resume:

# Later, potentially in a completely different process, after a human reviews the request
app.update_state(config, {"approved": True})
final_result = app.invoke(None, config=config)
print(final_result["result"])

Refund of $750 processed for cust-123

app.invoke(None, config=config) — passing None as input — resumes a graph from exactly where it left off, using the thread_id to look up its persisted state, rather than starting over from the entry point. This is the mechanism that makes "pause for a genuinely long time, then resume exactly where you stopped" actually work, rather than requiring the whole workflow to somehow stay running and blocked the entire time a human takes to respond.

Rejecting, not just approving

A real approval workflow needs a genuine rejection path too, not just the happy path shown above:

def route_after_review(state: RefundState) -> str:
    if state["approved"]:
        return "process"
    return "reject"
 
def reject_refund(state: RefundState) -> RefundState:
    state["result"] = f"Refund request for {state['customer_id']} was declined."
    return state
 
graph.add_node("reject_refund", reject_refund)
graph.add_conditional_edges(
    "evaluate_refund",
    route_after_review,
    {"process": "process_refund", "reject": "reject_refund"},
)

Combining a conditional edge (from the previous post in this series) with the interrupt means a reviewer's decision — not just whether to proceed, but which of several distinct outcomes to route to — genuinely determines the graph's subsequent path, rather than the interrupt being a simple binary gate on one fixed next step.

Persisting checkpoints beyond a single process

MemorySaver stores checkpoints in memory — genuinely fine for local development and testing, and gone the instant the process restarts, which is a real problem for anything actually waiting on a human who might take hours. Production deployments use a persistent checkpointer instead:

from langgraph.checkpoint.postgres import PostgresSaver
 
checkpointer = PostgresSaver.from_conn_string("postgresql://localhost/langgraph_checkpoints")
app = graph.compile(checkpointer=checkpointer, interrupt_before=["process_refund"])

The same swap-the-backend pattern covered repeatedly throughout this blog — Terraform's remote state backend, LangChain's swappable vector stores and chat history backends — applied here to checkpoint persistence: a graph paused on a request now survives a server restart, a redeploy, or the application running across multiple instances, because its actual state lives in Postgres, not in one specific process's memory.

Notifying a human that a decision is actually waiting

A paused graph is only useful if the person meant to review it actually finds out there's something to review — LangGraph itself doesn't send that notification; that's application logic wired around the interrupt:

def evaluate_refund(state: RefundState) -> RefundState:
    if state["request_amount"] > 500:
        state["approved"] = False
        notify_reviewer(
            f"Refund of ${state['request_amount']} for {state['customer_id']} needs approval",
            thread_id=state.get("thread_id"),
        )
    else:
        state["approved"] = True
    return state

A real system typically pairs the interrupt itself with a notification (a Slack message, an email, an internal dashboard entry) linking directly to the specific thread_id a reviewer needs to act on — the interrupt mechanism handles the "genuinely pause and preserve state" half of the problem; surfacing that pending decision to an actual person is a separate, equally necessary piece of the overall system.

Designing what actually requires human approval

Not every action needs a human in the loop, and treating every single step as needing approval defeats most of the point of automating a workflow at all. The pattern worth adopting deliberately: identify the specific, genuinely consequential decisions — above a dollar threshold, an irreversible action, anything with real legal or safety weight — and interrupt specifically before those, while everything else in the same graph runs fully automated. The evaluate_refund node above already embodies this: small refunds proceed automatically; only the larger, more consequential ones reach the interrupt point at all.

Time-boxing an approval: what happens if nobody responds

A request genuinely stuck waiting on human approval indefinitely — because the assigned reviewer is on leave, or the notification was missed — is its own kind of failure mode, distinct from the graph itself malfunctioning. A production system typically pairs the interrupt with an external timeout check, not something LangGraph enforces on its own:

from datetime import datetime, timedelta
 
def check_stale_approvals():
    for thread_id in get_pending_approval_threads():
        state = app.get_state({"configurable": {"thread_id": thread_id}})
        paused_at = state.created_at
        if datetime.now() - paused_at > timedelta(hours=24):
            escalate_to_secondary_reviewer(thread_id)

A scheduled job (the same cron-triggered pattern covered for Docker image rescanning and SageMaker pipeline retraining elsewhere in this blog) checking for approvals that have been pending longer than an acceptable window, and escalating them to a secondary reviewer or an alert, closes the gap between "the graph correctly paused" and "someone will actually notice and act on it in time."

Auditing every approval decision

Because a paused-then-approved graph represents a genuinely consequential decision, it's worth recording who approved it and when, not just that it was approved:

def approve_refund(thread_id: str, approver_email: str):
    config = {"configurable": {"thread_id": thread_id}}
    app.update_state(config, {
        "approved": True,
        "approved_by": approver_email,
        "approved_at": datetime.now().isoformat(),
    })
    return app.invoke(None, config=config)

Capturing approved_by and approved_at directly in the graph's own state gives a genuine audit trail for exactly this kind of consequential, human-gated action — the same audit discipline covered for Terraform state, GitHub Actions deployment history, and SageMaker's Model Registry approval status throughout this blog, applied here to a human-in-the-loop decision specifically.

What to actually remember from this post

• interrupt_before genuinely pauses a graph's execution at a specific node, not just an optional slowdown — real infrastructure is required to make this actually work.
• A checkpointer persists a graph's exact state, addressable later by thread_id, which is what allows resuming a graph that paused for hours or days, from a different process entirely.
• invoke(None, config=...) resumes from a checkpoint rather than restarting a graph's execution from its entry point.
• A conditional edge after the interrupt supports genuine reject paths, not just a binary approve-and-continue gate.
• Use a persistent checkpointer (Postgres or similar) in production — MemorySaver doesn't survive a process restart, a real problem for anything genuinely waiting on human timescales.
• Notifying the actual human reviewer is separate application logic, not something the interrupt mechanism handles on its own.

Next in the series: Multi-Agent Systems with LangGraph, the final post — covering how multiple distinct agents, each with their own responsibilities, coordinate within a single graph.