LangGraph advanced: cycles, HIL, persistence, sub-graphs

Why this matters

The features in this lesson — feedback loops, human-in-the-loop, durable state, and sub-graphs — are what separate a "demo agent" from a system you can run for a real business. Hiring managers specifically test for these in 2026.

Learning objectives

Add self-correction cycles and prevent infinite loops.
Pause a graph for human approval and resume it.
Persist state with InMemorySaver, SqliteSaver, and PostgresSaver.
Compose sub-graphs for reusable agent components.
Fan out parallel work with the Send API.

1. Cycles — agents that loop until quality is met

The simplest cycle: generate → critique → if not good, generate again.

python

class S(TypedDict):
    draft: str
    score: float
    iterations: int

def generate(state: S) -> dict:
    return {"draft": llm.invoke(f"Write a haiku about Mumbai. Iteration {state['iterations']+1}").content,
            "iterations": state["iterations"] + 1}

def critique(state: S) -> dict:
    score = float(critic_llm.invoke(f"Rate 0-1: {state['draft']}").content)
    return {"score": score}

def is_done(state: S) -> str:
    return "end" if state["score"] >= 0.8 or state["iterations"] >= 3 else "loop"

builder = StateGraph(S)
builder.add_node("generate", generate)
builder.add_node("critique", critique)
builder.add_edge(START, "generate")
builder.add_edge("generate", "critique")
builder.add_conditional_edges("critique", is_done, {"end": END, "loop": "generate"})
graph = builder.compile()

Always add an iteration cap. Otherwise a stubborn model will loop until the cost alarm fires.

2. Persistence with checkpointers

A checkpointer saves state after every node. You can resume on a different process, recover from crashes, time-travel, or pause for humans.

python

from langgraph.checkpoint.memory import InMemorySaver
graph = builder.compile(checkpointer=InMemorySaver())

For production, never use InMemorySaver. Use Sqlite for single-process or Postgres for cloud:

python

# Sqlite
# uv add langgraph-checkpoint-sqlite
from langgraph.checkpoint.sqlite import SqliteSaver
import sqlite3
conn = sqlite3.connect("checkpoints.db", check_same_thread=False)
graph = builder.compile(checkpointer=SqliteSaver(conn))

python

# Postgres
# uv add langgraph-checkpoint-postgres psycopg
from langgraph.checkpoint.postgres import PostgresSaver
DSN = "postgresql://user:pass@localhost:5432/postgres"
with PostgresSaver.from_conn_string(DSN) as cp:
    cp.setup()
    graph = builder.compile(checkpointer=cp)

For Redis there is RedisSaver (uv add langgraph-checkpoint-redis).

Thread IDs (one per "conversation")

Every invocation needs a thread_id so the checkpointer knows which state to load:

python

config = {"configurable": {"thread_id": "user-42-conv-1"}}
graph.invoke({"question": "Hi"}, config=config)
graph.invoke({"question": "What did I just ask?"}, config=config)  # remembers

Different thread_id = different conversation.

Time-travel

python

# Get full state of the latest checkpoint for this thread
state = graph.get_state(config)
print(state.values)            # current state
print(state.next)              # next nodes to run
print(state.tasks)             # any pending interrupts/tasks

# Walk history
for snap in graph.get_state_history(config):
    print(snap.config, snap.values)

You can resume from any historical checkpoint by passing its checkpoint_id in configurable. Brilliant for debugging.

3. Human-in-the-loop with `interrupt()`

The 2026 way to pause a graph for human input is the interrupt() primitive from langgraph.types. It works inside a node.

python

from langgraph.types import interrupt, Command
from langgraph.checkpoint.memory import InMemorySaver

class S(TypedDict):
    proposal: str
    approved: bool

def review_node(state: S) -> dict:
    decision = interrupt({
        "type": "approval",
        "message": f"Proposed action: {state['proposal']}\nApprove?",
    })
    return {"approved": bool(decision)}

def execute_node(state: S) -> dict:
    if state["approved"]:
        return {"proposal": state["proposal"] + " [EXECUTED]"}
    return {"proposal": state["proposal"] + " [SKIPPED]"}

g = StateGraph(S)
g.add_node("review",  review_node)
g.add_node("execute", execute_node)
g.add_edge(START, "review")
g.add_edge("review", "execute")
g.add_edge("execute", END)
graph = g.compile(checkpointer=InMemorySaver())

config = {"configurable": {"thread_id": "approve-1"}}

# 1. First call — pauses at interrupt
result = graph.invoke({"proposal": "send refund of $340", "approved": False}, config=config)
# `result` contains __interrupt__ with the question payload

# 2. Show the question to a human... they say YES.

# 3. Resume
final = graph.invoke(Command(resume=True), config=config)
print(final)

Key facts to remember:

You must compile with a checkpointer for interrupt() to work.
The node re-runs from the start when resumed — interrupt() returns the resume value the second time. Keep side-effects after the interrupt.
Resume value is whatever you pass to Command(resume=...) — string, dict, anything JSON-serialisable.

Static interrupts (alternative pattern)

You can also pause before/after specific nodes without changing the node code:

python

graph = builder.compile(checkpointer=cp, interrupt_before=["execute"])

To resume after gathering input, just call graph.invoke(None, config) (or with a state update). Use this for "review every plan before action" workflows.

4. Sub-graphs — reusable agent components

A sub-graph is a StateGraph you compile and use as a node in a parent graph. Same principle as functions — extract reusable pieces.

python

# Define a sub-graph for "do research"
class ResState(TypedDict):
    topic: str
    notes: list[str]

def search(state): ...
def summarise(state): ...

researcher = StateGraph(ResState)
researcher.add_node("search", search)
researcher.add_node("summarise", summarise)
researcher.add_edge(START, "search")
researcher.add_edge("search", "summarise")
researcher.add_edge("summarise", END)
researcher_app = researcher.compile()

# Use it as a node in a bigger graph
class TopState(TypedDict):
    topic: str
    notes: list[str]
    article: str

parent = StateGraph(TopState)
parent.add_node("research", researcher_app)   # subgraph as a node
parent.add_node("write",    write_node)
parent.add_edge(START, "research")
parent.add_edge("research", "write")
parent.add_edge("write", END)

State sharing: keys with the same name flow between parent and sub-graph. Keep namespaces tidy or use subgraph_router patterns.

5. The `Send` API — fan-out to N parallel workers

For "do this for each of 10 items in parallel," use Send:

python

from langgraph.types import Send

def fanout(state) -> list[Send]:
    return [Send("worker", {"item": x}) for x in state["items"]]

builder.add_conditional_edges("dispatcher", fanout, ["worker"])

Each Send invokes the worker node with its own state. Results are merged back via reducers (so worker should write to a list-typed key).

This is the LangGraph version of asyncio.gather. Use it for batch processing inside a graph.

6. Putting it together: a self-correcting writer with HIL approval

python

class S(TypedDict):
    topic: str
    draft: str
    feedback: str
    approved: bool
    iterations: int

def write(state):
    draft = llm.invoke(
        f"Write 200-word post on {state['topic']}." +
        (f"\nIncorporate feedback: {state['feedback']}" if state['feedback'] else "")
    ).content
    return {"draft": draft, "iterations": state["iterations"] + 1}

def critique(state):
    score = float(critic.invoke(f"Score 0-1: {state['draft']}").content)
    if score >= 0.85:
        return {"approved": True, "feedback": ""}
    return {"approved": False, "feedback": f"score {score}, improve clarity"}

def human_review(state):
    decision = interrupt({"draft": state["draft"], "auto_score_pass": state["approved"]})
    return {"approved": bool(decision.get("approved", state["approved"]))}

def is_done(state):
    return "end" if state["approved"] or state["iterations"] >= 3 else "rewrite"

g = StateGraph(S)
g.add_node("write",     write)
g.add_node("critique",  critique)
g.add_node("human",     human_review)
g.add_edge(START, "write")
g.add_edge("write", "critique")
g.add_edge("critique", "human")
g.add_conditional_edges("human", is_done, {"end": END, "rewrite": "write"})

graph = g.compile(checkpointer=SqliteSaver(sqlite3.connect("cp.db", check_same_thread=False)))

You now have:

An auto-critic.
A human gate.
An iteration cap.
Crash-resumable state.

This is the pattern for AI features that touch real users.

Hands-on lab (6 hours)

Build a content publishing assistant:

State carries: topic, outline, draft, seo_score, feedback, approved, iterations.

Graph:

outline — propose an outline.
human_outline_approval — interrupt to let the user edit the outline.
write — produce a draft from the approved outline.
seo_check — runs a heuristic + LLM judge on draft.
Conditional: if seo_score < 0.8 and iterations < 3, send back to write with feedback.
human_final_review — interrupt, allow yes/no/edits.
publish — write the draft to disk (or post to a blog API).

Acceptance:

Use SqliteSaver so the workflow can be killed and resumed.
Streamlit UI shows current node and surfaces interrupts as forms.
README has mermaid diagram + screenshots of the resume flow.

Common pitfalls

Side effects before interrupt() — they run twice. Place after.
No iteration cap on cycles — guaranteed runaway cost.
InMemorySaver in production — state lost on restart.
Different thread_id between turns — agent forgets context.
Sub-graph state collisions — different keys for parent and child of same name.
Using deprecated NodeInterrupt — switch to interrupt().

Self-check

Why is the checkpointer required for interrupt()?
What does Command(resume=value) do?
When would you use Send instead of a for loop inside a node?
How does state flow between a parent graph and a sub-graph?
What is the recommended persistent checkpointer for cloud production?

Why this matters

Learning objectives

1. Cycles — agents that loop until quality is met

2. Persistence with checkpointers

Thread IDs (one per "conversation")

Time-travel

3. Human-in-the-loop with interrupt()

Static interrupts (alternative pattern)

4. Sub-graphs — reusable agent components

5. The Send API — fan-out to N parallel workers

6. Putting it together: a self-correcting writer with HIL approval

Hands-on lab (6 hours)

Common pitfalls

Self-check

References

3. Human-in-the-loop with `interrupt()`

5. The `Send` API — fan-out to N parallel workers