LangGraph — Cycles, Human-in-the-Loop & Persistence
The three features that make a graph a production agent: cycles (loop until done, safely), persistence (durable state you can pause, resume, and inspect), and human-in-the-loop (stop for approval before a risky action, then continue). This is where the safety gate you've been promised since Chapter 4 finally becomes real.
Learning objectives
- Build cycles safely and bound them with a recursion limit and state counters.
- Add a checkpointer so graph state is durable and threads are resumable.
- Pause a graph with an interrupt, get human approval, and resume.
- Inspect and edit graph state between steps (time-travel debugging).
- Assemble it all into a real approval-gated agent — the capstone's core.
invoke can't give you — and they're exactly what separate a demo agent from one you'd let touch production.Cycles, bounded advanced
You already made a cycle in L4 (tools → model). The power of a loop is also its danger: a confused agent can loop forever, burning tokens. LangGraph bounds cycles two ways — a global recursion limit and your own state counters.
recursion_limit as a backstop. Two independent brakes so a loop can never run away.
A cycle is just an arrow in the graph that loops back to an earlier node so the agent can try again. This picture shows how to make that loop safe — how to guarantee it can't run forever.
- Follow the solid arrow left-to-right: the work node does something, then hands off to a check node that asks two questions —
done?andn < max?(have we tried too many times?). - The dashed arrow curving back to
work nodeis the cycle: it fires only "while not done AND n < max". That condition is the loop's brake. - If either brake trips — the task is done, or the attempt counter hit its cap — the solid arrow goes to END instead and the graph stops.
- The caption's key phrase is two independent brakes: your own counter in state, plus LangGraph's built-in
recursion_limitas a last-resort backstop.
In short: A loop needs a guaranteed way out. Here that way out is the check node routing to END the moment the task finishes or the tries run out.
Setup to run this snippet
END = "__end__" # sentinel used by graph examples
class _app_t:
invoke = 'demo'
def invoke(self, *a, **k): return 'demo'
def __getattr__(self, k): return 'demo'
app = _app_t()
class _Any:
'''stands in for any undefined demo value; supports call/attr/index/
iteration and basic arithmetic (as 0.7) so demo snippets run.'''
def __call__(self, *a, **k): return _Any()
def __getattr__(self, k): return _Any()
def __getitem__(self, k): return _Any()
def __iter__(self): return iter([])
def __len__(self): return 0
def __contains__(self, o): return True
def __enter__(self, *a): return _Any()
def __exit__(self, *a): return False
def __float__(self): return 0.7
def __int__(self): return 1
def __lt__(self, o): return True
def __gt__(self, o): return False
def __le__(self, o): return True
def __ge__(self, o): return False
def __add__(self, o): return o
def __radd__(self, o): return o
def __bool__(self): return True
def __repr__(self): return 'demo'
def __str__(self): return 'demo'
inputs = _Any()bounded_cycle.pydef route(state) -> str:
if state["done"]: return END
if state["attempts"] >= 5: return END # hard stop in your own logic
return "work"
# backstop: the framework raises if the graph runs too long
app.invoke(inputs, config={"recursion_limit": 25})
This is the routing function that decides, after each pass, whether the graph loops back to do more work or exits. It's the software brake the diagram above described. LangGraph calls it and uses the string it returns to pick the next node.
route(state)receives the current state (a dictionary the graph carries between steps) and returns a string naming where to go next.if state["done"]: return END— if the work is finished, return the specialENDvalue and the graph stops. This is the normal, happy exit.if state["attempts"] >= 5: return END— even if not done, stop after 5 tries. This is your own counter guarding against an agent that never converges.return "work"— otherwise, loop back to theworknode and try again. That returned name is what creates the cycle.- The last line,
app.invoke(inputs, config={"recursion_limit": 25}), is the backstop: even if your counter had a bug, LangGraph raises an error after 25 steps so the loop can't burn tokens forever.
What the output means: Nothing prints here — this is structure. In a real run, the graph would bounce between work and route until done is true or attempts reaches 5, then head to END.
Try this: Lower recursion_limit to 2 and give the agent a task that needs several passes — you'll see LangGraph raise a recursion error, proving the backstop works even when your own counter would have let it keep going.
Lab L5.1 · Persistence with a checkpointer advanced
By default a graph's state lives only for one invoke. Attach a checkpointer and LangGraph saves the state after every step, keyed by a thread id. Now the same conversation can span many calls, survive a restart, and be inspected step by step.
Requires: pip install langgraph
persist.pyfrom langgraph.checkpoint.memory import MemorySaver
# production: from langgraph.checkpoint.sqlite import SqliteSaver, etc.
app = g.compile(checkpointer=MemorySaver())
cfg = {"configurable": {"thread_id": "user-1"}} # names the conversation
app.invoke({"messages": [("user", "My name is Alice.")]}, cfg)
app.invoke({"messages": [("user", "What's my name?")]}, cfg) # remembers — same thread
snapshot = app.get_state(cfg) # the full state right now
print(snapshot.values["messages"])
By default a graph forgets everything the moment invoke returns. A checkpointer changes that: LangGraph saves the whole state after every step, so the same conversation can span many calls and even survive a restart. This is real, durable memory.
from langgraph.checkpoint.memory import MemorySaverimports the simplest checkpointer — it keeps state in memory. The comment notes that production uses a durable store like SQLite or Postgres instead.app = g.compile(checkpointer=MemorySaver())attaches the checkpointer when you build the graph. That one argument is what turns saving on.cfg = {"configurable": {"thread_id": "user-1"}}names the conversation. A thread id is like a folder label — every call using the same id shares one saved history, so different users get different threads.- The two
app.invoke(...)calls happen in separate requests, yet the second ("What's my name?") can answer "Alice" because both usedthread_id "user-1"— the checkpointer carried the first turn forward. snapshot = app.get_state(cfg)reads back the full saved state for that thread;snapshot.values["messages"]is the stored conversation so far.
What the output means: The final print shows the message list for thread user-1 — including Alice's name — proving the state persisted between the two independent invoke calls.
Try this: Run a third invoke with a different thread_id (say "user-2") and ask "What's my name?". It won't know — a fresh thread starts with an empty history. That's how one graph safely serves many users.
| Checkpointing gives you… | Why it matters |
|---|---|
| Durable memory | Conversation state survives across calls and restarts — real multi-session memory (vs L2's in-memory history) |
| Threads | Each thread_id is an isolated conversation — one graph serves many users |
| Resumability | A crashed or paused run continues from the last checkpoint, not from scratch |
| Inspection | get_state / get_state_history expose exactly what the agent believes at each step |
Lab L5.2 · Human-in-the-loop with interrupts advanced
This is the chapter's payoff. An interrupt pauses the graph mid-run, hands control back to your application, and waits. You show the pending action to a human; on approval you resume and the graph continues exactly where it stopped.
This is the human-in-the-loop gate: how the graph pauses before doing something irreversible, waits for a person to approve, and then continues. Read it as a left-to-right flow with one detour downward.
- The model node runs first and decides it wants to use a risky tool. The solid arrow carries the flow into the interrupt node ("before risky tool").
- At
interruptthe graph stops. The dashed arrow drops down to human decides — control has left the graph and gone back to your application, which shows the pending action to a person. - Nothing has happened yet — the risky action has not run. The graph's state is safely saved (checkpointed) while it waits, possibly for minutes or days.
- On approval, the resume arrow re-enters the graph at tool runs — it continues exactly where it paused, runs the tool, and flows to END.
- Notice the ordering in the caption: pause → approve → resume. The action can only happen after a human signs off.
In short: An interrupt is a checkpointed pause: the agent proposes, a human disposes, and only then does the irreversible step run. No approval, no action.
Requires: pip install langgraph
hitl.pyfrom langgraph.types import interrupt, Command
def risky_action(state):
# pause and surface the proposed action to a human
decision = interrupt({"action": "delete", "target": state["target"]})
if decision != "approve":
return {"result": "cancelled by human"}
return {"result": _do_delete(state["target"])}
# --- running it ---
cfg = {"configurable": {"thread_id": "job-42"}}
app.invoke(inputs, cfg) # runs until the interrupt, then STOPS
state = app.get_state(cfg) # inspect what it wants to do
print(state.next, state.values) # show the human the pending action
# human approves → resume with their decision
app.invoke(Command(resume="approve"), cfg)
This is the interrupt in code — the real safety gate the whole course has been building toward. A node calls interrupt(...), the graph halts, your app asks a human, and you resume with their answer. The risky work only runs on approval.
from langgraph.types import interrupt, Commandpulls in the two pieces:interruptto pause, andCommandto feed a decision back in when you resume.- Inside
risky_action(state), the calldecision = interrupt({...})does two things: it pauses the whole graph, and it surfaces a small package (the action and its target) so your app can show a human what's about to happen. - When you later resume, that human's answer becomes the value of
decision.if decision != "approve": return {"result": "cancelled by human"}means a rejection stops safely and the delete never runs. - Only on approval does
_do_delete(state["target"])actually execute — the irreversible step. - Below the function is the driver: the first
app.invoke(inputs, cfg)runs until the interrupt and then STOPS.app.get_state(cfg)lets you readstate.next(the paused node) andstate.valuesto show the human. Finallyapp.invoke(Command(resume="approve"), cfg)feeds "approve" back in and the graph finishes the job.
What the output means: First invoke: the run halts at the interrupt, nothing deleted. The print shows the pending action. After Command(resume="approve"), the delete runs and the graph completes. Resume with anything other than "approve" and it returns "cancelled by human".
Try this: Change the resume value to Command(resume="reject") and confirm _do_delete never runs. That one line is the difference between an agent that asks permission and one that acts on its own.
interrupt() function (shown) pauses inside a node and can return the human's input into the flow — the flexible, recommended approach. There's also a static interrupt_before=["node"] option at compile time that pauses before a named node. Both rely on the checkpointer to hold state while paused.Lab L5.3 · Inspecting & editing state expert
Because every step is checkpointed, you can walk the history, and even edit state before resuming — invaluable for debugging and for correcting an agent that went slightly wrong without restarting it.
Illustrative fragment — defines demo values / files are needed before this runs standalone.
timetravel.py# see every checkpoint (newest first)
for snap in app.get_state_history(cfg):
print(snap.next, len(snap.values["messages"]))
# correct the state, then resume from the fixed version
app.update_state(cfg, {"target": "the-right-record"})
app.invoke(Command(resume="approve"), cfg)
Because every step is saved, you can not only read the state but walk its whole history and even edit it before resuming. This is "time-travel debugging": rewind, fix a mistake, and replay — without re-running the agent from scratch.
for snap in app.get_state_history(cfg):loops over every saved checkpoint for the thread, newest first. Eachsnapis a frozen snapshot of the graph at one step.print(snap.next, len(snap.values["messages"]))shows, for each snapshot, which node was about to run and how many messages the state held then — a compact audit trail of how the run progressed.app.update_state(cfg, {"target": "the-right-record"})edits the stored state. Here it fixes a wrongtargetbefore continuing — correcting the agent instead of restarting it.app.invoke(Command(resume="approve"), cfg)then resumes from that corrected state, so the agent proceeds with the fixed value.
What the output means: The loop prints one line per checkpoint (the pending node and message count for each), letting you see exactly how the state evolved. After the edit and resume, the agent acts on "the-right-record" rather than the original mistaken target.
Try this: Add print(snap.values.get("target")) inside the loop to watch the target value at each checkpoint — you'll see the wrong value in history and the corrected one after update_state.
Putting it together: an approval-gated agent expert
Cycles + persistence + interrupts compose into the shape of a real production agent — and of the capstone.
| Capability | What it buys the agent |
|---|---|
| Cycle (bounded) | Keeps working until the problem is resolved — without looping forever |
| Checkpointer | Survives restarts; each incident is an inspectable, resumable thread |
| Interrupt | Never takes an irreversible action without human sign-off |
| State inspection | Full audit trail of what it believed and did at each step |
Common pitfalls expert
| Pitfall | Fix |
|---|---|
| Cycle with no exit | Guard in the router + set recursion_limit |
| Interrupt without a checkpointer | HITL requires persistence — compile with a checkpointer |
Reusing one thread_id for different users | One thread per conversation; mixing them cross-contaminates state |
MemorySaver in production | It's lost on restart — use a durable checkpointer (SQLite/Postgres) |
| Resuming without showing the human the action | Inspect get_state and surface the pending step before approving |
| Gating nothing / gating everything | Gate irreversible actions; let read-only steps run — match the gate to risk |
Exercises expert
Exercise L5.1 — Resumable chat
Context: Multi-user memory is a checkpointer plus a thread id — about ten lines.
Your task: Compile a simple chat graph with MemorySaver, run two turns on one thread_id, then a turn on a different one, and confirm the second thread doesn't know the first's context.
Requirements:
- Compile the chat graph with a checkpointer
- Run two turns on one thread id
- Run a turn on a second thread id
- Confirm the two threads have independent memory
💡 Hint: The thread id is the whole isolation mechanism — same graph, different thread, separate state.
Exercise L5.2 — Approve-before-delete
Context: An interrupt-gated delete is the concrete test that a human sits between intent and a destructive action.
Your task: Build a graph with a delete node behind an interrupt, run it, inspect the pending action, then reject it once and approve it once.
Requirements:
- A delete node sits behind an interrupt
- After the first invoke, the pending action is inspectable and the delete has NOT happened
- Rejecting cancels without ever running the delete
- Approving (resume) runs the delete exactly once
💡 Hint: Check get_state().next points at the interrupted node before approval; only a resume with an approve decision runs the delete.
Show what to verify
After the first invoke, get_state().next should point at the interrupted node and no delete has happened. Only after Command(resume="approve") does _do_delete run. Rejection returns "cancelled" and never calls it.
Exercise L5.3 — Bounded reflection loop
Context: The capstone combines every cycle idea: a bounded, checkpointed, self-correcting loop you can inspect step by step.
Your task: Combine everything into a generate→reflect cycle that loops until a quality check passes or 3 attempts, checkpointed so you can inspect each revision in get_state_history.
Requirements:
- A generate→reflect cycle (L1's reflection) as a graph
- Exit when the quality check passes
- Cap at 3 attempts as the backstop
- Checkpoint each step so revisions are inspectable via history
- Confirm you can walk the revisions after the run
💡 Hint: It's the bounded cycle (rung 1) plus a checkpointer (rung 3) plus reflection — the state history is what lets you audit each revision.
🪜 Practice ladder beginner → industry
Six graded exercises, easy to real-world. Try each before opening its solution.
Context: A loop needs a guaranteed exit. The robust pattern is two independent brakes: a done-check and an attempt counter.
Your task: Model a cycle that runs a work node until 'done' OR an attempt counter hits its cap.
Requirements:
- Run a work step that makes progress and increments an attempt counter
- Exit when the work is done
- Also exit when attempts reach a cap
- Report which brake ended the loop
💡 Hint: Two independent conditions on the while guard; the counter guarantees termination even if 'done' never becomes true.
Show solution
Two brakes: a done-check and a counter. Runnable stdlib:
def work(state):
state["progress"] += 40
state["attempts"] += 1
return state
def run(max_attempts=5):
state = {"progress": 0, "attempts": 0}
while state["progress"] < 100 and state["attempts"] < max_attempts:
state = work(state)
reason = "done" if state["progress"] >= 100 else "hit attempt cap"
return state, reason
print(run()) # ({'progress': 120, 'attempts': 3}, 'done')
The counter guarantees the loop can't run forever even if "done" never becomes true.
Context: LangGraph enforces a global recursion_limit as a last resort — a second, framework-level brake for the bug where your own exit condition never trips.
Your task: Model a driver that raises if total steps exceed a limit, independent of your own counter.
Requirements:
- Count total steps taken by the driver
- Raise a dedicated error when steps exceed the recursion limit
- The limit is independent of the node's own done-check
- Show it firing on a node that never sets 'done'
💡 Hint: Your counter is the intended exit; the recursion limit is the backstop — demonstrate it with a deliberately buggy step function.
Show solution
A second, framework-level brake. Runnable:
class RecursionLimit(Exception):
pass
def run_graph(step_fn, state, recursion_limit=25):
steps = 0
while not state.get("done"):
steps += 1
if steps > recursion_limit:
raise RecursionLimit(f"exceeded {recursion_limit} steps")
state = step_fn(state)
return state, steps
# a buggy node that never sets done -> the backstop fires
buggy = lambda s: {**s, "n": s.get("n", 0) + 1}
try:
run_graph(buggy, {}, recursion_limit=10)
except RecursionLimit as e:
print("backstop:", e)
Your counter is the intended exit; recursion_limit is the backstop for the bug where your exit condition never trips.
Context: Persistence saves graph state per thread so a run can pause, resume, and be inspected — exactly what the real MemorySaver/SqliteSaver do.
Your task: Model an in-memory checkpointer keyed by thread_id that a run reads from and writes back to.
Requirements:
- A store maps
thread_idto that thread's state - A step loads the thread's state, updates it, and persists it back
- State survives across separate step calls on the same thread
- A different thread has independent state
- Demonstrate two threads not seeing each other's state
💡 Hint: Load-update-save keyed by thread id; durable per-thread state is what makes a graph pausable and resumable.
Show solution
A checkpointer is a thread-keyed state store. Runnable:
class MemorySaver:
def __init__(self):
self.store = {}
def get(self, thread_id):
return self.store.get(thread_id, {"messages": [], "step": 0})
def put(self, thread_id, state):
self.store[thread_id] = state
saver = MemorySaver()
def step(thread_id, user_msg):
state = saver.get(thread_id) # resume where we left off
state["messages"].append(user_msg)
state["step"] += 1
saver.put(thread_id, state) # persist
return state
step("chat-1", "hello")
s = step("chat-1", "again")
print(s["step"], "steps on thread chat-1") # 2 -- state survived
print(saver.get("chat-2")["step"]) # 0 -- separate thread
The real MemorySaver/SqliteSaver do exactly this — durable per-thread state is what makes a graph pause/resume/inspectable.
Context: Human-in-the-loop pauses the graph before a risky action, waits for approval, then continues — the safety gate promised since Chapter 4, built on interrupts.
Your task: Model an interrupt: the run stops with a pending action, a human decides, and the run resumes.
Requirements:
- Progress until a dangerous action is pending, then pause with status 'interrupted'
- Surface the pending action while paused
- A resume step takes the human's decision (approve/deny)
- Approve executes the action; deny skips it
- Show both the approve and the deny outcome from the same paused state
💡 Hint: The real graph pauses at interrupt_before a node and resumes by invoking again on the same thread — model the stop, surface, decide, resume cycle.
Show solution
Interrupt = stop, surface the pending action, resume on a decision. Runnable:
def run_until_interrupt(state):
# progresses until it wants to run a dangerous action
state["plan"] = "delete_records"
state["status"] = "interrupted" # pause here for approval
return state
def resume(state, approved):
if state["status"] != "interrupted":
return state
if approved:
state["result"] = f"executed {state['plan']}"
else:
state["result"] = f"skipped {state['plan']} (denied)"
state["status"] = "done"
return state
s = run_until_interrupt({})
print("paused with pending:", s["plan"])
print(resume(dict(s), approved=False)["result"]) # skipped ...
print(resume(dict(s), approved=True)["result"]) # executed ...
The graph pauses at interrupt_before a node, a human approves, and invoke with the same thread resumes — the safety gate promised since Chapter 4.
Context: Because every step is checkpointed, you can rewind to any point, edit the state, and resume — time-travel debugging without rerunning from scratch.
Your task: Model a state history you can rewind to and patch, then continue from.
Requirements:
- Snapshot the state after each step into a history
- Rewind to restore an earlier checkpoint
- Edit the restored state before continuing
- Resume stepping from the edited state
- Show that the resumed run reflects the edit
💡 Hint: Keep a list of per-step snapshots; rewinding returns a copy of an earlier one that you patch and feed back into the step function.
Show solution
Checkpoint history enables rewind + edit. Runnable:
class Graph:
def __init__(self):
self.history = []
def step(self, state):
state = {**state, "n": state.get("n", 0) + 1}
self.history.append(dict(state)) # snapshot each step
return state
def rewind_to(self, i):
return dict(self.history[i]) # restore an earlier checkpoint
g = Graph()
s = {}
for _ in range(3):
s = g.step(s)
print("now:", s) # {'n': 3}
patched = g.rewind_to(0) # go back to after step 1
patched["n"] = 99 # edit state
s = g.step(patched) # resume from the edit
print("after time-travel:", s) # {'n': 100}
Because every step is checkpointed, you can rewind to any point, edit the state, and resume — debugging a run without rerunning from scratch.
Context: The capstone's core is a production HITL graph: a checkpointer, interrupt_before a tools node for approval, then resume — the durable version of every gate so far.
Your task: Write the real approval-gated graph using documented langgraph APIs. (Needs the libraries installed.)
Requirements:
- A typed State with an add-messages reducer, plus a model node and a tools node
- Compile with a checkpointer (e.g.
MemorySaver) andinterrupt_before=["tools"] - Invoke with a
thread_idconfig so state is durable - Resume by invoking with
Noneon the same thread after approval - Label it as needing
langgraph/langchain-anthropic+ a key
💡 Hint: The checkpointer makes state durable, interrupt_before pauses at the risky node, and resuming with None on the same thread continues — verify imports against your version.
Show solution
Correct HITL graph. Needs pip install langgraph langchain-anthropic + ANTHROPIC_API_KEY:
from typing import Annotated, TypedDict
from langgraph.graph import StateGraph, START, END
from langgraph.graph.message import add_messages
from langgraph.checkpoint.memory import MemorySaver
class State(TypedDict):
messages: Annotated[list, add_messages]
def model(state): return {"messages": []} # calls Claude + tools in reality
def tools(state): return {"messages": []}
builder = StateGraph(State)
builder.add_node("model", model)
builder.add_node("tools", tools)
builder.add_edge(START, "model")
builder.add_edge("model", "tools")
builder.add_edge("tools", END)
# checkpointer + interrupt = pause before running tools for human approval
graph = builder.compile(checkpointer=MemorySaver(), interrupt_before=["tools"])
cfg = {"configurable": {"thread_id": "run-1"}}
graph.invoke({"messages": [{"role": "user", "content": "delete old records"}]}, cfg)
# ...inspect graph.get_state(cfg), get human approval, then:
graph.invoke(None, cfg) # resume from the interrupt on the same thread
The checkpointer makes state durable; interrupt_before pauses at the risky node; resuming with None on the same thread continues. This is the capstone's core. Verify imports against your version.
✓ Checkpoint — you can move on when you can…
- Build a cycle and bound it with a counter and
recursion_limit. - Add a checkpointer and run a resumable, threaded conversation.
- Pause with
interrupt, inspect the pending action, and resume. - Walk
get_state_historyand edit state before resuming. - Describe how cycles + persistence + interrupts compose into a production agent.
terraform apply or kubectl delete runs without human approval. Chapter 8c builds exactly this gate. Next module takes these single-graph skills into multi-agent systems — CrewAI, AutoGen, and the supervisor/network topologies from L1. See the safety-gate build →Knowledge check check yourself
Why must a checkpointer be attached before you can use an interrupt for human-in-the-loop, and what does the checkpointer buy you?
Show answer
thread_id), resumability from the last checkpoint, and step-by-step inspection, which is exactly what a pause-and-resume gate requires.What are the two independent brakes that keep a LangGraph cycle from running forever?
Show answer
attempts >= 5 or the task is done), and LangGraph's built-in recursion_limit passed to invoke as a framework backstop that raises after N steps even if your counter has a bug.