Control Flow & Human-in-the-Loop
routing, pausing, fanning out

Sahil's research agent now drafts emails and can actually send them. That is useful and a little terrifying. An LLM that can hit "send" with no oversight will, eventually, email the wrong person something wrong. Sahil does not want to remove the power; he wants a checkpoint where a human approves before anything irreversible happens. He also wants the agent to research several sub-topics at once instead of one slow loop. Both needs are about control flow: who decides what runs, when execution pauses, and how work spreads out.

In Module 01 a node returned an update, and a separate conditional edge decided where to go next. Sometimes you want a node to do both at once: change state and say where to go. That is what Command is for. A node returns a Command carrying an update (state change) and a goto (next node). It folds the router into the node itself.

from langgraph.types import Command
from typing import Literal

def triage(state) -> Command[Literal["send", "revise"]]:
    if looks_good(state):
        return Command(update={"status": "ok"}, goto="send")
    return Command(update={"status": "redo"}, goto="revise")
# No add_conditional_edges needed, the node routes itself.
# The Literal[...] hint tells LangGraph the possible destinations.

This is the heart of the module. interrupt() stops the graph mid-node, saves everything (thanks to the checkpointer), and hands control back to your application along with whatever you want to show the human. The run is not blocked in memory waiting; it is persisted and paused. Hours later, you resume by sending the human's response back in, and execution continues from exactly where it stopped.

from langgraph.types import interrupt, Command

def human_approval(state):
    # execution PAUSES here. The value is surfaced to your app.
    decision = interrupt({"draft": state["draft_email"], "ask": "Send this?"})
    # ...resumes HERE with whatever the human sent back...
    if decision == "approve":
        return Command(goto="send_email")
    return Command(goto="revise")

# 1) run until the interrupt, the graph stops cleanly
graph.invoke(inputs, config)        # requires a checkpointer + thread_id
# 2) later, resume by passing a Command(resume=...) on the SAME thread
graph.invoke(Command(resume="approve"), config)

Almost every "human in the loop" requirement is one of these four shapes. They are all interrupt() under the hood; the difference is what you ask for and what you do with the answer.

Sahil's other wish was speed: research five sub-topics at once, not one at a time in a loop. The problem is he does not know how many sub-topics there will be until runtime. Send solves exactly this. From a node you can return a list of Send objects, one per item, each kicking off the same worker node with its own slice of input. The runtime runs them in parallel; a reducer on the result field merges what they each return. That is map-reduce.

from langgraph.types import Send

def assign_workers(state):
    # MAP: one Send per subtopic → N parallel "research_one" runs
    return [Send("research_one", {"topic": t}) for t in state["subtopics"]]

builder.add_conditional_edges("plan", assign_workers, ["research_one"])

# research_one returns {"findings": [one_result]}; the reducer (add) on
# "findings" REDUCES all parallel results into one list. Map, then reduce.

As agents grow, you want to reuse and isolate chunks of logic. A subgraph is a compiled graph used as a node inside a bigger graph. The research loop Sahil built can become a single "research" node in a larger workflow, with its own internal state, tested on its own, and dropped in wherever needed. It is the same modularity instinct that made you split a big function into small ones, applied to graphs.

research_graph = research_builder.compile()      # a complete graph

parent = StateGraph(ParentState)
parent.add_node("research", research_graph)        # ← used as a single node
parent.add_node("write_report", write_report)
parent.add_edge("research", "write_report")

Real agents call flaky things: network APIs, rate-limited models, tools that occasionally time out. Two features keep a long run from dying on a transient hiccup. A retry policy on a node automatically re-runs it on failure, with backoff. And because every step is checkpointed, a crash does not lose the whole task: on restart the graph resumes from the last good checkpoint instead of from zero. That property is called durable execution, and it is a big reason teams pick LangGraph for production.

from langgraph.types import RetryPolicy

builder.add_node(
    "call_api", call_api,
    retry_policy=RetryPolicy(max_attempts=3),   # transient failures → retried
)
# Crash mid-run? Re-invoke on the same thread_id; the checkpointer resumes
# from the last completed step. No checkpointer = no durability.

Give Sahil's agent the email gate. Draft, pause for a human, then either send (just print, do not actually email) or revise. Run it twice: once approving, once rejecting, on two different threads.

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

def draft(state):       return {"draft": "Hi, ..."}

def approve(state):
    # TODO 1: call interrupt({"draft": state["draft"]}) → decision
    # TODO 2: return Command(goto="send") if approved else Command(goto="revise")
    ...

def send(state):     print("SENT:", state["draft"]); return {}
def revise(state):   print("revising"); return {}

b = StateGraph(dict)
for n in (draft, approve, send, revise): b.add_node(n.__name__, n)
b.add_edge(START, "draft"); b.add_edge("draft", "approve")
b.add_edge("send", END); b.add_edge("revise", END)
graph = b.compile(checkpointer=InMemorySaver())   # TODO 3: why is this required?

cfg = {"configurable": {"thread_id": "a"}}
graph.invoke({}, cfg)                       # runs to the interrupt, pauses
graph.invoke(Command(resume="approve"), cfg)  # resumes → SENT

• The first invoke stopped at the interrupt without finishing.
• Resuming with "approve" printed SENT; resuming with anything else went to revise.
• You can explain why removing the checkpointer breaks interrupt().
• You kept all side effects after the interrupt and can say why.
• Bonus: you fanned out with Send and a reducer merged the parallel results.

More in Module 05. = must-know cold.