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.