LangChainLangGraphhuman-in-the-loopagentsPython

Collecting Human Feedback in LangGraph

Pramod Dutta · Jul 1, 2026 · 13 min read

LangGraph human feedback means pausing a running graph at a specific node, handing control back to a person, and resuming execution with whatever that person decided, an approval, an edit, or a rejection. This matters because most production agents cannot be trusted to take irreversible actions (sending an email, running a database migration, spending money) without a checkpoint where a human signs off. LangGraph gives you this out of the box through the interrupt() function paired with a checkpointer, so you don't have to build your own polling loop or stash state in a side database.

This guide walks through the mechanics: how interrupt() actually pauses a graph, how to resume it with Command(resume=...), and how to build the three patterns you'll need most, approval gates, editable state, and multi-turn feedback loops. Every example is runnable Python against a current LangGraph release.

Why LangGraph human feedback needs a checkpointer

A LangGraph graph is a state machine. When you call interrupt() inside a node, LangGraph doesn't just block a thread waiting for input, it actually raises a special exception that unwinds the graph run and returns control to your calling code. For the graph to know where it stopped and what state it had when it stopped, that state has to be persisted somewhere. That's the job of a checkpointer.

Without a checkpointer, interrupt() will fail immediately, because there's nowhere to save the pause point. The simplest checkpointer for local development is MemorySaver, which keeps checkpoints in process memory:

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

class State(TypedDict):
    topic: str
    draft: str
    approved: bool

def write_draft(state: State) -> State:
    return {"draft": f"A short article about {state['topic']}."}

builder = StateGraph(State)
builder.add_node("write_draft", write_draft)
builder.add_edge(START, "write_draft")
builder.add_edge("write_draft", END)

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

For anything beyond a demo, swap MemorySaver for a persistent backend, PostgresSaver or SqliteSaver, so an interrupted run survives a process restart. A human might not respond for hours, and your server will almost certainly redeploy in the meantime.

from langgraph.checkpoint.postgres import PostgresSaver

DB_URI = "postgresql://user:password@localhost:5432/langgraph"

with PostgresSaver.from_conn_string(DB_URI) as checkpointer:
    checkpointer.setup()
    graph = builder.compile(checkpointer=checkpointer)

Every run against a compiled graph needs a thread_id in its config. That thread ID is the handle you use later to find the paused run and resume it:

config = {"configurable": {"thread_id": "review-42"}}

Pausing a node with interrupt()

The core building block for LangGraph human feedback is the interrupt() function from langgraph.types. Call it from inside any node, pass it a payload describing what you need reviewed, and the graph run stops there. The value you eventually resume with becomes the return value of that same interrupt() call, as if it were a blocking input().

from langgraph.types import interrupt, Command

def human_review(state: State) -> State:
    decision = interrupt({
        "question": "Approve this draft?",
        "draft": state["draft"],
    })
    return {"approved": decision == "approve"}

builder = StateGraph(State)
builder.add_node("write_draft", write_draft)
builder.add_node("human_review", human_review)
builder.add_edge(START, "write_draft")
builder.add_edge("write_draft", "human_review")
builder.add_edge("human_review", END)

graph = builder.compile(checkpointer=checkpointer)

When you invoke the graph, it runs until it hits interrupt() and then stops, returning an __interrupt__ key in the output instead of raising an uncaught error:

result = graph.invoke({"topic": "vector databases", "draft": "", "approved": False}, config)
print(result["__interrupt__"])
# [Interrupt(value={'question': 'Approve this draft?', 'draft': '...'}, ...)]

At this point the graph is genuinely paused. No thread is blocked, no cron job is polling. The checkpointer has the full state saved against thread_id="review-42", and you can shut down the process entirely. Later, from any process that shares the same checkpointer backend, you resume with Command(resume=...):

final = graph.invoke(Command(resume="approve"), config)
print(final["approved"])  # True

That single call re-enters the graph at the exact node that called interrupt(), substitutes your resume value as the return of that call, and continues execution forward. This is the entire mechanism behind LangGraph human feedback: pause, persist, resume.

Building an approval gate

The most common pattern for LangGraph human feedback is a binary approval gate in front of a risky action, sending an email, executing a trade, deleting a record. Model it as a conditional edge that branches on the human's decision rather than baking the branch into the node itself, so your graph structure documents the policy.

from langgraph.graph import StateGraph, START, END
from langgraph.types import interrupt, Command
from typing_extensions import TypedDict

class EmailState(TypedDict):
    recipient: str
    body: str
    decision: str

def draft_email(state: EmailState) -> EmailState:
    return {"body": f"Hi, following up on your request. Details attached."}

def request_approval(state: EmailState) -> EmailState:
    decision = interrupt({
        "action": "send_email",
        "recipient": state["recipient"],
        "body": state["body"],
    })
    return {"decision": decision}

def send_email(state: EmailState) -> EmailState:
    print(f"Sending to {state['recipient']}: {state['body']}")
    return {}

def cancel(state: EmailState) -> EmailState:
    print("Send cancelled by reviewer")
    return {}

def route(state: EmailState) -> str:
    return "send_email" if state["decision"] == "approve" else "cancel"

builder = StateGraph(EmailState)
builder.add_node("draft_email", draft_email)
builder.add_node("request_approval", request_approval)
builder.add_node("send_email", send_email)
builder.add_node("cancel", cancel)
builder.add_edge(START, "draft_email")
builder.add_edge("draft_email", "request_approval")
builder.add_conditional_edges("request_approval", route, {
    "send_email": "send_email",
    "cancel": "cancel",
})
builder.add_edge("send_email", END)
builder.add_edge("cancel", END)

graph = builder.compile(checkpointer=checkpointer)

Run it, hit the interrupt, then resume with either "approve" or "reject" depending on what your UI collected. The routing function reads the state the node wrote after resuming, keeping the decision logic out of the interrupt call itself.

Letting the human edit state, not just approve it

Approval gates are binary, but a lot of real review work is editorial: a person wants to fix a typo in the draft, adjust a dollar amount, or swap a tool argument before the agent proceeds. Because whatever value you pass to Command(resume=...) becomes the return value of interrupt(), you can pass back an entire edited object instead of a string, and write it into state.

def human_review(state: State) -> State:
    payload = interrupt({
        "question": "Edit this draft if needed, otherwise resend as-is.",
        "draft": state["draft"],
    })
    return {"draft": payload["draft"], "approved": payload["approved"]}

On the client side, the reviewer sees the draft, edits it in a text box, and your app resumes with the edited version:

edited = {"draft": "A short, punchier article about vector databases.", "approved": True}
graph.invoke(Command(resume=edited), config)

This pattern is what you want for reviewing tool calls before an agent executes them. Instead of approving a search_flights(origin="SFO") call blindly, surface the parsed arguments, let the human correct origin to "OAK", and resume with the corrected arguments so the actual tool call reflects the fix rather than the model's first guess.

def review_tool_call(state: AgentState) -> AgentState:
    proposed = state["tool_call"]
    approved_call = interrupt({
        "tool": proposed["name"],
        "args": proposed["args"],
    })
    return {"tool_call": approved_call}

Multi-turn feedback loops

Some workflows need more than one round trip, draft, feedback, revise, review again, until the human is satisfied. Model this as a loop in the graph: a conditional edge sends control back to a revision node whenever the human's decision isn't final.

class LoopState(TypedDict):
    draft: str
    feedback: str
    status: str

def write(state: LoopState) -> LoopState:
    note = f" Incorporating feedback: {state['feedback']}" if state.get("feedback") else ""
    return {"draft": f"Draft.{note}"}

def review(state: LoopState) -> LoopState:
    result = interrupt({"draft": state["draft"]})
    return {"status": result["status"], "feedback": result.get("feedback", "")}

def route(state: LoopState) -> str:
    return "write" if state["status"] == "revise" else END

builder = StateGraph(LoopState)
builder.add_node("write", write)
builder.add_node("review", review)
builder.add_edge(START, "write")
builder.add_edge("write", "review")
builder.add_conditional_edges("review", route, {"write": "write", END: END})

graph = builder.compile(checkpointer=checkpointer)

Each time the reviewer sends back {"status": "revise", "feedback": "make it shorter"}, the graph loops back through write, regenerates the draft with that feedback baked into the prompt, and pauses again at review. When the reviewer finally sends {"status": "approve"}, the loop exits. Because every pause is checkpointed, this can span days without holding any server resources open, and different people can review different rounds if ownership changes hands.

Streaming interrupts to a UI

If you're building a web app around this, don't call graph.invoke() and block on the whole run, stream it so your frontend can render intermediate steps and react to the interrupt event as soon as it fires.

for event in graph.stream(
    {"topic": "vector databases", "draft": "", "approved": False},
    config,
    stream_mode="values",
):
    if "__interrupt__" in event:
        interrupt_obj = event["__interrupt__"][0]
        # send interrupt_obj.value to your UI, store thread_id for the resume call
        break

Your API layer typically persists thread_id alongside a pending-review record in your own application database (separate from the LangGraph checkpointer), so a reviewer opening a "pending approvals" dashboard hours later can look up which thread to resume. When they submit their decision, your backend calls graph.invoke(Command(resume=decision), {"configurable": {"thread_id": thread_id}}) and the graph continues from exactly where it paused.

Inspecting and testing paused graphs

Before wiring up a UI, it helps to inspect the graph's state directly with get_state(), which shows you what's pending without resuming anything:

snapshot = graph.get_state(config)
print(snapshot.next)          # ('human_review',) - the node waiting to resume
print(snapshot.tasks)          # includes the interrupt payload

This is also the cleanest way to write tests for LangGraph human feedback flows. Invoke the graph, assert that snapshot.next contains the review node and that the interrupt payload matches what you expect, then resume with a fixed value and assert on the final state:

def test_approval_gate_rejects():
    config = {"configurable": {"thread_id": "test-1"}}
    graph.invoke({"topic": "x", "draft": "", "approved": False}, config)
    snapshot = graph.get_state(config)
    assert snapshot.next == ("human_review",)

    result = graph.invoke(Command(resume="reject"), config)
    assert result["approved"] is False

Because everything runs against MemorySaver in tests, there's no need to mock a UI or a queue, you're testing the actual pause-and-resume contract your production code relies on.

Routing feedback to the right reviewer

Not every pause should go to the same person. A support-ticket triage agent might need a manager to approve refunds over a threshold but let any agent approve a canned reply. Model this by writing a reviewer_role (or reviewer_id) into state before the interrupt fires, and have the interrupt payload carry that assignment so your UI layer can route the pending review to the right queue.

def request_approval(state: EmailState) -> EmailState:
    reviewer = "manager" if state.get("amount", 0) > 500 else "any"
    decision = interrupt({
        "action": "issue_refund",
        "amount": state.get("amount"),
        "assigned_to": reviewer,
    })
    return {"decision": decision}

Your application layer reads assigned_to off the interrupt value returned in event["__interrupt__"] and inserts a row into whatever queue table your reviewers pull from, keyed by thread_id. LangGraph itself doesn't know or care who resumes a thread, it only cares that Command(resume=...) eventually arrives on the same thread_id. That separation is deliberate: LangGraph owns pause-and-resume mechanics, your application owns identity, permissions, and notification.

If you need a hard guarantee that only the assigned reviewer can resume a given thread, enforce that check in your API handler before calling graph.invoke(Command(resume=...), config), not inside the graph. By the time the resume call reaches the graph, the decision has already been made, the graph's job is to act on it, not to authenticate who sent it.

Keeping an audit trail of decisions

Compliance-sensitive workflows, refunds, medical triage, financial trades, usually need a record of who approved what and when, separate from the LangGraph checkpoint itself. The checkpoint stores enough to resume the run, but it isn't a durable audit log by design, older checkpoints can be pruned once a thread completes.

The straightforward approach is to write an audit row from your API handler at the moment you call Command(resume=...), before invoking the graph:

def resume_review(thread_id: str, reviewer_email: str, decision: dict) -> dict:
    audit_log.insert({
        "thread_id": thread_id,
        "reviewer": reviewer_email,
        "decision": decision,
        "timestamp": datetime.utcnow(),
    })
    config = {"configurable": {"thread_id": thread_id}}
    return graph.invoke(Command(resume=decision), config)

This keeps the audit trail outside the graph's state, so it survives regardless of how long you retain checkpoints, and it captures the reviewer's identity, which never needs to flow through the graph itself. If you also want the decision embedded in the conversation history the agent sees on later turns (useful for a support agent that references "the manager approved this refund on Tuesday"), write a short summary of the decision into the graph state from inside the node after interrupt() returns, in addition to the external audit row.

Handling timeouts and abandoned reviews

Human reviewers don't always respond. A refund approval sitting in a queue for three days might need to auto-escalate or auto-expire rather than block the workflow indefinitely. LangGraph doesn't have a built-in timeout for interrupt(), since the graph is genuinely idle while paused, not polling a clock. You implement timeouts at the application layer instead: store the timestamp at which the interrupt fired (available from snapshot.tasks when you call graph.get_state(config)), and run a periodic job that checks for threads paused longer than your SLA.

def check_stale_reviews(graph, thread_ids, max_age_hours=24):
    stale = []
    for thread_id in thread_ids:
        config = {"configurable": {"thread_id": thread_id}}
        snapshot = graph.get_state(config)
        if not snapshot.next:
            continue  # already resumed
        age = datetime.utcnow() - snapshot.created_at
        if age.total_seconds() > max_age_hours * 3600:
            stale.append(thread_id)
    return stale

For threads that go stale, you can resume them programmatically with a default decision (Command(resume={"status": "escalate"})) rather than leaving a human waiting indefinitely, or fire a notification to a backup reviewer. Either way, the resume mechanism is identical to a human-submitted one, your escalation job is just another caller of Command(resume=...).

Common mistakes to avoid

A few things trip people up the first time they wire this up.

Forgetting the checkpointer. Calling interrupt() on a graph compiled without one raises an error immediately, since there's no state to pause and resume against.
Reusing a `thread_id` across unrelated runs. Each independent review needs its own thread, otherwise resuming one run can clobber another's checkpoint.
Putting side effects before the interrupt in the same node. If a node calls interrupt() after already sending an email or writing to a database, re-running that node on resume can repeat the side effect. Keep the interrupt as the first thing a node does, or split side effects into a separate downstream node that only runs after approval.
Treating the resume value as optional validation. Whatever you pass to Command(resume=...) becomes trusted state. Validate it in the node the same way you'd validate any external input, a reviewer's client could send a malformed payload.

FAQ

What's the difference between `interrupt()` and just raising an exception? interrupt() is a controlled pause that the checkpointer persists, so the graph can be resumed later from any process. A raised exception ends the run; there's no Command(resume=...) path back into the same execution point.

Does `interrupt()` block a thread while waiting for a human? No. The graph run actually returns control to your calling code once it hits the interrupt. No worker, thread, or connection stays open waiting, which is why this scales to reviews that take hours or days.

Can I have multiple interrupts in a single node? Yes, and each call is matched to its resume value by the order it executes in during replay. If you need more than two interrupts in one node, it's usually clearer to split them into separate nodes so each pause has its own name in snapshot.next.

Which checkpointer should I use in production? PostgresSaver is the standard production choice because it survives restarts and supports concurrent access from multiple app instances. MemorySaver is fine for local development and tests, but any interrupt raised against it disappears if the process exits.

How do I resume from a different process than the one that started the run? As long as both processes point at the same checkpointer backend and you have the correct thread_id, you can call graph.invoke(Command(resume=...), config) from anywhere, an API server, a scheduled job, or a script. The graph itself doesn't need to be the same Python object instance.

Can the human reject and end the run entirely instead of looping back? Yes, route the rejection branch to END in your conditional edges, as shown in the approval gate example. The graph simply stops there with whatever final state you wrote before ending.

Is `interrupt()` the same as LangGraph's older `interrupt_before` and `interrupt_after` compile-time options? No. Those older options pause execution before or after a fixed list of nodes, decided at compile time. interrupt() is called dynamically inside node code, so you can decide at runtime whether a particular execution needs review, and you can pass a rich payload describing exactly what the human needs to see.