Human-in-the-Loop¶

FlowgentraAI supports interrupting graph execution before or after specific nodes, allowing human review and modification before continuing.

Setup¶

Enable checkpointing and set interrupt points:

from flowgentra_ai import StateGraphBuilder, SharedState, END

def draft(state):
    state["draft"] = f"Article about {state['topic']}"
    return state

def publish(state):
    state["published"] = True
    return state

builder = StateGraphBuilder()
builder.add_node("draft", draft)
builder.add_node("publish", publish)
builder.set_entry_point("draft")
builder.add_edge("draft", "publish")
builder.add_edge("publish", END)

# Interrupt BEFORE the publish node runs
builder.interrupt_before("publish")

# Enable file-based checkpointing (required for resume)
builder.set_checkpointer("./checkpoints")

graph = builder.compile()

Interrupt and Resume¶

# First invocation -- runs "draft", then pauses before "publish"
result = graph.invoke_with_thread("thread-1", SharedState({"topic": "AI"}))
print(result["draft"])  # "Article about AI"

# ... human reviews the draft ...

# Resume execution (continues from where it stopped)
result = graph.resume("thread-1")
print(result["published"])  # True

Resume with State Modifications¶

Inject human edits before resuming:

# First run
result = graph.invoke_with_thread("thread-1", SharedState({"topic": "AI"}))

# Human modifies the draft
updates = SharedState({"draft": "Improved article about AI trends"})
result = graph.resume_with_state("thread-1", updates)

Interrupt After¶

You can also interrupt after a node runs:

builder.interrupt_after("draft")

This lets the node execute, then pauses -- useful for review of the output.

Multiple Interrupt Points¶

You can set multiple interrupt points in a single graph:

builder.interrupt_before("review")
builder.interrupt_before("publish")

Each resume() call advances to the next interrupt point.