Chapter 18 . LangGraph . Lesson 7 of 11

Human approval gates

interrupt() pauses the graph and waits for a person. Command(resume=...) continues it with their answer.

Next: An LLM inside a node All lessons Script on GitHub

01Real-world example

Illustration: Before quarantining tests, the pipeline waits for a QA lead to approve, like a manual gate before a production deploy. — Before quarantining tests, the pipeline waits for a QA lead to approve, like a manual gate before a production deploy.

02The graph

Human approval gates as a graph — ask_approval pauses the graph. Command(resume=...) on the same thread_id carries your answer back in.

03How the code does it

interrupt(payload) stops the graph inside ask_approval and hands the payload to the caller under "__interrupt__".
The checkpointer saves the pause: app.get_state(cfg).next shows ('ask_approval',).
app.invoke(Command(resume=answer), cfg) runs ask_approval again; this time interrupt() returns the answer.
Resume on the same thread_id, or there is no saved pause to resume.

Key points

interrupt() needs a checkpointer
Resume with Command(resume=answer)
Same thread_id, or it starts over

04Run it

terminal

cd chapter_18_LangGraph/src/chapters
python 007_Human_In_The_Loop.py yes

output

Graph paused. It is asking:
  Quarantine these tests? (yes/no)
   - login_redirect
   - search_autocomplete
Paused before node: ('ask_approval',)

Resumed with 'yes' -> quarantined 2 tests

DDrills for lesson 7

Work in chapter_18_LangGraph/src/chapters. Change one thing, run it, and compare with the expected result.

Say no
Run python 007_Human_In_The_Loop.py no.

Expected result
Resumed with 'no' -> left as is, nothing changed
Resume on the wrong thread
Call invoke(Command(resume="yes"), ...) with a thread_id that never started.

Expected result
Nothing is resumed. The new thread starts from START, runs detect and pauses at ask_approval again.
Count the node runs
Count how many times ask_approval runs for one approval (a print at its top is enough).

Expected result
Twice. A resumed node runs again from the top, so put side effects after interrupt() or make them safe to repeat.

SThe full script

The exact file from the course repo, plus the output it printed when this page was written.

007_Human_In_The_Loop.py

"""007 - Human in the loop: pause the graph and ask before doing something risky.

interrupt(payload) freezes the graph and hands `payload` to the caller.
You resume it later with Command(resume=answer) on the SAME thread_id.
A checkpointer is required: the pause has to be saved somewhere.

    python 007_Human_In_The_Loop.py        # asks you
    python 007_Human_In_The_Loop.py yes    # answers for you (CI / demos)
"""
import sys
from typing import Literal, TypedDict

from langgraph.checkpoint.memory import InMemorySaver
from langgraph.graph import END, START, StateGraph
from langgraph.types import Command, interrupt


class State(TypedDict, total=False):
    flaky_tests: list[str]
    approved: bool
    outcome: str


def detect(state: State) -> dict:
    return {"flaky_tests": ["login_redirect", "search_autocomplete"]}


def ask_approval(state: State) -> dict:
    answer = interrupt({
        "question": "Quarantine these tests? (yes/no)",
        "tests": state["flaky_tests"],
    })
    return {"approved": str(answer).strip().lower() in ("y", "yes")}


def route(state: State) -> Literal["quarantine", "skip"]:
    return "quarantine" if state["approved"] else "skip"


def quarantine(state: State) -> dict:
    return {"outcome": f"quarantined {len(state['flaky_tests'])} tests"}


def skip(state: State) -> dict:
    return {"outcome": "left as is, nothing changed"}


builder = StateGraph(State)
builder.add_node("detect", detect)
builder.add_node("ask_approval", ask_approval)
builder.add_node("quarantine", quarantine)
builder.add_node("skip", skip)
builder.add_edge(START, "detect")
builder.add_edge("detect", "ask_approval")
builder.add_conditional_edges("ask_approval", route, ["quarantine", "skip"])
builder.add_edge("quarantine", END)
builder.add_edge("skip", END)

app = builder.compile(checkpointer=InMemorySaver())

if __name__ == "__main__":
    cfg = {"configurable": {"thread_id": "flaky-review-1"}}

    first = app.invoke({}, cfg)
    pause = first["__interrupt__"][0].value
    print("Graph paused. It is asking:")
    print(" ", pause["question"])
    for t in pause["tests"]:
        print("   -", t)
    print("Paused before node:", app.get_state(cfg).next)

    answer = sys.argv[1] if len(sys.argv) > 1 else input("\nYour answer: ")
    final = app.invoke(Command(resume=answer), cfg)
    print("\nResumed with", repr(answer), "->", final["outcome"])

Expected output

output

Graph paused. It is asking:
  Quarantine these tests? (yes/no)
   - login_redirect
   - search_autocomplete
Paused before node: ('ask_approval',)

Resumed with 'yes' -> quarantined 2 tests