langchain-langgraph-human-in-loop

What this skill does

# 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:

```python
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.

| 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](references/interrupt-decision-tree.md) 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:

```python
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](references/state-serialization-for-interrupts.md)
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.

```python
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": "[email protected]"}, 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:

```python
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](references/resume-patterns.md) 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:

```json
[
  {
    "thread_id": "conv-abc123",
    "checkpoint_id": "01JABC...",
    "interrupted_at": "2026-04-21T15:32:11Z",
    "node": "send_email",
    "state_diff": {"draft": {"to": "[email protected]", "subject": "Welcome"}}
  }
]
```

**POST `/approvals/<thread-id>/decision`** applies the decision:

```json
{
  "decision": "approve" | "reject" | "edit",
  "edits": {"recipient": "[email protected]"},
  "approver": "[email protected]",
  "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 (B