LangGraph human-in-the-loop workflows

LangGraph’s human-in-the-loop support is the difference between a demo agent and a system you can ship. The framework gives you interrupt() for pausing execution, checkpointing for free state persistence, and a clean API for resuming with human input. This guide walks through the patterns that work in production, the ones that don’t, and the tradeoffs you’ll face at each step.

The core pattern: interrupt and resume

LangGraph’s interrupt() function pauses graph execution and serializes state to the checkpointer. When you’re ready to continue, you invoke the graph with a Command(resume=...) object. The mental model is synchronous: the graph stops, your application collects input, the graph resumes.

from langgraph.graph import StateGraph, START, END
from langgraph.checkpoint.memory import MemorySaver
from langgraph.types import interrupt, Command
from typing import TypedDict, Literal

class State(TypedDict):
    question: str
    answer: str
    approved: bool

def draft_answer(state: State):
    # In practice, call your LLM here
    return {"answer": f"Draft response to: {state['question']}"}

def human_review(state: State):
    # This pauses execution and returns control to the caller
    decision = interrupt({
        "question": state["question"],
        "draft": state["answer"],
        "prompt": "Approve, edit, or reject?"
    })
    return {"approved": decision["action"] == "approve", "answer": decision.get("edited", state["answer"])}

def route_after_review(state: State) -> Literal["approved", "rejected"]:
    return "approved" if state["approved"] else "rejected"

builder = StateGraph(State)
builder.add_node("draft", draft_answer)
builder.add_node("review", human_review)
builder.add_edge(START, "draft")
builder.add_edge("draft", "review")
builder.add_conditional_edges("review", route_after_review, {
    "approved": END,
    "rejected": "draft"
})

checkpointer = MemorySaver()
graph = builder.compile(checkpointer=checkpointer)

The checkpointer is required. Without it, interrupt() raises an error. MemorySaver works for development; production needs PostgresSaver or SqliteSaver for durability across restarts.

Running the workflow

config = {"configurable": {"thread_id": "session-123"}}

# First invocation — runs until interrupt
result = graph.invoke({"question": "What's the refund policy?"}, config=config)
print(result["__interrupt__"])
# Output: Interrupt(value={'question': '...', 'draft': '...', 'prompt': '...'}, resumable=True)

# Human provides input (in your UI, CLI, Slack bot, etc.)
human_input = {"action": "approve"}  # or {"action": "edit", "edited": "Custom response"}

# Resume with Command
result = graph.invoke(Command(resume=human_input), config=config)
print(result["answer"])

The thread_id scopes the checkpoint. Each conversation, ticket, or task gets its own thread. This is how you support concurrent human-in-the-loop sessions.

Common approval patterns

Binary approve/reject

Simplest pattern. The human sees a draft and clicks approve or reject. On reject, loop back to the generator node.

def human_approval(state: State):
    decision = interrupt({"draft": state["draft"], "type": "approval"})
    return {"approved": decision == "approve"}

# Route: approved -> END, rejected -> generator

Tradeoff: low friction, but no way to capture why something was rejected. Add a required comment field if you need audit trail.

Edit-in-place

Let the human modify the draft directly. The edited version becomes the final output.

def human_edit(state: State):
    decision = interrupt({"draft": state["draft"], "type": "edit"})
    if decision["action"] == "save":
        return {"final": decision["content"]}
    return {"final": state["draft"]}  # fallback

Tradeoff: higher cognitive load on the human. Works well for short text (email subject lines, commit messages). Breaks down for long-form content where diffs are clearer.

Structured feedback with re-generation

The human provides structured critique; the generator re-runs with that context.

def human_feedback(state: State):
    decision = interrupt({"draft": state["draft"], "type": "feedback"})
    return {"feedback": decision.get("notes", ""), "retry_count": state.get("retry_count", 0) + 1}

def regenerate(state: State):
    # Pass feedback to your LLM prompt
    prompt = f"Previous draft: {state['draft']}\nFeedback: {state['feedback']}\nRewrite:"
    new_draft = llm.invoke(prompt)
    return {"draft": new_draft, "retry_count": state["retry_count"]}

# Route: feedback -> regenerate -> review (with max_retries guard)

Tradeoff: adds latency (another LLM call) but produces better results. Cap retries at 2-3 to avoid infinite loops.

State design for human-in-the-loop

Keep interrupt payloads small and serializable. The checkpointer stores the entire state snapshot at each interrupt. Large objects (base64 images, full conversation histories) bloat your database and slow resume.

# Good: minimal interrupt payload
interrupt({"ticket_id": "TKT-442", "summary": "Refund request", "draft": "..."})

# Bad: entire conversation history in interrupt
interrupt({"messages": state["messages"], "draft": "..."})  # don't do this

Store large artifacts in object storage (S3, GCS) and pass references. The checkpointer only needs the pointer.

class State(TypedDict):
    ticket_id: str
    artifact_ref: str  # s3://bucket/key
    draft: str
    human_decision: dict | None

Handling timeouts and abandonment

Humans walk away. Your system needs to handle stalled interrupts.

Option 1: TTL on checkpoints

Set a TTL on the checkpoint row (Postgres: expires_at column). A background job marks stale threads as expired and routes them to an escalation queue.

# In your checkpointer setup (PostgresSaver example)
from langgraph.checkpoint.postgres import PostgresSaver

saver = PostgresSaver(conn_string)
saver.setup()  # creates tables with expires_at column

# When compiling, you can't set TTL directly in LangGraph yet.
# Instead, run a cron: UPDATE checkpoints SET metadata = jsonb_set(metadata, '{expired}', 'true') 
# WHERE updated_at < NOW() - INTERVAL '24 hours' AND metadata->>'interrupt' IS NOT NULL;

Option 2: Explicit timeout node

Add a timer node that runs in parallel (requires langgraph>=0.2 with async support) or handle it in your application layer: when the user returns, check state["created_at"] and auto-escalate if stale.

import time
from datetime import datetime, timedelta

def check_timeout(state: State):
    created = datetime.fromisoformat(state["interrupt_created_at"])
    if datetime.utcnow() - created > timedelta(hours=4):
        return {"escalated": True}
    return {"escalated": False}

# Add as first node after interrupt, route to escalation if true

Tradeoff: application-layer timeout is simpler but requires the human to return to trigger it. Database TTL catches abandonment even if the user never comes back.

Multi-human workflows

Some processes need multiple reviewers in sequence or parallel.

Sequential approval

def legal_review(state: State):
    decision = interrupt({"stage": "legal", "content": state["draft"]})
    return {"legal_approved": decision == "approve"}

def security_review(state: State):
    decision = interrupt({"stage": "security", "content": state["draft"]})
    return {"security_approved": decision == "approve"}

# Route: draft -> legal -> security -> END
# Each interrupt pauses for a different role

Parallel approval (fan-out)

LangGraph doesn’t have native parallel interrupts — the graph is single-threaded per invocation. Simulate it by collecting all approvals in one interrupt payload, or use subgraphs.

def parallel_review(state: State):
    # Single interrupt with multiple fields
    decision = interrupt({
        "legal": {"content": state["draft"], "required": True},
        "security": {"content": state["draft"], "required": True},
        "finance": {"content": state["draft"], "required": False},
    })
    return {
        "legal_approved": decision["legal"] == "approve",
        "security_approved": decision["security"] == "approve",
        "finance_approved": decision.get("finance") == "approve",
    }

Tradeoff: single interrupt means all reviewers must act before the graph resumes. If legal approves but security is on vacation, the draft sits. For true async parallel approval, use separate threads per reviewer and a coordinator node that polls their status.

Integrating with external systems

Most human-in-the-loop workflows live outside the graph: Slack, email, Jira, custom dashboards. The pattern is always the same:

1. Graph invokes, hits interrupt(), returns Interrupt object
2. Your application serializes the interrupt payload, sends to external system
3. External system posts back to your /resume endpoint
4. Your endpoint calls graph.invoke(Command(resume=payload), config)

# FastAPI example
from fastapi import FastAPI, HTTPException
from pydantic import BaseModel

app = FastAPI()

class ResumeRequest(BaseModel):
    thread_id: str
    payload: dict

@app.post("/resume")
async def resume_workflow(req: ResumeRequest):
    config = {"configurable": {"thread_id": req.thread_id}}
    try:
        result = graph.invoke(Command(resume=req.payload), config=config)
        return {"status": "completed", "result": result}
    except Exception as e:
        # Handle graph errors, invalid state, etc.
        raise HTTPException(500, str(e))

The external system only needs the thread_id and the interrupt payload schema. Keep that schema stable — version it if you change fields.

Observability and debugging

You need to answer: “Where is this ticket stuck?” and “Why did the human reject this?”

Log interrupt events

import structlog

logger = structlog.get_logger()

def human_review(state: State):
    logger.info("human_review_interrupt", thread_id=config["configurable"]["thread_id"], draft_length=len(state["draft"]))
    decision = interrupt({...})
    logger.info("human_review_resumed", thread_id=..., action=decision["action"])
    return {...}

Replay from checkpoint

LangGraph’s checkpointer lets you rewind. Useful for debugging: load a past checkpoint, resume with different input, see what changes.

# Get checkpoint history
checkpoints = list(saver.list(config))
for cp in checkpoints:
    print(cp.config["configurable"]["checkpoint_id"], cp.metadata)

# Resume from a specific checkpoint
past_config = {"configurable": {"thread_id": "session-123", "checkpoint_id": "abc-123"}}
result = graph.invoke(Command(resume=new_input), config=past_config)

This is also how you implement “undo” in your UI.

Common pitfalls

Forgetting the checkpointer

graph.compile() without a checkpointer works for simple graphs. Add interrupt() and it fails at runtime with a cryptic error about missing checkpoint. Always pass a checkpointer if you use interrupts.

Mutating state in interrupt payload

The interrupt payload is derived from state, but it’s a copy. Mutating the payload doesn’t affect state. Return a dict from the interrupt node to update state.

# Wrong
def bad_review(state: State):
    decision = interrupt({"draft": state["draft"]})
    state["draft"] = decision["edited"]  # does nothing
    return {}

# Right
def good_review(state: State):
    decision = interrupt({"draft": state["draft"]})
    return {"draft": decision["edited"]}

Blocking the event loop

interrupt() is synchronous. In an async FastAPI handler, graph.invoke() blocks. Use graph.ainvoke() with an async checkpointer (AsyncPostgresSaver) or run the invoke in a thread pool.

from concurrent.futures import ThreadPoolExecutor

executor = ThreadPoolExecutor(max_workers=4)

@app.post("/resume")
async def resume(req: ResumeRequest):
    loop = asyncio.get_event_loop()
    config = {"configurable": {"thread_id": req.thread_id}}
    result = await loop.run_in_executor(executor, lambda: graph.invoke(Command(resume=req.payload), config=config))
    return result

Race conditions on resume

Two humans click “approve” simultaneously on the same thread. The second resume gets a stale checkpoint error. Prevent this with:

1. Optimistic locking: include checkpoint_id in resume request, reject if mismatched
2. Application-level locking: Redis lock per thread_id during resume
3. Idempotency keys: client generates UUID, server deduplicates

import redis

redis_client = redis.Redis()

@app.post("/resume")
async def resume(req: ResumeRequest):
    lock_key = f"lock:thread:{req.thread_id}"
    acquired = redis_client.set(lock_key, "1", nx=True, ex=30)
    if not acquired:
        raise HTTPException(409, "Concurrent resume in progress")
    try:
        config = {"configurable": {"thread_id": req.thread_id}}
        result = graph.invoke(Command(resume=req.payload), config=config)
        return result
    finally:
        redis_client.delete(lock_key)

When to use a gateway for model calls

Your human-in-the-loop nodes often call LLMs for drafting, summarizing, or re-generating. If you’re routing across multiple providers (OpenAI, Anthropic, open models), a gateway simplifies the node code. You get one endpoint, automatic fallback when a provider degrades, and per-token metering without wiring it per node. n4n.ai handles this with an OpenAI-compatible endpoint addressing 240+ models and forwards provider cache-control hints so your draft nodes benefit from prompt caching automatically.

Testing human-in-the-loop flows

Unit test the nodes in isolation. Integration test the full graph with a MemorySaver.

import pytest

def test_approval_flow():
    checkpointer = MemorySaver()
    graph = builder.compile(checkpointer=checkpointer)
    config = {"configurable": {"thread_id": "test-1"}}
    
    # Run to interrupt
    result = graph.invoke({"question": "Test?"}, config=config)
    assert "__interrupt__" in result
    
    # Resume with approval
    result = graph.invoke(Command(resume={"action": "approve"}), config=config)
    assert result["approved"] is True
    assert "__interrupt__" not in result  # reached END

def test_rejection_loops_back():
    checkpointer = MemorySaver()
    graph = builder.compile(checkpointer=checkpointer)
    config = {"configurable": {"thread_id": "test-2"}}
    
    graph.invoke({"question": "Test?"}, config=config)
    result = graph.invoke(Command(resume={"action": "reject"}), config=config)
    
    # Should be back at draft node (or review node depending on graph structure)
    assert "draft" in result  # or checkpointer.get_state(config).values

Test timeout paths, escalation paths, and concurrent resume attempts. Mock the checkpointer for unit tests; use real SqliteSaver for integration tests.

Scaling considerations

Summary

LangGraph human-in-the-loop workflows center on interrupt() and Command(resume=...). The checkpointer is mandatory. Design small, serializable interrupt payloads. Handle abandonment with TTLs or application timeouts. Build your external integration (Slack, email, dashboard) around the thread_id + resume endpoint pattern. Log everything for replay and debugging. Test the full cycle: invoke → interrupt → resume → complete.

The patterns here scale from a single approval step to multi-stage reviews with parallel reviewers. Start simple — binary approve/reject with a MemorySaver — then add structure as your process demands it.