Breakpoints and Approvals

Capsule overview

In the previous capsule you learned interrupt() — a way to pause execution inside a node, exactly where you decide. But there's another way to pause: breakpoints. Breakpoints pause execution between nodes — before or after a specific node runs. You don't need to modify the node's code. You configure them when you compile the graph.

What's the practical difference? With interrupt() you control the exact point inside a node's logic. With breakpoints, you say "before this node runs, pause" or "after this node runs, pause." It's an architecture-level decision, not an implementation-level one.

Combined with Command, breakpoints let you build complete approval flows: the agent reaches a critical point, pauses, the human inspects what the agent plans to do, and decides whether to continue, redirect, or abort. All without touching the nodes' internal code.

Breakpoints vs interrupt(): two tools, two purposes

Before we look at code, the distinction:

Think of it this way:

• interrupt() → "Pause here, at this specific line, and ask the user something"
• interrupt_before → "Before this node runs, pause for approval"
• interrupt_after → "After this node runs, pause for review"

interrupt_before: pausing BEFORE a node

The most common case: you have a node that runs a dangerous, expensive, or irreversible action. You want the human to approve before that node runs.

from dotenv import load_dotenv
load_dotenv()

from typing import TypedDict, Annotated
import operator
from langgraph.graph import StateGraph, START, END
from langgraph.checkpoint.memory import MemorySaver

class State(TypedDict):
    task: str
    plan: str
    result: str
    log: Annotated[list[str], operator.add]

def planner(state: State) -> dict:
    plan = f"Plan for '{state['task']}': 1) Gather data, 2) Analyze, 3) Execute action"
    return {"plan": plan, "log": ["planner_done"]}

def dangerous_action(state: State) -> dict:
    result = f"Action executed per plan: {state['plan'][:50]}..."
    return {"result": result, "log": ["action_executed"]}

graph_builder = StateGraph(State)
graph_builder.add_node("planner", planner)
graph_builder.add_node("dangerous_action", dangerous_action)

graph_builder.add_edge(START, "planner")
graph_builder.add_edge("planner", "dangerous_action")
graph_builder.add_edge("dangerous_action", END)

checkpointer = MemorySaver()
graph = graph_builder.compile(
    checkpointer=checkpointer,
    interrupt_before=["dangerous_action"],
)

config = {"configurable": {"thread_id": "approval-001"}}

print("=== Step 1: Run until the breakpoint ===\n")
result = graph.invoke(
    {"task": "Send a mass email to 10,000 customers", "plan": "", "result": "", "log": []},
    config,
)

state = graph.get_state(config)
print(f"Plan generated: {state.values['plan']}")
print(f"Next node: {state.next}")
print(f"Result: '{state.values['result']}'")
print(f"Log: {state.values['log']}")

print("\n=== Step 2: Human approves → continue ===\n")
result = graph.invoke(None, config)

state = graph.get_state(config)
print(f"Result: {state.values['result']}")
print(f"Log: {state.values['log']}")
print(f"Next node: {state.next}")
# Expected output:
# === Step 1: Run until the breakpoint ===
#
# Plan generated: Plan for 'Send a mass email to 10,000 customers': 1) Gather data, 2) Analyze, 3) Execute action
# Next node: ('dangerous_action',)
# Result: ''
# Log: ['planner_done']
#
# === Step 2: Human approves → continue ===
#
# Result: Action executed per plan: Plan for 'Send a mass email to 10,000 cus...
# Log: ['planner_done', 'action_executed']
# Next node: ()

The mechanics:

1. interrupt_before=["dangerous_action"] — at compile time, you declare that the graph must pause before running dangerous_action
2. The first invoke runs planner and stops before dangerous_action
3. state.next shows ('dangerous_action',) — the node is pending
4. The second invoke(None, config) resumes: it runs dangerous_action and finishes

The dangerous_action node has no interruption code at all. It doesn't know there's a breakpoint. That decision lives in the graph's compilation.

interrupt_after: pausing AFTER a node

Sometimes you want a node to run and then pause so the human can review the result before continuing:

from dotenv import load_dotenv
load_dotenv()

from typing import TypedDict, Annotated
import operator
from langgraph.graph import StateGraph, START, END
from langgraph.checkpoint.memory import MemorySaver

class State(TypedDict):
    query: str
    analysis: str
    report: str
    log: Annotated[list[str], operator.add]

def analyze(state: State) -> dict:
    analysis = f"Analysis of '{state['query']}': 3 trends identified, 2 risks detected"
    return {"analysis": analysis, "log": ["analysis_done"]}

def generate_report(state: State) -> dict:
    report = f"FINAL REPORT: {state['analysis']}"
    return {"report": report, "log": ["report_done"]}

graph_builder = StateGraph(State)
graph_builder.add_node("analyze", analyze)
graph_builder.add_node("generate_report", generate_report)

graph_builder.add_edge(START, "analyze")
graph_builder.add_edge("analyze", "generate_report")
graph_builder.add_edge("generate_report", END)

checkpointer = MemorySaver()
graph = graph_builder.compile(
    checkpointer=checkpointer,
    interrupt_after=["analyze"],
)

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

print("=== Step 1: Run the analysis and pause for review ===\n")
graph.invoke(
    {"query": "State of the AI market 2025", "analysis": "", "report": "", "log": []},
    config,
)

state = graph.get_state(config)
print(f"Analysis complete: {state.values['analysis']}")
print(f"Next node: {state.next}")
print(f"Report: '{state.values['report']}'")

print("\n=== Step 2: Human reviews the analysis → approves ===\n")
result = graph.invoke(None, config)

state = graph.get_state(config)
print(f"Report generated: {state.values['report']}")
print(f"Full log: {state.values['log']}")
# Expected output:
# === Step 1: Run the analysis and pause for review ===
#
# Analysis complete: Analysis of 'State of the AI market 2025': 3 trends identified, 2 risks detected
# Next node: ('generate_report',)
# Report: ''
#
# === Step 2: Human reviews the analysis → approves ===
#
# Report generated: FINAL REPORT: Analysis of 'State of the AI market 2025': 3 trends identified, 2 risks detected
# Full log: ['analysis_done', 'report_done']

With interrupt_after=["analyze"]:

1. The analyze node runs to completion
2. The graph pauses after analyze, before moving to generate_report
3. The human can inspect the analysis and decide whether to proceed

The full approval flow: approve, reject, redirect

Pausing the graph is only the first step. The powerful part is what you do during the pause. There are three fundamental actions:

from dotenv import load_dotenv
load_dotenv()

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

class State(TypedDict):
    task: str
    plan: str
    result: str
    status: str
    log: Annotated[list[str], operator.add]

def create_plan(state: State) -> dict:
    plan = f"Plan: run '{state['task']}' with 3 automated steps"
    return {"plan": plan, "status": "plan_ready", "log": ["plan_created"]}

def execute_plan(state: State) -> dict:
    result = f"Executed successfully: {state['plan']}"
    return {"result": result, "status": "executed", "log": ["executed"]}

def handle_rejection(state: State) -> dict:
    return {"result": "Task cancelled by the user", "status": "rejected", "log": ["rejected"]}

graph_builder = StateGraph(State)
graph_builder.add_node("create_plan", create_plan)
graph_builder.add_node("execute_plan", execute_plan)
graph_builder.add_node("handle_rejection", handle_rejection)

graph_builder.add_edge(START, "create_plan")
graph_builder.add_edge("create_plan", "execute_plan")
graph_builder.add_edge("execute_plan", END)
graph_builder.add_edge("handle_rejection", END)

checkpointer = MemorySaver()
graph = graph_builder.compile(
    checkpointer=checkpointer,
    interrupt_before=["execute_plan"],
)

print("=== SCENARIO 1: Approve ===\n")

config_approve = {"configurable": {"thread_id": "approve-flow"}}
graph.invoke(
    {"task": "Publish blog article", "plan": "", "result": "", "status": "", "log": []},
    config_approve,
)

state = graph.get_state(config_approve)
print(f"Plan: {state.values['plan']}")
print(f"Next: {state.next}")

print("→ Human approves: we continue")
result = graph.invoke(None, config_approve)
print(f"Result: {graph.get_state(config_approve).values['result']}")

print("\n=== SCENARIO 2: Reject (redirect to another node) ===\n")

config_reject = {"configurable": {"thread_id": "reject-flow"}}
graph.invoke(
    {"task": "Drop the production database", "plan": "", "result": "", "status": "", "log": []},
    config_reject,
)

state = graph.get_state(config_reject)
print(f"Plan: {state.values['plan']}")
print(f"Next: {state.next}")

print("→ Human rejects: we redirect to handle_rejection")
graph.update_state(config_reject, None, as_node="execute_plan")
graph.update_state(
    config_reject,
    {"status": "redirected", "log": ["human_rejected"]},
    as_node="handle_rejection",
)

final_state = graph.get_state(config_reject)
print(f"Result: {final_state.values['result']}")
print(f"Status: {final_state.values['status']}")
print(f"Log: {final_state.values['log']}")
# Expected output:
# === SCENARIO 1: Approve ===
#
# Plan: Plan: run 'Publish blog article' with 3 automated steps
# Next: ('execute_plan',)
# → Human approves: we continue
# Result: Executed successfully: Plan: run 'Publish blog article' with 3 automated steps
#
# === SCENARIO 2: Reject (redirect to another node) ===
#
# Plan: Plan: run 'Drop the production database' with 3 automated steps
# Next: ('execute_plan',)
# → Human rejects: we redirect to handle_rejection
# Result: Task cancelled by the user
# Status: redirected
# Log: ['plan_created', 'human_rejected']

The three paths:

• ✅ Approve: graph.invoke(None, config) — the graph continues normally
• ❌ Reject and redirect: graph.update_state() with as_node to simulate a different node answering, changing the flow
• ⚠️ Approve with modifications: edit the state before continuing (you'll see this in detail in capsule 04)

Command: sending instructions on resume

When you use interrupt() inside a node, the value of Command(resume=...) goes straight back to the variable that receives the interrupt(). With breakpoints, the mechanism is different: you use graph.invoke(None, config) to continue, or graph.update_state() to modify the state before continuing.

But if you combine breakpoints with interrupt() inside the node, you get the best of both worlds:

from dotenv import load_dotenv
load_dotenv()

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

class State(TypedDict):
    task: str
    plan: str
    human_feedback: str
    result: str
    log: Annotated[list[str], operator.add]

def create_plan(state: State) -> dict:
    plan = f"Automatic plan for: '{state['task']}'"
    return {"plan": plan, "log": ["plan_created"]}

def request_approval(state: State) -> dict:
    decision = interrupt({
        "question": "Do you approve this plan?",
        "plan": state["plan"],
        "options": ["approve", "approve_with_edits", "reject"],
    })

    if isinstance(decision, dict):
        approved = decision.get("approved", False)
        feedback = decision.get("feedback", "")
    else:
        approved = decision == "approve"
        feedback = ""

    if approved:
        return {
            "human_feedback": feedback if feedback else "Approved with no changes",
            "log": ["approved"],
        }
    else:
        return {
            "human_feedback": f"rejected: {feedback}",
            "result": "Cancelled by the user",
            "log": ["rejected"],
        }

def execute(state: State) -> dict:
    if "rejected" in state.get("human_feedback", ""):
        return {"result": "Not executed — rejected", "log": ["skipped"]}
    result = f"Executed: {state['plan']} | Feedback: {state['human_feedback']}"
    return {"result": result, "log": ["executed"]}

graph_builder = StateGraph(State)
graph_builder.add_node("create_plan", create_plan)
graph_builder.add_node("request_approval", request_approval)
graph_builder.add_node("execute", execute)

graph_builder.add_edge(START, "create_plan")
graph_builder.add_edge("create_plan", "request_approval")
graph_builder.add_edge("request_approval", "execute")
graph_builder.add_edge("execute", END)

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

config = {"configurable": {"thread_id": "command-demo"}}

print("=== Step 1: Run until it asks for approval ===\n")
graph.invoke(
    {"task": "Migrate the database", "plan": "", "human_feedback": "", "result": "", "log": []},
    config,
)

state = graph.get_state(config)
print(f"Plan: {state.values['plan']}")
print(f"Next: {state.next}")

print("\n=== Step 2: Human approves with feedback ===\n")
result = graph.invoke(
    Command(resume={"approved": True, "feedback": "Go ahead but take a backup first"}),
    config,
)

final = graph.get_state(config)
print(f"Feedback: {final.values['human_feedback']}")
print(f"Result: {final.values['result']}")
print(f"Log: {final.values['log']}")
# Expected output:
# === Step 1: Run until it asks for approval ===
#
# Plan: Automatic plan for: 'Migrate the database'
# Next: ('request_approval',)
#
# === Step 2: Human approves with feedback ===
#
# Feedback: Go ahead but take a backup first
# Result: Executed: Automatic plan for: 'Migrate the database' | Feedback: Go ahead but take a backup first
# Log: ['plan_created', 'approved', 'executed']

The interrupt() + Command(resume=) pattern lets you send structured data back into the node: not just "yes" or "no", but full objects with feedback, instructions, and extra context.

Multiple breakpoints: inspection at every critical step

You can set breakpoints on multiple nodes to create a pipeline with human checkpoints:

from dotenv import load_dotenv
load_dotenv()

from typing import TypedDict, Annotated
import operator
from langgraph.graph import StateGraph, START, END
from langgraph.checkpoint.memory import MemorySaver

class State(TypedDict):
    request: str
    research: str
    draft: str
    final: str
    log: Annotated[list[str], operator.add]

def research(state: State) -> dict:
    return {
        "research": f"Research on '{state['request']}': 5 sources found",
        "log": ["research_done"],
    }

def draft(state: State) -> dict:
    return {
        "draft": f"Draft based on: {state['research'][:40]}...",
        "log": ["draft_done"],
    }

def finalize(state: State) -> dict:
    return {
        "final": f"FINAL DOCUMENT: {state['draft']}",
        "log": ["finalized"],
    }

graph_builder = StateGraph(State)
graph_builder.add_node("research", research)
graph_builder.add_node("draft", draft)
graph_builder.add_node("finalize", finalize)

graph_builder.add_edge(START, "research")
graph_builder.add_edge("research", "draft")
graph_builder.add_edge("draft", "finalize")
graph_builder.add_edge("finalize", END)

checkpointer = MemorySaver()
graph = graph_builder.compile(
    checkpointer=checkpointer,
    interrupt_after=["research", "draft"],
)

config = {"configurable": {"thread_id": "multi-bp"}}

print("=== Step 1: Research runs, pauses for review ===\n")
graph.invoke(
    {"request": "AI competitive analysis", "research": "", "draft": "", "final": "", "log": []},
    config,
)
state = graph.get_state(config)
print(f"Research: {state.values['research']}")
print(f"Next: {state.next}")

print("\n=== Step 2: Human approves research → Draft runs, pauses ===\n")
graph.invoke(None, config)
state = graph.get_state(config)
print(f"Draft: {state.values['draft']}")
print(f"Next: {state.next}")

print("\n=== Step 3: Human approves draft → Finalize runs ===\n")
graph.invoke(None, config)
state = graph.get_state(config)
print(f"Final: {state.values['final']}")
print(f"Next: {state.next}")
print(f"Log: {state.values['log']}")
# Expected output:
# === Step 1: Research runs, pauses for review ===
#
# Research: Research on 'AI competitive analysis': 5 sources found
# Next: ('draft',)
#
# === Step 2: Human approves research → Draft runs, pauses ===
#
# Draft: Draft based on: Research on 'AI competitive analy...
# Next: ('finalize',)
#
# === Step 3: Human approves draft → Finalize runs ===
#
# Final: FINAL DOCUMENT: Draft based on: Research on 'AI competitive analy...
# Next: ()
# Log: ['research_done', 'draft_done', 'finalized']

The flow has two human checkpoints: one after research (are the sources good?) and another after draft (is the draft right?). Only after two approvals does the final document get generated.

When to use breakpoints vs interrupt()

The general rule:

In practice, you combine them. A production graph might have:

• interrupt_before=["send_email", "execute_trade"] — automatic gates for irreversible actions
• interrupt() inside a planning node — to ask the user for detailed feedback

Full workflow: Plan → Approve → Execute → Review → Deliver

A realistic production approval flow that combines several techniques:

from dotenv import load_dotenv
load_dotenv()

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

class WorkflowState(TypedDict):
    objective: str
    plan: str
    execution_result: str
    review_notes: str
    deliverable: str
    log: Annotated[list[str], operator.add]

def plan_step(state: WorkflowState) -> dict:
    plan = f"Plan for '{state['objective']}':\n"
    plan += "  1. Gather data from 3 sources\n"
    plan += "  2. Analyze trends\n"
    plan += "  3. Generate visualizations\n"
    plan += "  4. Compile the executive report"
    return {"plan": plan, "log": ["plan_created"]}

def approve_step(state: WorkflowState) -> dict:
    decision = interrupt({
        "message": "Review the plan before executing",
        "plan": state["plan"],
        "action_required": "Respond with {approved: true/false, notes: '...'}",
    })

    if decision.get("approved"):
        notes = decision.get("notes", "No additional notes")
        return {"review_notes": f"approved: {notes}", "log": ["plan_approved"]}
    else:
        return {
            "review_notes": f"rejected: {decision.get('notes', 'No reason given')}",
            "deliverable": "Workflow cancelled",
            "log": ["plan_rejected"],
        }

def execute_step(state: WorkflowState) -> dict:
    if "rejected" in state["review_notes"]:
        return {"execution_result": "Not executed", "log": ["execution_skipped"]}
    result = f"Executed successfully. Data from 3 sources processed. "
    result += f"Reviewer notes: {state['review_notes']}"
    return {"execution_result": result, "log": ["executed"]}

def deliver_step(state: WorkflowState) -> dict:
    if state.get("deliverable") == "Workflow cancelled":
        return {"log": ["cancelled"]}
    deliverable = f"DELIVERABLE: {state['execution_result'][:60]}..."
    return {"deliverable": deliverable, "log": ["delivered"]}

graph_builder = StateGraph(WorkflowState)
graph_builder.add_node("plan", plan_step)
graph_builder.add_node("approve", approve_step)
graph_builder.add_node("execute", execute_step)
graph_builder.add_node("deliver", deliver_step)

graph_builder.add_edge(START, "plan")
graph_builder.add_edge("plan", "approve")
graph_builder.add_edge("approve", "execute")
graph_builder.add_edge("execute", "deliver")
graph_builder.add_edge("deliver", END)

checkpointer = MemorySaver()
graph = graph_builder.compile(
    checkpointer=checkpointer,
    interrupt_after=["execute"],
)

config = {"configurable": {"thread_id": "full-workflow"}}

print("=== STEP 1: Plan gets generated, approval requested ===\n")
graph.invoke(
    {
        "objective": "Quarterly AI sales report",
        "plan": "", "execution_result": "",
        "review_notes": "", "deliverable": "", "log": [],
    },
    config,
)

state = graph.get_state(config)
print(f"Status: waiting for approval at node '{state.next}'")

print("\n=== STEP 2: Human approves with notes ===\n")
graph.invoke(
    Command(resume={"approved": True, "notes": "Include Q3 data too"}),
    config,
)

state = graph.get_state(config)
print(f"Review: {state.values['review_notes']}")
print(f"Execution: {state.values['execution_result'][:80]}...")
print(f"Next: {state.next}")

print("\n=== STEP 3: Human reviews the execution → continue to delivery ===\n")
result = graph.invoke(None, config)

final = graph.get_state(config)
print(f"Deliverable: {final.values['deliverable']}")
print(f"Full log: {final.values['log']}")
# Expected output:
# === STEP 1: Plan gets generated, approval requested ===
#
# Status: waiting for approval at node '('approve',)'
#
# === STEP 2: Human approves with notes ===
#
# Review: approved: Include Q3 data too
# Execution: Executed successfully. Data from 3 sources processed. Reviewer notes: a...
# Next: ('deliver',)
#
# === STEP 3: Human reviews the execution → continue to delivery ===
#
# Deliverable: DELIVERABLE: Executed successfully. Data from 3 sources processed....
# Full log: ['plan_created', 'plan_approved', 'executed', 'delivered']

This workflow has two human control points:

1. Plan approval: interrupt() inside approve — the human sends structured data with Command(resume=)
2. Execution review: interrupt_after=["execute"] — the human reviews the results before delivery

Combining interrupt_before and interrupt_after

You can use both in the same graph:

from dotenv import load_dotenv
load_dotenv()

from typing import TypedDict, Annotated
import operator
from langgraph.graph import StateGraph, START, END
from langgraph.checkpoint.memory import MemorySaver

class State(TypedDict):
    data: str
    analyzed: str
    transformed: str
    saved: str
    log: Annotated[list[str], operator.add]

def analyze(state: State) -> dict:
    return {"analyzed": f"Analysis of: {state['data']}", "log": ["analyzed"]}

def transform(state: State) -> dict:
    return {"transformed": f"Transformed: {state['analyzed']}", "log": ["transformed"]}

def save_to_db(state: State) -> dict:
    return {"saved": f"Saved to DB: {state['transformed'][:30]}...", "log": ["saved"]}

graph_builder = StateGraph(State)
graph_builder.add_node("analyze", analyze)
graph_builder.add_node("transform", transform)
graph_builder.add_node("save_to_db", save_to_db)

graph_builder.add_edge(START, "analyze")
graph_builder.add_edge("analyze", "transform")
graph_builder.add_edge("transform", "save_to_db")
graph_builder.add_edge("save_to_db", END)

checkpointer = MemorySaver()
graph = graph_builder.compile(
    checkpointer=checkpointer,
    interrupt_after=["analyze"],
    interrupt_before=["save_to_db"],
)

config = {"configurable": {"thread_id": "combo-bp"}}

print("=== Step 1: Analyze runs, pauses after ===")
graph.invoke({"data": "Q4 sales", "analyzed": "", "transformed": "", "saved": "", "log": []}, config)
state = graph.get_state(config)
print(f"  Analyzed: {state.values['analyzed']}")
print(f"  Next: {state.next}")

print("\n=== Step 2: We approve the analysis → Transform runs → Pauses before save_to_db ===")
graph.invoke(None, config)
state = graph.get_state(config)
print(f"  Transformed: {state.values['transformed']}")
print(f"  Next: {state.next}")

print("\n=== Step 3: We approve the save → save_to_db runs ===")
graph.invoke(None, config)
state = graph.get_state(config)
print(f"  Saved: {state.values['saved']}")
print(f"  Log: {state.values['log']}")
# Expected output:
# === Step 1: Analyze runs, pauses after ===
#   Analyzed: Analysis of: Q4 sales
#   Next: ('transform',)
#
# === Step 2: We approve the analysis → Transform runs → Pauses before save_to_db ===
#   Transformed: Transformed: Analysis of: Q4 sales
#   Next: ('save_to_db',)
#
# === Step 3: We approve the save → save_to_db runs ===
#   Saved: Saved to DB: Transformed: Analysis of: Q...
#   Log: ['analyzed', 'transformed', 'saved']

Two different breakpoints:

• interrupt_after=["analyze"] — "let me see the analysis before transforming"
• interrupt_before=["save_to_db"] — "let me approve before writing to the database"

Troubleshooting

Problem 1: "The graph doesn't stop at the breakpoint"

Symptom: You set interrupt_before=["my_node"] but the graph runs everything without pausing.

Cause: The node name doesn't match the one you registered in add_node(), or the graph has no checkpointer.

Fix: Check that the name is identical to the one you used in graph_builder.add_node("exact_name", fn) and that you compiled with a checkpointer:

graph_builder.add_node("dangerous_action", my_function)

graph = graph_builder.compile(
    checkpointer=MemorySaver(),
    interrupt_before=["dangerous_action"],
)

Problem 2: "graph.invoke(None, config) doesn't resume"

Symptom: After the breakpoint, invoke(None, config) does nothing or throws an error.

Cause: The thread_id in the config doesn't match the one from the original run.

Fix: Use exactly the same config:

config = {"configurable": {"thread_id": "my-thread"}}
graph.invoke(input, config)
# ... breakpoint ...
graph.invoke(None, config)  # same config

Problem 3: "interrupt_before and interrupt_after don't work together"

Symptom: Only one of the two breakpoints pauses.

Cause: This isn't a problem — it's the expected behavior. If you have interrupt_after=["A"] and interrupt_before=["B"], and A → B is a direct transition, the graph pauses once (after A, which is the same as before B).

Fix: You don't need both for the same transition. Use interrupt_after=["A"] OR interrupt_before=["B"], not both.

Problem 4: "I want to cancel the run, not continue"

Symptom: The graph is paused and you want to abort entirely.

Cause: There's no explicit "abort" — but you can modify the state to redirect.

Fix: Use update_state to mark the task as cancelled and let the graph finish cleanly:

graph.update_state(config, {"status": "cancelled"})
graph.invoke(None, config)

Exercises

Exercise 1: A basic breakpoint with interrupt_before (Easy)

Build a 3-node graph: prepare → send_email → confirm. Set a breakpoint before send_email. Run the graph, verify it pauses before sending, inspect the state, and then continue.

from dotenv import load_dotenv
load_dotenv()

from typing import TypedDict, Annotated
import operator
from langgraph.graph import StateGraph, START, END
from langgraph.checkpoint.memory import MemorySaver

class State(TypedDict):
    recipient: str
    subject: str
    body: str
    sent: bool
    log: Annotated[list[str], operator.add]

def prepare(state: State) -> dict:
    body = f"Dear user, this is an email about: {state['subject']}"
    return {"body": body, "log": ["prepared"]}

def send_email(state: State) -> dict:
    return {"sent": True, "log": ["email_sent"]}

def confirm(state: State) -> dict:
    return {"log": [f"confirmed_to_{state['recipient']}"]}

graph_builder = StateGraph(State)
graph_builder.add_node("prepare", prepare)
graph_builder.add_node("send_email", send_email)
graph_builder.add_node("confirm", confirm)

graph_builder.add_edge(START, "prepare")
graph_builder.add_edge("prepare", "send_email")
graph_builder.add_edge("send_email", "confirm")
graph_builder.add_edge("confirm", END)

checkpointer = MemorySaver()
graph = graph_builder.compile(
    checkpointer=checkpointer,
    interrupt_before=["send_email"],
)

config = {"configurable": {"thread_id": "email-bp"}}
graph.invoke(
    {"recipient": "customer@company.com", "subject": "Special offer", "body": "", "sent": False, "log": []},
    config,
)

state = graph.get_state(config)
print(f"Body prepared: {state.values['body']}")
print(f"Sent already?: {state.values['sent']}")
print(f"Next node: {state.next}")

print("\n→ We approve the send")
graph.invoke(None, config)

state = graph.get_state(config)
print(f"Sent?: {state.values['sent']}")
print(f"Log: {state.values['log']}")
# Expected output:
# Body prepared: Dear user, this is an email about: Special offer
# Sent already?: False
# Next node: ('send_email',)
#
# → We approve the send
# Sent?: True
# Log: ['prepared', 'email_sent', 'confirmed_to_customer@company.com']

Exercise 2: Review with interrupt_after (Easy)

Build a graph where a generate_summary node produces a summary and then pauses so the human can review it. If the human approves, the publish node publishes it. Use interrupt_after on generate_summary.

from dotenv import load_dotenv
load_dotenv()

from typing import TypedDict, Annotated
import operator
from langgraph.graph import StateGraph, START, END
from langgraph.checkpoint.memory import MemorySaver

class State(TypedDict):
    topic: str
    summary: str
    published: bool
    log: Annotated[list[str], operator.add]

def generate_summary(state: State) -> dict:
    summary = f"Summary of '{state['topic']}': 3 key points identified."
    return {"summary": summary, "log": ["summary_generated"]}

def publish(state: State) -> dict:
    return {"published": True, "log": ["published"]}

graph_builder = StateGraph(State)
graph_builder.add_node("generate_summary", generate_summary)
graph_builder.add_node("publish", publish)

graph_builder.add_edge(START, "generate_summary")
graph_builder.add_edge("generate_summary", "publish")
graph_builder.add_edge("publish", END)

checkpointer = MemorySaver()
graph = graph_builder.compile(
    checkpointer=checkpointer,
    interrupt_after=["generate_summary"],
)

config = {"configurable": {"thread_id": "review-summary"}}
graph.invoke(
    {"topic": "LangGraph HITL", "summary": "", "published": False, "log": []},
    config,
)

state = graph.get_state(config)
print(f"Summary generated: {state.values['summary']}")
print(f"Published?: {state.values['published']}")
print(f"Next: {state.next}")

print("\n→ Human reviews and approves")
graph.invoke(None, config)

state = graph.get_state(config)
print(f"Published?: {state.values['published']}")
print(f"Log: {state.values['log']}")
# Expected output:
# Summary generated: Summary of 'LangGraph HITL': 3 key points identified.
# Published?: False
# Next: ('publish',)
#
# → Human reviews and approves
# Published?: True
# Log: ['summary_generated', 'published']

Exercise 3: Approval with Command(resume) (Medium)

Build a workflow where a propose_action node generates a proposal and uses interrupt() to ask for approval. The human answers with Command(resume={"approved": True, "priority": "high"}). The next node (execute_action) uses the priority to decide how to execute.

from dotenv import load_dotenv
load_dotenv()

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

class State(TypedDict):
    task: str
    proposal: str
    priority: str
    result: str
    log: Annotated[list[str], operator.add]

def propose_action(state: State) -> dict:
    proposal = f"Proposal: automate '{state['task']}'"

    decision = interrupt({
        "message": "Review this proposal",
        "proposal": proposal,
        "respond_with": "{approved: bool, priority: 'low'|'medium'|'high'}",
    })

    return {
        "proposal": proposal,
        "priority": decision.get("priority", "medium"),
        "log": [f"proposal_{'approved' if decision.get('approved') else 'rejected'}"],
    }

def execute_action(state: State) -> dict:
    if state["priority"] == "high":
        method = "immediate execution with full resources"
    elif state["priority"] == "medium":
        method = "standard execution"
    else:
        method = "background execution, low priority"

    result = f"{state['proposal']} → {method}"
    return {"result": result, "log": ["executed"]}

graph_builder = StateGraph(State)
graph_builder.add_node("propose_action", propose_action)
graph_builder.add_node("execute_action", execute_action)

graph_builder.add_edge(START, "propose_action")
graph_builder.add_edge("propose_action", "execute_action")
graph_builder.add_edge("execute_action", END)

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

config = {"configurable": {"thread_id": "priority-approval"}}

graph.invoke(
    {"task": "Deploy v2.0", "proposal": "", "priority": "", "result": "", "log": []},
    config,
)

state = graph.get_state(config)
print(f"Waiting at: {state.next}")

result = graph.invoke(
    Command(resume={"approved": True, "priority": "high"}),
    config,
)

final = graph.get_state(config)
print(f"Priority: {final.values['priority']}")
print(f"Result: {final.values['result']}")
print(f"Log: {final.values['log']}")
# Expected output:
# Waiting at: ('propose_action',)
# Priority: high
# Result: Proposal: automate 'Deploy v2.0' → immediate execution with full resources
# Log: ['proposal_approved', 'executed']

Exercise 4: A pipeline with multiple breakpoints (Medium)

Build a 4-node pipeline: collect_data → clean_data → train_model → deploy_model. Set breakpoints after clean_data (to review data quality) and before deploy_model (to approve the deployment). Run the whole flow with the 3 invocations it needs.

from dotenv import load_dotenv
load_dotenv()

from typing import TypedDict, Annotated
import operator
from langgraph.graph import StateGraph, START, END
from langgraph.checkpoint.memory import MemorySaver

class State(TypedDict):
    dataset_name: str
    raw_rows: int
    clean_rows: int
    model_accuracy: float
    deployed: bool
    log: Annotated[list[str], operator.add]

def collect_data(state: State) -> dict:
    return {"raw_rows": 10000, "log": ["data_collected"]}

def clean_data(state: State) -> dict:
    clean = int(state["raw_rows"] * 0.85)
    return {"clean_rows": clean, "log": [f"cleaned_{clean}_rows"]}

def train_model(state: State) -> dict:
    accuracy = 0.92 if state["clean_rows"] > 5000 else 0.75
    return {"model_accuracy": accuracy, "log": [f"trained_accuracy_{accuracy}"]}

def deploy_model(state: State) -> dict:
    return {"deployed": True, "log": ["deployed"]}

graph_builder = StateGraph(State)
graph_builder.add_node("collect_data", collect_data)
graph_builder.add_node("clean_data", clean_data)
graph_builder.add_node("train_model", train_model)
graph_builder.add_node("deploy_model", deploy_model)

graph_builder.add_edge(START, "collect_data")
graph_builder.add_edge("collect_data", "clean_data")
graph_builder.add_edge("clean_data", "train_model")
graph_builder.add_edge("train_model", "deploy_model")
graph_builder.add_edge("deploy_model", END)

checkpointer = MemorySaver()
graph = graph_builder.compile(
    checkpointer=checkpointer,
    interrupt_after=["clean_data"],
    interrupt_before=["deploy_model"],
)

config = {"configurable": {"thread_id": "ml-pipeline"}}

print("=== Phase 1: Collect + Clean → Pause for data review ===")
graph.invoke(
    {"dataset_name": "sales_2025", "raw_rows": 0, "clean_rows": 0,
     "model_accuracy": 0.0, "deployed": False, "log": []},
    config,
)
state = graph.get_state(config)
print(f"  Raw: {state.values['raw_rows']} → Clean: {state.values['clean_rows']}")
print(f"  Next: {state.next}")

print("\n=== Phase 2: Train → Pause before deploy ===")
graph.invoke(None, config)
state = graph.get_state(config)
print(f"  Accuracy: {state.values['model_accuracy']}")
print(f"  Next: {state.next}")

print("\n=== Phase 3: Deploy approved ===")
graph.invoke(None, config)
state = graph.get_state(config)
print(f"  Deployed: {state.values['deployed']}")
print(f"  Log: {state.values['log']}")
# Expected output:
# === Phase 1: Collect + Clean → Pause for data review ===
#   Raw: 10000 → Clean: 8500
#   Next: ('train_model',)
#
# === Phase 2: Train → Pause before deploy ===
#   Accuracy: 0.92
#   Next: ('deploy_model',)
#
# === Phase 3: Deploy approved ===
#   Deployed: True
#   Log: ['data_collected', 'cleaned_8500_rows', 'trained_accuracy_0.92', 'deployed']

Exercise 5: Approval workflow with rejection and redirect (Advanced)

Build a workflow with 4 nodes: draft_email → review_email → send_email → log_sent. The review_email node uses interrupt() to ask for approval. If the human rejects, the flow must go back to draft_email with feedback to redo the draft. Implement a cycle: draft → review → (reject → draft again) → review → (approve) → send → log.

from dotenv import load_dotenv
load_dotenv()

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

class State(TypedDict):
    recipient: str
    draft: str
    feedback: str
    revision_count: int
    sent: bool
    log: Annotated[list[str], operator.add]

def draft_email(state: State) -> dict:
    revision = state.get("revision_count", 0)
    feedback = state.get("feedback", "")

    if revision == 0:
        draft = f"Draft v1 for {state['recipient']}: Collaboration proposal"
    else:
        draft = f"Draft v{revision + 1} for {state['recipient']}: "
        draft += f"(Incorporating feedback: {feedback[:40]}...)"

    return {
        "draft": draft,
        "revision_count": revision + 1,
        "log": [f"draft_v{revision + 1}"],
    }

def review_email(state: State) -> dict:
    decision = interrupt({
        "message": "Review this email draft",
        "draft": state["draft"],
        "options": "Respond {approved: true} or {approved: false, feedback: '...'}",
    })

    if decision.get("approved"):
        return {"log": ["review_approved"]}
    else:
        return {
            "feedback": decision.get("feedback", "Needs improvement"),
            "log": ["review_rejected"],
        }

def send_email(state: State) -> dict:
    return {"sent": True, "log": ["email_sent"]}

def log_sent(state: State) -> dict:
    return {"log": [f"logged_email_to_{state['recipient']}"]}

def route_after_review(state: State) -> str:
    if state["log"][-1] == "review_rejected":
        return "draft_email"
    return "send_email"

graph_builder = StateGraph(State)
graph_builder.add_node("draft_email", draft_email)
graph_builder.add_node("review_email", review_email)
graph_builder.add_node("send_email", send_email)
graph_builder.add_node("log_sent", log_sent)

graph_builder.add_edge(START, "draft_email")
graph_builder.add_edge("draft_email", "review_email")
graph_builder.add_conditional_edges("review_email", route_after_review)
graph_builder.add_edge("send_email", "log_sent")
graph_builder.add_edge("log_sent", END)

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

config = {"configurable": {"thread_id": "email-cycle"}}

print("=== Round 1: Draft → Review ===")
graph.invoke(
    {"recipient": "ceo@company.com", "draft": "", "feedback": "",
     "revision_count": 0, "sent": False, "log": []},
    config,
)
state = graph.get_state(config)
print(f"Draft: {state.values['draft']}")

print("\n=== Round 1: Human rejects ===")
graph.invoke(
    Command(resume={"approved": False, "feedback": "Tone is too informal, make it more professional"}),
    config,
)
state = graph.get_state(config)
print(f"Feedback: {state.values['feedback']}")
print(f"Current draft: {state.values['draft']}")
print(f"Revisions: {state.values['revision_count']}")

print("\n=== Round 2: Human approves ===")
graph.invoke(
    Command(resume={"approved": True}),
    config,
)
final = graph.get_state(config)
print(f"Sent?: {final.values['sent']}")
print(f"Full log: {final.values['log']}")
# Expected output:
# === Round 1: Draft → Review ===
# Draft: Draft v1 for ceo@company.com: Collaboration proposal
#
# === Round 1: Human rejects ===
# Feedback: Tone is too informal, make it more professional
# Current draft: Draft v2 for ceo@company.com: (Incorporating feedback: Tone is too informal, make it more p...)
# Revisions: 2
#
# === Round 2: Human approves ===
# Sent?: True
# Full log: ['draft_v1', 'review_rejected', 'draft_v2', 'review_approved', 'email_sent', 'logged_email_to_ceo@company.com']

Exercise 6: Comparing breakpoints vs interrupt() in the same graph (Advanced)

Build a graph with 4 nodes: research → validate → execute → report. Use interrupt_before on execute (a compile-level approval gate) and interrupt() inside validate to ask the human to confirm the data. Implement both pause-and-resume flows.

from dotenv import load_dotenv
load_dotenv()

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

class State(TypedDict):
    topic: str
    findings: str
    validated: bool
    human_confirmation: str
    execution_result: str
    report: str
    log: Annotated[list[str], operator.add]

def research(state: State) -> dict:
    findings = f"3 findings on '{state['topic']}': upward trend, moderate risk, opportunity in Q2"
    return {"findings": findings, "log": ["researched"]}

def validate(state: State) -> dict:
    confirmation = interrupt({
        "message": "Validate these findings before continuing",
        "findings": state["findings"],
        "action": "Respond {confirmed: true/false, notes: '...'}",
    })

    confirmed = confirmation.get("confirmed", False)
    notes = confirmation.get("notes", "")
    return {
        "validated": confirmed,
        "human_confirmation": notes if notes else ("Validated" if confirmed else "Rejected"),
        "log": [f"validation_{'passed' if confirmed else 'failed'}"],
    }

def execute(state: State) -> dict:
    if not state["validated"]:
        return {"execution_result": "Not executed — validation failed", "log": ["skipped"]}
    return {
        "execution_result": f"Executed with confirmation: {state['human_confirmation']}",
        "log": ["executed"],
    }

def report(state: State) -> dict:
    return {"report": f"Report: {state['execution_result']}", "log": ["reported"]}

graph_builder = StateGraph(State)
graph_builder.add_node("research", research)
graph_builder.add_node("validate", validate)
graph_builder.add_node("execute", execute)
graph_builder.add_node("report", report)

graph_builder.add_edge(START, "research")
graph_builder.add_edge("research", "validate")
graph_builder.add_edge("validate", "execute")
graph_builder.add_edge("execute", "report")
graph_builder.add_edge("report", END)

checkpointer = MemorySaver()
graph = graph_builder.compile(
    checkpointer=checkpointer,
    interrupt_before=["execute"],
)

config = {"configurable": {"thread_id": "combo-exercise"}}

print("=== Step 1: Research completes, validate asks for input (interrupt) ===")
graph.invoke(
    {"topic": "AI market 2025", "findings": "", "validated": False,
     "human_confirmation": "", "execution_result": "", "report": "", "log": []},
    config,
)
state = graph.get_state(config)
print(f"  Waiting at: {state.next}")

print("\n=== Step 2: Human confirms the findings (Command resume) ===")
graph.invoke(
    Command(resume={"confirmed": True, "notes": "Data verified against an internal source"}),
    config,
)
state = graph.get_state(config)
print(f"  Validated: {state.values['validated']}")
print(f"  Confirmation: {state.values['human_confirmation']}")
print(f"  Waiting at: {state.next}")

print("\n=== Step 3: Human approves the execution (breakpoint) ===")
graph.invoke(None, config)
final = graph.get_state(config)
print(f"  Result: {final.values['execution_result']}")
print(f"  Report: {final.values['report']}")
print(f"  Log: {final.values['log']}")
# Expected output:
# === Step 1: Research completes, validate asks for input (interrupt) ===
#   Waiting at: ('validate',)
#
# === Step 2: Human confirms the findings (Command resume) ===
#   Validated: True
#   Confirmation: Data verified against an internal source
#   Waiting at: ('execute',)
#
# === Step 3: Human approves the execution (breakpoint) ===
#   Result: Executed with confirmation: Data verified against an internal source
#   Report: Report: Executed with confirmation: Data verified against an internal source
#   Log: ['researched', 'validation_passed', 'executed', 'reported']

Summary

In this capsule you learned:

• Breakpoints pause between nodes, without modifying the node's code. You configure them at compile time: interrupt_before=["node"] pauses before, interrupt_after=["node"] pauses after. The node doesn't know there's a breakpoint
• interrupt() pauses inside a node, at the exact point you choose. It's more granular than breakpoints and lets you collect data from the user with Command(resume=)
• The approval flow has three paths: approve (graph.invoke(None, config)), reject and redirect (graph.update_state() with as_node), or approve with modifications (edit state before continuing)
• Command(resume=) lets you send structured data back to the node that called interrupt(). Not just "yes/no" — you can send objects with feedback, priority, instructions
• You can combine multiple breakpoints in the same graph to build pipelines with human checkpoints at every critical step
• Breakpoints + interrupt() complement each other: use breakpoints for architecture-level approval gates, use interrupt() for detailed interactions inside a node
• Everything requires a checkpointer. Without one, state is lost between the pause and the resume. Every HITL example must compile with checkpointer=MemorySaver() (development) or a durable checkpointer (production)

Next capsule: Editable State — not just approving or rejecting, but inspecting and modifying the agent's state during a pause. Fixing wrong data, adding context, changing the direction of the run.

Additional resources

1. LangGraph — Human-in-the-Loop — Official HITL concepts in LangGraph
2. How to add breakpoints — Practical guide to configuring interrupt_before and interrupt_after
3. How to wait for user input — Patterns for collecting human input during execution
4. LangGraph — Command — Documentation for the Command object used to send instructions to the graph
5. How to review tool calls — Approval pattern before executing tool calls

Module 9 — LangChain & LangGraph: From Chains to Agents