Langchain Langgraph Human In Loop

LangChain LangGraph Human-in-the-Loop (Python)

Overview

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:

graph.invoke(Command(update={"messages": [corrected_msg]}, resume="approved"), config)

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.

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:

Ship a pre-interrupt scanner in dev and CI:

import json
from typing import Any

class NonSerializableStateError(TypeError):
    """Raised when state contains values the checkpointer cannot serialize."""

def assert_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 is None or isinstance(v, (bool, int, float, str)):
        return
    if isinstance(v, list):
        for i, item in enumerate(v):
            _walk(item, f"{path}[{i}]")
        return
    if isinstance(v, dict):
        for k, val in v.items():
            _walk(val, f"{path}.{k}")
        return
    raise 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

class AgentState(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.

Step 4 — Wire the approval UI

Two HTTP endpoints. Keep them boring.

GET /approvals/pending lists paused threads:

[
  {
    "thread_id": "conv-abc123",
    "checkpoint_id": "01JABC...",
    "interrupted_at": "2026-04-21T15:32:11Z",
    "node": "send_email",
    "state_diff": {"draft": {"to": "user@example.com", "subject": "Welcome"}}
  }
]

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.

Server-side flow: authz → idempotency dedupe → checkpoint check → audit-log write (BEFORE mutation) → build Command → graph.ainvoke(cmd, config) → audit-log finalize.

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

def route_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,
})

Pattern B — Command(goto=END) at resume:

graph.invoke(Command(resume="rejected", goto=END), config)

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

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
• JSON-only state: datetime → ISO strings, bytes → base64, Pydantic → .model_dump(mode="json"), custom classes → dicts
• 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

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.

Resources

• LangGraph: Human-in-the-loop (concepts)
• LangGraph: How to add human-in-the-loop
• LangGraph: Command type reference
• LangGraph: interrupt function reference
• LangGraph: Persistence and checkpointers
• LangGraph 1.0 release notes
• Pack pain catalog: docs/pain-catalog.md (entries P16, P17, P18, P20)
• Related skills: langchain-langgraph-basics, langchain-langgraph-checkpointing, langchain-middleware-patterns

Source: jeremylongshore/claude-code-plugins-plus-skills → skills/.curated/langchain-langgraph-human-in-loop/SKILL.md

Also appears in: jeremylongshore/claude-code-plugins-plus-skills/plugins/saas-packs/langchain-py-pack/skills/langchain-langgraph-human-in-loop/SKILL.md