5. Human-in-the-Loop

Some pipelines can’t run end-to-end without a human in the loop. A validation step finds a problem and needs review. An output is generated but must be approved before side effects. A privileged operation requires confirmation.

NeoGraph makes this a one-line addition: pass interrupt_when= to @node. When the condition returns a truthy value, the graph pauses via LangGraph’s interrupt() mechanism, checkpoints its state, and waits. Resume with run(graph, resume={...}, config=config).

The graph

The Operator modifier adds an interrupt checkpoint after validate. If the condition returns truthy, the graph pauses. Resume with run(graph, resume={...}) to continue.

The scenario

A pipeline analyzes a set of requirements and checks whether coverage meets a quality bar. If it doesn’t, a human must review the gaps before the final report is generated.

from __future__ import annotations

import sys
from langgraph.checkpoint.memory import MemorySaver
from pydantic import BaseModel

from neograph import compile, construct_from_module, node, run


# ── Schemas ──

class Analysis(BaseModel, frozen=True):
    claims: list[str]
    coverage_pct: int

class ValidationResult(BaseModel, frozen=True):
    passed: bool
    issues: list[str]

class FinalReport(BaseModel, frozen=True):
    text: str


# ── Pipeline ──

@node(outputs=Analysis)
def analyze() -> Analysis:
    return Analysis(claims=["auth", "logging", "encryption"], coverage_pct=55)


@node(
    outputs=ValidationResult,
    interrupt_when=lambda state: (
        {"issues": state.check.issues, "message": "Please review and approve"}
        if state.check and not state.check.passed
        else None
    ),
)
def check(analyze: Analysis) -> ValidationResult:
    if analyze.coverage_pct < 80:
        return ValidationResult(
            passed=False,
            issues=[f"Coverage {analyze.coverage_pct}% is below 80% threshold"],
        )
    return ValidationResult(passed=True, issues=[])


@node(outputs=FinalReport)
def report(analyze: Analysis) -> FinalReport:
    return FinalReport(
        text=f"Report: {analyze.claims}, coverage: {analyze.coverage_pct}%"
    )


pipeline = construct_from_module(sys.modules[__name__], name="review-pipeline")

The `interrupt_when` kwarg

interrupt_when accepts either a callable or a registered condition name:

Callable (inline): interrupt_when=lambda state: payload_or_none. The function receives the full pipeline state and returns either None (continue) or a dict (pause, with the dict as the interrupt payload).
String (registered): interrupt_when='condition_name'. The name must be supplied to compile() via the conditions= kwarg: compile(pipeline, conditions={'condition_name': lambda state: ...}).

The inline form is usually cleaner. When the condition returns a dict, the graph pauses and LangGraph’s interrupt() fires with that payload as the reason.

Running with a checkpointer

Interrupt/resume requires a checkpointer — LangGraph needs somewhere to persist state between the pause and the resume. The compiler enforces this: calling compile(pipeline) on a pipeline with interrupt_when and no checkpointer raises an error.

from langgraph.checkpoint.memory import MemorySaver

graph = compile(pipeline, checkpointer=MemorySaver())

# thread_id identifies this execution for later resume
config = {"configurable": {"thread_id": "review-001"}}

MemorySaver is fine for development. For production, use SqliteSaver, PostgresSaver, or any LangGraph-compatible checkpointer.

First run — pauses

result = run(graph, input={"node_id": "REQ-001"}, config=config)

if "__interrupt__" in result:
    interrupt_data = result["__interrupt__"]
    for interrupt in interrupt_data:
        print(f"Paused: {interrupt.value}")
        # Paused: {'issues': ['Coverage 55% is below 80% threshold'],
        #          'message': 'Please review and approve'}

When the graph pauses, run() returns with the partial state plus an __interrupt__ key containing the interrupt payloads. Everything up to the pause point is already in the checkpoint — analyze ran, check ran and produced a failing ValidationResult, then the interrupt fired.

Resume with human feedback

result = run(graph, resume={"approved": True, "reviewer": "alice"}, config=config)

print(result["human_feedback"])   # {'approved': True, 'reviewer': 'alice'}
print(result["report"].text)      # The report ran after the resume

Calling run() with resume= instead of input= continues the paused graph. The resume dict is stored in state.human_feedback so downstream nodes can read the decision. The graph then continues from check onward — report runs, and you get the final state.

When the condition is falsy

If analyze.coverage_pct had been 85% instead of 55%, check would return passed=True, the lambda would return None, and the interrupt would never fire. The graph runs straight through to report as if interrupt_when weren’t there.

Why not Operator directly?

The Operator modifier still exists for runtime construction. For pipelines written with @node, interrupt_when= is the cleaner path — the condition lambda is co-located with the node it guards, and the graph wiring happens automatically at construct_from_module time.

Gating tools before side effects: `gate_tools_when=`

interrupt_when= reviews a node’s finished output — too late to stop a tool that already ran. For an agent/act node whose tools have side effects (write a file, charge a card, send a message), you often want the opposite: pause and get approval before the tools run at all. That is gate_tools_when=.

Pass it a predicate on @node. On every ReAct turn where the agent wants to call tools, the predicate runs before the tool step. Return a truthy payload to pause — the graph checkpoints and surfaces the payload as an interrupt, exactly like interrupt_when=. Return a falsy value to let the tools run untouched.

from neograph import node, Tool


def approve_tools(state) -> dict | None:
    # Any truthy return pauses BEFORE the tools run; the value is the
    # interrupt payload shown to the human. Return None to run without pausing.
    return {"pending": "tool calls", "message": "approve before running?"}


@node(
    mode="agent",
    outputs=Report,
    model="reason",
    prompt="research/summarize",
    tools=[Tool(name="write_file", budget=3)],
    gate_tools_when=approve_tools,
)
def research() -> Report: ...

The run pauses before write_file ever executes, so nothing has happened yet when the human sees the request:

result = run(graph, input={"node_id": "REQ-1"}, config=config)

if "__interrupt__" in result:
    # result["__interrupt__"][0].value == {"pending": "tool calls",
    #                                       "message": "approve before running?"}
    result = run(graph, resume={"approved": True}, config=config)
    # NOW the tools run — exactly once.

Because the gate sits on the tools boundary of the agent subgraph, it works the same across run/arun and survives a checkpointed resume. This is the structural version of “approve before side effects”: you never modify the tool, and the tool cannot fire before approval.

gate_tools_when= only makes sense on an agent/act node (the only nodes with a tool step) — setting it on any other node raises ConstructError at assembly. It gates the whole tool step per turn; to pause inside one specific tool’s logic, use ask_human() below.

Mid-loop pauses: inside an `agent`/`act` tool loop

interrupt_when= pauses at node boundaries and gate_tools_when= pauses before the tool step. But sometimes a single tool itself needs to stop and ask a human partway through its own logic. ask_human() is the blessed path for that: call it from inside a tool body.

from pydantic import BaseModel
from neograph import ask_human


class ReviewRequest(BaseModel):
    question: str
    candidate: str

class Decision(BaseModel):
    approved: bool
    note: str


class DecideTool:
    name = "decide"

    def invoke(self, args: dict) -> str:
        # Pause the tool loop and ask a human. The graph checkpoints here.
        decision = ask_human(
            ReviewRequest(question="Approve this candidate?", candidate=args["name"]),
            resume_model=Decision,
        )
        # After resume, `decision` is a validated Decision instance.
        return f"approved={decision.approved}: {decision.note}"

    async def ainvoke(self, *a, **k) -> str:
        return self.invoke(*a, **k)

The payload surfaces exactly like any other interrupt — result["__interrupt__"][0].value holds payload.model_dump(). Resume the graph the same way:

result = run(graph, resume={"approved": True, "note": "looks good"}, config=config)

When you pass resume_model=, the resumed dict is validated into that model before your tool sees it — a malformed answer raises pydantic.ValidationError at the ask_human() call, not deep inside your tool. Omit resume_model= to receive the raw resume value unchanged.

ask_human() is pure, optional sugar: it adds the typed contract and makes the pause a marker the linter can see. It contains zero execution logic — the pause/resume path is byte-identical to calling LangGraph’s interrupt() directly, which remains fully supported. Use whichever you prefer:

from langgraph.types import interrupt

# Raw path — no typed contract, but works identically.
answer = interrupt({"question": "Approve this candidate?"})

Each ReAct turn compiles to its own checkpointed superstep ({node}__agent → {node}__tools → back), so a tool that ran in an earlier turn is already committed when a later turn pauses — it does not re-fire on resume. This is idempotency by construction across turns: a pre-interrupt side-effecting tool runs exactly once across an interrupt+resume (pinned by the test_pre_interrupt_tool_runs_exactly_once_across_resume keystone). The old monolithic agent loop re-ran the whole node on resume and would double-fire; that footgun is gone.

The one case that still needs care is within a single turn: if the model requests several tool calls in the same turn and one of them calls ask_human(), side effects from the earlier calls in that same turn can replay on resume. Keep same-turn mutations before an ask_human() idempotent — or reach for gate_tools_when= above, which pauses before any tool in the turn runs, so nothing has fired when the human is asked. The linter still warns (ask_human_in_mutating_node) when ask_human() is reachable from an act-mode node — a nudge to check same-turn ordering.

Key takeaways

interrupt_when= pauses the graph when the condition returns a truthy payload
gate_tools_when= pauses an agent/act node before its tools run — approve side effects before they happen, without touching tool code
ask_human() pauses inside a specific tool body, with an optional typed resume contract
Idempotency across ReAct turns is by construction — a pre-interrupt tool runs exactly once across resume
Inline lambdas are cleaner than registered conditions for most cases
A checkpointer is required — MemorySaver for dev, persistent savers for prod
run(resume={...}, config=config) continues from the checkpoint
The resume payload lands in state.human_feedback