Build LangGraph 1.0 human-in-the-loop approval flows with interrupt_before /
interrupt_after and Command(resume=...) — JSON-serializable state, clean
resume semantics, and UI wiring for approval decisions. Use when adding an
approval gate before an expensive tool call, wiring a Slack/web UI for agent
approvals, or debugging a graph that crashes on interrupt.
Trigger with "langgraph human in loop", "langgraph interrupt_before",
"langgraph approval flow", "Command resume", "langgraph HITL".
Standardmäßig ist der Prompt ausgewählt, der zuerst die Quelle prüft. Sie können zu einem direkten Befehl wechseln oder eine lokale Kopie herunterladen.
Quelldateien prüfen
Lesen Sie SKILL.md und alle von SkillsMP angezeigten Begleitdateien, bevor Sie sich für eine Installation entscheiden.
Mit Codex oder Claude installieren Kopieren Sie diesen Prompt, fügen Sie ihn in Codex, Claude oder einen anderen Assistant ein und lassen Sie die Skill-Seite prüfen und installieren.
Ein direkter Befehl überspringt den Prüf-Prompt. Prüfen Sie die Quelle, bevor Sie ihn ausführen.
Der Befehl bleibt in einer Zeile. Scrollen Sie horizontal, um ihn vor dem Kopieren vollständig zu prüfen.
Sie bevorzugen eine lokale Kopie? Laden Sie die Dateien herunter, die SkillsMP derzeit vorliegen.
Datei-Explorer
6 Dateien
SKILL.md wird angezeigt
SKILL.md
Quellanweisungen · Schreibgeschützte Vorschau
name
langchain-langgraph-human-in-loop
description
Build LangGraph 1.0 human-in-the-loop approval flows with interrupt_before /
interrupt_after and Command(resume=...) — JSON-serializable state, clean
resume semantics, and UI wiring for approval decisions. Use when adding an
approval gate before an expensive tool call, wiring a Slack/web UI for agent
approvals, or debugging a graph that crashes on interrupt.
Trigger with "langgraph human in loop", "langgraph interrupt_before",
"langgraph approval flow", "Command resume", "langgraph HITL".
A team adds interrupt_before=["send_email"] to require a human approval
before the email goes out. First integration test crashes at the interrupt
boundary with:
TypeError: Object of type datetime is not JSON serializable
The culprit is two nodes upstream: a classify node stashed
"received_at": datetime.utcnow() into state. Every node-level unit test
passed because node completion does not serialize state — only the
checkpointer does, and only at supersteps that include an interrupt. The
failure is invisible until interrupt time (P17).
A week later the resume path ships. The human reviews the draft, clicks
"approve with edits," and the backend runs:
The prior 47 messages vanish. messages was typed as plain
list[AnyMessage] with no reducer, so update replaces the field instead of
appending (P18).
This skill covers: three interrupt styles (interrupt_before,
interrupt_after, inline interrupt()), the JSON-only state invariant with
a pre-interrupt scanner, the Command(resume=...) /
Command(update=..., resume=...) contract, an approval UI wire format
(GET pending / POST decision with optimistic concurrency), safe-cancellation
routing to END, and the tradeoff between native interrupts and a separate
approval service. Pin: langgraph 1.0.x, langgraph-checkpoint 2.0.x.
Pain-catalog anchors: P17, P18 (adjacent: P16, P20).
Prerequisites
Python 3.10+
langgraph >= 1.0, < 2.0
A checkpointer: MemorySaver (dev), PostgresSaver (prod), or SqliteSaver (single-box)
A thread_id contract at the app boundary (see langchain-langgraph-checkpointing)
Familiarity with langchain-langgraph-basics — nodes, edges, TypedDict state with reducers
Instructions
Step 1 — Choose the interrupt style
LangGraph 1.0 exposes three interrupt mechanisms. They are not interchangeable.
Style
Syntax
Use when
interrupt_before=[node]
compile(interrupt_before=["send_email"])
Review inputs before an irreversible tool. Graph pauses before node runs. State shown is the input.
interrupt_after=[node]
compile(interrupt_after=["draft_email"])
Review output of a node (e.g., an LLM draft). Graph pauses after node completes.
Inline interrupt()
Inside a node: decision = interrupt({"kind": "..."})
Structured prompt mid-node with custom payload. Most flexible; lives in node code.
Rule of thumb: prefer interrupt_before for hard gates (tool must not run
without approval). Use interrupt_after for review loops (draft → approve →
send). Use inline interrupt() when the prompt varies on intermediate
computation.
Typical interrupt round-trip latency in production is 50-300 ms from
pause to checkpoint write (local Postgres) plus UI time; budget 1-5 s
total for a Slack-based approval. Checkpoint row sizes average 2-20 KB on
small graphs and cap at ~1 MB on PostgresSaver before historical
checkpoints need pruning.
See Interrupt Decision Tree for full
criteria, multiple-interrupt-per-graph patterns, and the interrupt-vs-tool
comparison.
Step 2 — Enforce the JSON-serializable state invariant (P17)
Checkpointers serialize state to JSON on every superstep. Any non-JSON type
raises TypeError at the interrupt boundary — not at the offending node.
Canonical offenders:
Type
Fix
datetime / date
dt.isoformat() — ISO 8601 string
bytes
base64.b64encode(b).decode()
set
sorted(s)
Pydantic BaseModel with non-primitive fields
.model_dump(mode="json")
Custom classes
dataclasses.asdict(obj) or vars(obj)
numpy.ndarray
.tolist()
decimal.Decimal
str(d) or float(d) (lossy)
float("nan") / float("inf")
None (JSON forbids them; some savers crash on allow_nan=False)
Ship a pre-interrupt scanner in dev and CI:
import json
from typing importAnyclassNonSerializableStateError(TypeError):
"""Raised when state contains values the checkpointer cannot serialize."""defassert_state_is_json_serializable(state: dict[str, Any], *, path: str = "state") -> None:
"""Walk state depth-first and raise a typed error naming the offending key path."""
_walk(state, path)
def_walk(v: Any, path: str) -> None:
if v isNoneorisinstance(v, (bool, int, float, str)):
returnifisinstance(v, list):
for i, item inenumerate(v):
_walk(item, f"{path}[{i}]")
returnifisinstance(v, dict):
for k, val in v.items():
_walk(val, f"{path}.{k}")
returnraise NonSerializableStateError(
f"{path} is {type(v).__name__}, not JSON-serializable. "f"Convert at node boundary."
)
Call assert_state_is_json_serializable(state) at the end of every node
preceding an interrupt-flagged node, or attach as LangGraph middleware. In
CI, run the full graph to interrupt against a fixture that exercises every
branch — the only way to catch P17 before prod.
See State Serialization for Interrupts
for the full forbidden-types list, the Pydantic-in-state pattern, and the
integration-test harness.
Step 3 — The resume contract
Two shapes. They are not equivalent.
from langgraph.types import Command
# Shape A — resume only: human approved as-is
graph.invoke(Command(resume="approved"), config)
# Shape B — update + resume: human edited state mid-graph
graph.invoke(
Command(update={"recipient": "new@example.com"}, resume="approved"),
config,
)
resume="..." is the value returned from inline interrupt() inside the
node (if any). For interrupt_before / interrupt_after, no node reads
resume, but the checkpoint records it for audit.
update={...} merges into state via the reducer declared in the TypedDict.
Without a reducer, update replaces the field (P18). Always annotate
list and dict state:
from typing import Annotated, TypedDict
from langchain_core.messages import AnyMessage
from langgraph.graph.message import add_messages
classAgentState(TypedDict):
messages: Annotated[list[AnyMessage], add_messages] # append, not replace
approvals: Annotated[list[dict], lambda l, r: l + r] # custom append reducer
draft: Annotated[dict, lambda l, r: {**l, **r}] # dict merge reducer
last_decision: str# scalar: replace is fine
See Resume Patterns for the five canonical
resume shapes (plain approve, approve with edits, reject to END, partial
approval, inline-interrupt structured return), the reducer cookbook, and the
audit-log write order.
POST /approvals/<thread-id>/decision applies the decision:
{"decision":"approve" | "reject" | "edit","edits":{"recipient":"corrected@example.com"},"approver":"jeremy@intentsolutions.io","reason":"Verified against ticket INT-4821","expected_checkpoint_id":"01JABC...","idempotency_key":"c2f5e8a0-..."}
Optimistic concurrency (the expected_checkpoint_id check) matters the
moment two approvers open the same thread in two browser tabs. Without it,
the second click silently overwrites the first. Return 409 Conflict on
mismatch; UI refreshes.
See Approval UI Wiring for the full
HTTP contract with status codes, FastAPI implementation, Slack Block Kit
mapping, state-diff redaction, and an audit-log schema compatible with SOC2
evidence requirements.
Step 5 — Safe cancellation: route to END on reject
When the human rejects, the gated node must NOT execute. Two clean patterns:
Pattern A — conditional edge after the interrupted node (preferred):
from langgraph.graph import END
defroute_after_approval(state: AgentState) -> str:
if state.get("last_decision") == "rejected":
return END
return"send_email"
builder.add_conditional_edges("await_approval", route_after_approval, {
"send_email": "send_email",
END: END,
})
Prefer Pattern A in production: graph topology stays the source of truth,
audit replays work without the UI. Always log the rejection to the checkpoint
via Command(update={"last_decision": "rejected", "reject_reason": ...})
BEFORE routing to END — otherwise the audit trail lives only in the UI DB.
Step 6 — Native interrupts vs a separate approval service
Dimension
LangGraph interrupts
Separate approval service
Latency
50-300 ms pause + human time
Human time + queue latency
State coherence
Single source of truth (checkpoint)
Two systems to reconcile
Concurrency
Checkpoint-based optimistic locking
Whatever the queue provides
Multi-graph
Per-graph, per-thread
Centralized policy engine
Observability
get_state() + checkpoint history
Separate audit system
Failure mode
JSON-serialization at interrupt (P17)
Network partition between services
Best for
Single LangGraph app, 1-10 approval types, <1k/day
Multi-app enterprise, complex RBAC, 10k+/day
Single LangGraph app with fewer than a dozen approval types: native
interrupts are simpler and more reliable. Cross-app approval platform with
escalations, delegations, and SLAs: run a dedicated service and call it from
a tool, not from an interrupt.
Output
Graph compiled with explicit interrupt_before / interrupt_after lists, or inline interrupt() calls where payload structure matters
TypedDict state with explicit reducers on every list and dict field
Pre-interrupt state scanner attached as middleware or called at node exits; raises NonSerializableStateError with a key path
Approval HTTP endpoints: GET pending with state diffs, POST decision with expected_checkpoint_id optimistic-concurrency check and idempotency_key dedupe
Rejection routes to END via conditional edge (Pattern A) with last_decision recorded in state for audit
Audit log written BEFORE state mutation with approver, reason, thread_id, checkpoint_id_before, checkpoint_id_after
Error Handling
Error
Cause
Fix
TypeError: Object of type datetime is not JSON serializable at interrupt
Non-JSON value in state (P17)
Convert at node boundary; add pre-interrupt scanner in CI
Resume with Command(update={"messages": [new]}) loses history
messages field missing reducer (P18)
Annotate as Annotated[list[AnyMessage], add_messages]
ValueError: Thread ... has no interrupted nodes on resume
Graph already ran to completion, or thread_id mismatch
Call graph.get_state(config) first; assert snapshot.next is non-empty
Human clicks approve, nothing happens
Missing checkpointer on compile() — interrupts require persistence
graph.compile(checkpointer=MemorySaver() or PostgresSaver(...))
Two approvers both click approve, second one's edits win silently
No optimistic concurrency
Include expected_checkpoint_id in POST body; return 409 on mismatch
KeyError: 'configurable' at resume
config dict missing thread_id
config = {"configurable": {"thread_id": tid}} — required by every checkpointer
Approval UI shows stale state after another approver acted
Cached GET /pending response
Cache-Control: no-store on the pending endpoint
Graph halts silently after reject
Conditional edge router returned value not in path_map
Include END in path_map; assert router output in keyset
Examples
Approval gate before an expensive tool
Email-sending agent that must not send without approval. State carries
draft: {to, subject, body}, graph compiles with
interrupt_before=["send_email"], resume either invokes the send tool or
routes to END on reject. See
Resume Patterns for the full worked example
including audit-log write order.
Partial approval — approve one argument, edit another
Human accepts the recipient but rewrites the subject. Resume is
Command(update={"draft": {**state["draft"], "subject": new_subject}}, resume="approved").
Note the spread — without it the draft is replaced. Scalar dicts replace by
default; declare a dict reducer to merge partials cleanly. See
Resume Patterns.
Inline interrupt() with a custom payload
Inside a validate_purchase node, the model has decided to buy three items
at USD 450 total. The node calls
decision = interrupt({"kind": "confirm_purchase", "items": items, "total": 450})
and the UI reads the payload to render a rich confirmation dialog. On resume,
decision is whatever the UI sent via
Command(resume={"approved": True, "notes": "..."}). See
Interrupt Decision Tree.
Slack-driven approval
GET /pending feeds a cron that posts Block Kit messages with approve/reject
buttons. Button callback POSTs to /decision. Slack's interaction payload
carries user.id, which becomes approver in the audit log. See
Approval UI Wiring for the Block Kit
template and signing-secret validation.
