Langgraph Human In The Loop

from langgraph.types import interrupt, Commandfrom langgraph.checkpoint.memory import InMemorySaverfrom langgraph.graph import StateGraph, START, ENDfrom typing_extensions import TypedDict
class State(TypedDict):    approved: bool
def approval_node(state: State):    # Pause and ask for approval    approved = interrupt("Do you approve this action?")    # When resumed, Command(resume=...) returns that value here    return {"approved": approved}
checkpointer = InMemorySaver()graph = (    StateGraph(State)    .add_node("approval", approval_node)    .add_edge(START, "approval")    .add_edge("approval", END)    .compile(checkpointer=checkpointer))
config = {"configurable": {"thread_id": "thread-1"}}
# Initial run — hits interrupt and pausesresult = graph.invoke({"approved": False}, config)print(result["__interrupt__"])# [Interrupt(value='Do you approve this action?')]
# Resume with the human's responseresult = graph.invoke(Command(resume=True), config)print(result["approved"])  # True

from langgraph.types import interrupt, Commandfrom langgraph.graph import StateGraph, START, ENDfrom typing import Literalfrom typing_extensions import TypedDict
class EmailAgentState(TypedDict):    email_content: str    draft_response: str    classification: dict
def human_review(state: EmailAgentState) -> Command[Literal["send_reply", "__end__"]]:    """Pause for human review using interrupt and route based on decision."""    classification = state.get("classification", {})
    # interrupt() must come first — any code before it will re-run on resume    human_decision = interrupt({        "email_id": state.get("email_content", ""),        "draft_response": state.get("draft_response", ""),        "urgency": classification.get("urgency"),        "action": "Please review and approve/edit this response"    })
    # Process the human's decision    if human_decision.get("approved"):        return Command(            update={"draft_response": human_decision.get("edited_response", state.get("draft_response", ""))},            goto="send_reply"        )    else:        # Rejection — human will handle directly        return Command(update={}, goto=END)

from langgraph.types import interrupt
def get_age_node(state):    prompt = "What is your age?"
    while True:        answer = interrupt(prompt)
        # Validate the input        if isinstance(answer, int) and answer > 0:            break        else:            # Invalid input — ask again with a more specific prompt            prompt = f"'{answer}' is not a valid age. Please enter a positive number."
    return {"age": answer}

config = {"configurable": {"thread_id": "form-1"}}first = graph.invoke({"age": None}, config)# __interrupt__: "What is your age?"
retry = graph.invoke(Command(resume="thirty"), config)# __interrupt__: "'thirty' is not a valid age..."
final = graph.invoke(Command(resume=30), config)print(final["age"])  # 30

from typing import Annotated, TypedDictimport operatorfrom langgraph.checkpoint.memory import InMemorySaverfrom langgraph.graph import START, END, StateGraphfrom langgraph.types import Command, interrupt
class State(TypedDict):    vals: Annotated[list[str], operator.add]
def node_a(state):    answer = interrupt("question_a")    return {"vals": [f"a:{answer}"]}
def node_b(state):    answer = interrupt("question_b")    return {"vals": [f"b:{answer}"]}
graph = (    StateGraph(State)    .add_node("a", node_a)    .add_node("b", node_b)    .add_edge(START, "a")    .add_edge(START, "b")    .add_edge("a", END)    .add_edge("b", END)    .compile(checkpointer=InMemorySaver()))
config = {"configurable": {"thread_id": "1"}}
# Both parallel nodes hit interrupt() and pauseresult = graph.invoke({"vals": []}, config)# result["__interrupt__"] contains both Interrupt objects with IDs
# Resume all pending interrupts at once using a map of id -> valueresume_map = {    i.id: f"answer for {i.value}"    for i in result["__interrupt__"]}result = graph.invoke(Command(resume=resume_map), config)# result["vals"] = ["a:answer for question_a", "b:answer for question_b"]

# GOOD: Upsert is idempotent — safe before interruptdef node_a(state: State):    db.upsert_user(user_id=state["user_id"], status="pending_approval")    approved = interrupt("Approve this change?")    return {"approved": approved}
# GOOD: Side effect AFTER interrupt — only runs oncedef node_a(state: State):    approved = interrupt("Approve this change?")    if approved:        db.create_audit_log(user_id=state["user_id"], action="approved")    return {"approved": approved}
# BAD: Insert creates duplicates on each resume!def node_a(state: State):    audit_id = db.create_audit_log({  # Runs again on resume!        "user_id": state["user_id"],        "action": "pending_approval",    })    approved = interrupt("Approve this change?")    return {"approved": approved}

def node_in_parent_graph(state: State):    some_code()  # <-- Re-executes on resume    subgraph_result = subgraph.invoke(some_input)    # ...
def node_in_subgraph(state: State):    some_other_code()  # <-- Also re-executes on resume    result = interrupt("What's your name?")    # ...

# WRONGgraph = builder.compile()
# CORRECTgraph = builder.compile(checkpointer=InMemorySaver())

# WRONGgraph.invoke({"resume_data": "approve"}, config)
# CORRECTgraph.invoke(Command(resume="approve"), config)