Day 9 of LangGraph 14‑Day Series: Adding Human‑in‑the‑Loop Review to AI Workflows

Problem: AI can execute dangerous operations such as deleting data or transferring funds without control. Solution: pause the AI at critical steps and wait for human confirmation before continuing.

Why Human‑in‑the‑Loop is needed

Operations that typically require human intervention:

Deleting user data

Financial transactions

Sending important emails

Modifying system configuration

Comparison of three common approaches:

Fully automatic – high efficiency, uncontrolled risk.

Ask every time – safe, poor user experience.

Human‑in‑the‑Loop (HITL) – combines safety and efficiency, requires UI support.

Core: interrupt() mechanism

LangGraph provides interrupt() to pause execution until a human approves.

from langgraph.types import interrupt, Command

def submit_request(state: ApprovalState) -> dict:
    # Pause execution, wait for human approval
    response = interrupt(f"请审批请求: {state['request']}")
    return {"approved": response}

Features of interrupt():

Execution automatically pauses when the line is reached.

Returns to the caller with the interruption reason.

Resumes when a Command(resume=...) is sent.

Complete approval workflow

from typing import Optional, TypedDict
from langgraph.graph import StateGraph, START, END
from langgraph.checkpoint.memory import InMemorySaver
from langgraph.types import interrupt, Command

class ApprovalState(TypedDict):
    request: str
    approved: Optional[bool]
    result: str

def submit_request(state: ApprovalState) -> dict:
    """Submit request and wait for approval"""
    response = interrupt(f"请审批: {state['request']}")
    return {"approved": response}

def process_approved(state: ApprovalState) -> dict:
    return {"result": f"✅ 已批准: {state['request']}"}

memory = InMemorySaver()
builder = StateGraph(ApprovalState)
builder.add_node("submit", submit_request)
builder.add_node("approve", process_approved)
builder.add_edge(START, "submit")
builder.add_edge("submit", "approve")
builder.add_edge("approve", END)
app = builder.compile(checkpointer=memory)

Execution flow

Step 1: Submit request
result = app.invoke({"request": "删除所有数据", ...}, config)
# Returns: {'__interrupt__': (Interrupt(value='请审批: 删除所有数据'),)}

Step 2: Human approval (UI action)
# User clicks "Approve" or "Reject"

Step 3: Resume execution
result = app.invoke(Command(resume=True), config)
# Continues with the approve node

Command(resume=…) recovery mechanism

# Approve
result = app.invoke(Command(resume=True), config)

# Reject with reason
result = app.invoke(Command(resume={"reason": "风险太高"}), config)

Parameters of Command: resume: value passed when resuming. goto (optional): jump to a specific node. update (optional): update the state.

Conditional branch: approve vs reject

def process_result(state: ApprovalState) -> dict:
    if state["approved"]:
        return {"result": f"✅ 已批准: {state['request']}"}
    else:
        return {"result": f"❌ 已拒绝: {state['request']}"}

# Add conditional edges to the graph
builder.add_conditional_edges(
    "submit",
    lambda s: "approve" if s.get("approved") else "reject",
    {"approve": "approve", "reject": "reject"}
)

Practical example: email sending approval

┌─────────────────────────────────────┐
│        Email Sending Approval       │
│  ┌─────────┐                       │
│  │  Write  │                       │
│  │  Email  │                       │
│  └────┬────┘                       │
│       │                             │
│       ▼                             │
│  ┌─────────┐                       │
│  │interrupt│ ◀── Pause, wait for human │
│  │ (approval)│                     │
│  └────┬────┘                       │
│       │                             │
│   ┌───┴───┐                         │
│   ▼       ▼                         │
│ Approve  Reject                     │
│   │       │                         │
│   ▼       ▼                         │
│ Send   Archive                       │
└─────────────────────────────────────┘

Day review

interrupt() : pauses execution and waits for human input.

Command(resume=) : resumes the paused execution.

Checkpointer : required configuration to use interrupt().

Conditional branching : directs flow based on approval result.

Related links

Official documentation: https://langchain-ai.github.io/langgraph/concepts/human_in_the_loop/

GitHub repository: https://github.com/langchain-ai/langgraph