Human in the loop (LangGraph + DRAgent)

This page describes how interrupt / resume works when you use LangGraph inside datarobot_genai and the e2e DRAgent sample.

Why you need a checkpointer

LangGraph only remembers a paused run if the compiled graph was built with a Checkpointer. The LangGraphAgent constructor accepts checkpointer=... and passes it to StateGraph.compile(...).

Passing a checkpointer

LangGraphAgent forwards whatever you pass as checkpointer= to StateGraph.compile(...) (see langgraph/agent.py). If you omit it, the graph compiles without persistence and interrupt() / resume will not restore prior state.

flowchart TD
    A["LangGraphAgent(...)"] --> B{"checkpointer set?"}
    B -->|yes| C["StateGraph.compile(checkpointer=...)"]
    B -->|no| D["No persistence"]

Option	Constructor	Resume across requests?	Typical use
None (default)	omit `checkpointer`	No	Graphs without `interrupt()` / HITL
Any saver	`checkpointer=...`	Depends on saver	E2E (`InMemorySaver`), DR FS, Postgres, etc.

1. No checkpointer (default):

agent = MyLangGraphAgent(llm=llm, tools=tools)

2. In-memory (e2e / single process) — use one shared instance per process (see myagent.py):

from langgraph.checkpoint.memory import InMemorySaver

HITL_CHECKPOINTER = InMemorySaver()

agent = MyLangGraphAgent(llm=llm, tools=tools, checkpointer=HITL_CHECKPOINTER)

What clients see in the event stream

When the graph reports an __interrupt__ update, streaming emits a CUSTOM event named on_interrupt, then a RUN_FINISHED with result["langgraph"]["interrupted"] so UIs can show approval UI before the next call.

Compile-time breakpoints (optional)

You can also pass interrupt_before / interrupt_after when constructing the agent; they are forwarded to StateGraph.compile(...), for example to pause before a node without custom interrupt() code in that node.

DRAgent / NAT: passing the checkpointer (example from e2e-tests)

The minimal register.py builds MyAgent(..., checkpointer=HITL_E2E_CHECKPOINTER).

A new InMemorySaver() on every request would drop checkpoint state, so the interrupt response and the resume request would not see the same graph state. The sample uses a module-level shared InMemorySaver for e2e only. Real deployments should use a durable checkpointer appropriate to your environment.

Sequence Diagram

sequenceDiagram
    participant UI as AG-UI
    participant LGA as LangGraph Agent

    UI->>LGA: invoke(RunAgentInput)
    LGA-->>UI: RunStartedEvent

    Note over LGA: Working Agent Progression<br/>until user defined conditional edge

    LGA-->>UI: Interrupt: ToolCall (name = ui-confirmation)
    LGA-->>UI: RunFinishedEvent (interrupted = true, kind = on_interrupt)

    Note over UI: Display prompt approval to User<br/>(Implementation Required for AG-UI tool: ui-confirmation)

    alt Resume with explicit payload
        UI->>LGA: invoke(state = langgraph_resume)
        LGA->>LGA: _build_input_command → Command(resume=…, goto=START)
    else Resume with next user message (default)
        UI->>LGA: invoke(user message)
        LGA->>LGA: map user text → Command(resume=…, goto=START)
    end

    LGA-->>UI: RunStartedEvent

    Note over LGA: Conditional logic to<br/>continue, halt, etc.

    LGA-->>UI: RunFinishedEvent

Location	What it shows
`e2e-tests/dragent_tests/test_interrupt_resume.py`	HTTP/SSE: interrupt stream, then resume with a plain user message.
`datarobot_genai.langgraph.agent`	`LANGGRAPH_RESUME_STATE_KEY`, input resolution, and AG-UI events.