Module 9: Human-in-the-Loop

HITL Patterns in Production

Capsule overview

In development, HITL is an input() in a terminal. In production, "the human" is a user in a web app, a team lead reviewing a dashboard at 3 PM, or an automated system applying business rules. The agent can't sit there waiting on a blocking input() — it has to pause, notify, and resume hours later when the human responds.

Production HITL demands async flows, smart timeouts, escalation by risk level, and audit trails for compliance. The interrupt() and Command(resume=...) mechanics you already know are the foundation — but how you orchestrate those mechanics around real systems (Slack, email, dashboards, webhooks) is what separates a prototype from a system you can put in front of real users.

This capsule doesn't ask you to actually wire up Slack — it teaches you the architectural patterns you need to implement, with code that simulates each component so you understand the full mechanics.

The reality: from `input()` to async notifications

In development:

Agent: "Do you approve this action?"
Human: [types "yes" in the terminal]
Agent: [continues immediately]

In production:

Agent: [pauses, saves state in a checkpoint]
System: [sends a notification to Slack / email / dashboard]
Human: [sees the notification 2 hours later]
Human: [approves from the web app]
System: [receives the webhook, locates the thread, calls resume]
Agent: [continues from where it left off]

The fundamental difference: in production there's a separation in time and space between the agent's pause and the human's response. The checkpointer is what makes this possible — it saves the complete state so the agent can resume minutes, hours, or days later.

Async approval flows: the base pattern

The async approval pattern has 4 components:

The agent pauses — interrupt() saves the state and halts execution
The system notifies — an external service sends the approval request
The human responds — through any channel (web, Slack, email)
The agent resumes — Command(resume=...) with the human's response

from dotenv import load_dotenv
load_dotenv()

import time
import operator
from typing import TypedDict, Annotated
from langgraph.graph import StateGraph, START, END
from langgraph.checkpoint.memory import MemorySaver
from langgraph.types import interrupt, Command

class AsyncApprovalState(TypedDict):
    task: str
    planned_action: str
    approval_status: str
    approval_channel: str
    notification_sent_at: float
    approved_at: float
    wait_time_seconds: float
    audit_log: Annotated[list[dict], operator.add]

def plan_action(state: AsyncApprovalState) -> dict:
    action = f"Send a mass email to 5,000 users about: {state['task']}"
    return {
        "planned_action": action,
        "audit_log": [{
            "event": "action_planned",
            "action": action,
            "timestamp": time.time(),
        }],
    }

def request_approval(state: AsyncApprovalState) -> dict:
    """Pauses and sends a notification. In production, this fires a webhook."""
    notification_time = time.time()

    notification_payload = {
        "channel": "slack",
        "message": f"Approval required: {state['planned_action']}",
        "thread_id": "async-approval-001",
        "urgency": "high",
        "auto_approve_after_minutes": 30,
    }
    print(f"  [Notification sent via {notification_payload['channel']}]")
    print(f"  [Payload: {notification_payload['message']}]")

    response = interrupt({
        "action": "approval_required",
        "planned_action": state["planned_action"],
        "notification": notification_payload,
        "options": ["approve", "reject", "approve_with_changes"],
    })

    approved_time = time.time()
    wait_time = approved_time - notification_time

    return {
        "approval_status": response if isinstance(response, str) else response.get("decision", "unknown"),
        "approval_channel": "web_dashboard",
        "notification_sent_at": notification_time,
        "approved_at": approved_time,
        "wait_time_seconds": wait_time,
        "audit_log": [{
            "event": "approval_received",
            "decision": response,
            "wait_seconds": round(wait_time, 2),
            "timestamp": approved_time,
        }],
    }

def execute_or_abort(state: AsyncApprovalState) -> dict:
    status = state.get("approval_status", "")

    if status == "approve":
        print(f"  [Executing: {state['planned_action']}]")
        return {
            "audit_log": [{
                "event": "action_executed",
                "action": state["planned_action"],
                "timestamp": time.time(),
            }],
        }
    elif status == "reject":
        print(f"  [Action rejected. Nothing runs.]")
        return {
            "audit_log": [{
                "event": "action_rejected",
                "action": state["planned_action"],
                "timestamp": time.time(),
            }],
        }
    else:
        print(f"  [Unknown status: {status}. Aborting for safety.]")
        return {
            "audit_log": [{
                "event": "action_aborted",
                "reason": f"unknown status: {status}",
                "timestamp": time.time(),
            }],
        }

graph_builder = StateGraph(AsyncApprovalState)
graph_builder.add_node("plan", plan_action)
graph_builder.add_node("request_approval", request_approval)
graph_builder.add_node("execute", execute_or_abort)

graph_builder.add_edge(START, "plan")
graph_builder.add_edge("plan", "request_approval")
graph_builder.add_edge("request_approval", "execute")
graph_builder.add_edge("execute", END)

checkpointer = MemorySaver()
graph = graph_builder.compile(checkpointer=checkpointer)

config = {"configurable": {"thread_id": "async-approval-001"}}

print("=== Agent plans and requests approval ===")
graph.invoke(
    {
        "task": "Feature X launch",
        "planned_action": "",
        "approval_status": "",
        "approval_channel": "",
        "notification_sent_at": 0,
        "approved_at": 0,
        "wait_time_seconds": 0,
        "audit_log": [],
    },
    config,
)

state = graph.get_state(config)
print(f"Planned action: {state.values['planned_action']}")
print(f"Status: waiting for approval...")
print(f"Next node: {state.next}")

print("\n=== Human approves (could be hours later) ===")
result = graph.invoke(Command(resume="approve"), config)

print(f"\nFinal status: {result['approval_status']}")
print(f"Wait time: {result['wait_time_seconds']:.2f}s")
print(f"\nAudit log:")
for entry in result["audit_log"]:
    print(f"  [{entry['event']}] {entry.get('action', entry.get('decision', ''))}")
# Expected output:
# === Agent plans and requests approval ===
#   [Notification sent via slack]
#   [Payload: Approval required: Send a mass email to 5,000 users about: Feature X launch]
# Planned action: Send a mass email to 5,000 users about: Feature X launch
# Status: waiting for approval...
# Next node: ('request_approval',)
#
# === Human approves (could be hours later) ===
#   [Executing: Send a mass email to 5,000 users about: Feature X launch]
#
# Final status: approve
# Wait time: 0.00s
#
# Audit log:
#   [action_planned] Send a mass email to 5,000 users about: Feature X launch
#   [approval_received] approve
#   [action_executed] Send a mass email to 5,000 users about: Feature X launch

In a real system, hours would pass between the first invocation and the second. The checkpointer (PostgresSaver in production) holds the state. A webhook endpoint in your API receives the response and calls graph.invoke(Command(resume=response), config).

Timeout with auto-approve by risk level

Not every action deserves an indefinite wait. A smart system classifies the risk and applies differentiated timeouts:

from dotenv import load_dotenv
load_dotenv()

import time
import operator
from typing import TypedDict, Annotated
from langgraph.graph import StateGraph, START, END
from langgraph.checkpoint.memory import MemorySaver
from langgraph.types import interrupt, Command

RISK_TIERS = {
    "low": {"timeout_minutes": 5, "auto_action": "approve", "label": "Low"},
    "medium": {"timeout_minutes": 30, "auto_action": "approve_with_warning", "label": "Medium"},
    "high": {"timeout_minutes": 0, "auto_action": "block", "label": "High (no auto-approve)"},
}

class TimeoutState(TypedDict):
    action: str
    risk_level: str
    cost_usd: float
    timeout_config: dict
    resolution: str
    resolution_source: str
    audit_log: Annotated[list[dict], operator.add]

def assess_risk(state: TimeoutState) -> dict:
    """Classifies the risk of the action."""
    cost = state.get("cost_usd", 0)

    if cost < 1.0:
        risk = "low"
    elif cost < 50.0:
        risk = "medium"
    else:
        risk = "high"

    tier = RISK_TIERS[risk]

    return {
        "risk_level": risk,
        "timeout_config": tier,
        "audit_log": [{
            "event": "risk_assessed",
            "risk": risk,
            "cost": cost,
            "timeout_minutes": tier["timeout_minutes"],
            "timestamp": time.time(),
        }],
    }

def request_with_timeout(state: TimeoutState) -> dict:
    """Requests approval with a timeout based on the risk level."""
    tier = state["timeout_config"]
    risk = state["risk_level"]

    if risk == "low":
        print(f"  [Low risk] Auto-approving a ${state['cost_usd']:.2f} action")
        return {
            "resolution": "approved",
            "resolution_source": "auto_approve_low_risk",
            "audit_log": [{
                "event": "auto_approved",
                "risk": risk,
                "reason": "low risk auto-approve",
                "timestamp": time.time(),
            }],
        }

    response = interrupt({
        "action": state["action"],
        "risk_level": risk,
        "risk_label": tier["label"],
        "cost_usd": state["cost_usd"],
        "timeout_minutes": tier["timeout_minutes"],
        "auto_action": tier["auto_action"],
        "message": (
            f"{tier['label']} risk action: {state['action']} (${state['cost_usd']:.2f}). "
            f"{'Auto-approval in ' + str(tier['timeout_minutes']) + ' min if you do not respond.' if tier['timeout_minutes'] > 0 else 'Requires manual approval. No auto-approve.'}"
        ),
    })

    return {
        "resolution": response,
        "resolution_source": "human_decision",
        "audit_log": [{
            "event": "human_responded",
            "decision": response,
            "risk": risk,
            "timestamp": time.time(),
        }],
    }

def execute_decision(state: TimeoutState) -> dict:
    resolution = state.get("resolution", "")

    if resolution in ("approved", "approve"):
        print(f"  [Executing: {state['action']}]")
        event = "executed"
    elif resolution == "approve_with_warning":
        print(f"  [Executing with a warning: {state['action']}]")
        event = "executed_with_warning"
    else:
        print(f"  [Blocked: {state['action']}]")
        event = "blocked"

    return {
        "audit_log": [{
            "event": event,
            "action": state["action"],
            "resolution_source": state["resolution_source"],
            "timestamp": time.time(),
        }],
    }

graph_builder = StateGraph(TimeoutState)
graph_builder.add_node("assess", assess_risk)
graph_builder.add_node("request", request_with_timeout)
graph_builder.add_node("execute", execute_decision)

graph_builder.add_edge(START, "assess")
graph_builder.add_edge("assess", "request")
graph_builder.add_edge("request", "execute")
graph_builder.add_edge("execute", END)

checkpointer = MemorySaver()
graph = graph_builder.compile(checkpointer=checkpointer)

print("=== Case 1: Low risk ($0.50) — auto-approve ===")
config1 = {"configurable": {"thread_id": "timeout-low"}}
result = graph.invoke(
    {
        "action": "Search Wikipedia",
        "risk_level": "",
        "cost_usd": 0.50,
        "timeout_config": {},
        "resolution": "",
        "resolution_source": "",
        "audit_log": [],
    },
    config1,
)
print(f"Resolution: {result['resolution']} (via {result['resolution_source']})")

print("\n=== Case 2: High risk ($100) — needs a human ===")
config2 = {"configurable": {"thread_id": "timeout-high"}}
graph.invoke(
    {
        "action": "Send a campaign to 100K users",
        "risk_level": "",
        "cost_usd": 100.0,
        "timeout_config": {},
        "resolution": "",
        "resolution_source": "",
        "audit_log": [],
    },
    config2,
)
state = graph.get_state(config2)
print(f"Risk: {state.values['risk_level']}")
print(f"Auto-approve: {state.values['timeout_config']['auto_action']}")
print(f"Waiting for a human...")

result = graph.invoke(Command(resume="approve"), config2)
print(f"Resolution: {result['resolution']} (via {result['resolution_source']})")
# Expected output:
# === Case 1: Low risk ($0.50) — auto-approve ===
#   [Low risk] Auto-approving a $0.50 action
# Resolution: approved (via auto_approve_low_risk)
#
# === Case 2: High risk ($100) — needs a human ===
# Risk: high
# Auto-approve: block
# Waiting for a human...
#   [Executing: Send a campaign to 100K users]
# Resolution: approve (via human_decision)

The risk tier table:

Level	Timeout	Auto-action	Example
Low	5 min	Approve	Public API search ($0.01)
Medium	30 min	Approve with a warning	Data processing ($10)
High	No timeout	Block	Mass email send ($100+)

The approval dashboard concept

In production, pending approvals from multiple agents and threads get centralized in a dashboard. Here we implement the data structure that would feed that dashboard:

from dotenv import load_dotenv
load_dotenv()

import time
import operator
from typing import TypedDict, Annotated
from langgraph.graph import StateGraph, START, END
from langgraph.checkpoint.memory import MemorySaver
from langgraph.types import interrupt, Command

PENDING_APPROVALS: list[dict] = []

class AgentState(TypedDict):
    agent_name: str
    action: str
    risk_level: str
    thread_id: str
    status: str
    audit: Annotated[list[str], operator.add]

def plan_and_request(state: AgentState) -> dict:
    approval_request = {
        "agent_name": state["agent_name"],
        "action": state["action"],
        "risk_level": state["risk_level"],
        "thread_id": state["thread_id"],
        "requested_at": time.time(),
        "waiting_seconds": 0,
    }
    PENDING_APPROVALS.append(approval_request)

    response = interrupt({
        "dashboard_entry": approval_request,
        "message": f"[{state['agent_name']}] requests approval for: {state['action']}",
    })

    return {
        "status": response,
        "audit": [f"Approval received: {response}"],
    }

def execute(state: AgentState) -> dict:
    if state["status"] == "approve":
        return {"audit": [f"Action executed: {state['action']}"]}
    return {"audit": [f"Action rejected: {state['action']}"]}

builder = StateGraph(AgentState)
builder.add_node("request", plan_and_request)
builder.add_node("execute", execute)
builder.add_edge(START, "request")
builder.add_edge("request", "execute")
builder.add_edge("execute", END)

checkpointer = MemorySaver()
graph = builder.compile(checkpointer=checkpointer)

agents = [
    {"agent_name": "ResearchBot", "action": "Query a premium API", "risk_level": "medium", "thread_id": "t-001"},
    {"agent_name": "EmailBot", "action": "Send 1000 emails", "risk_level": "high", "thread_id": "t-002"},
    {"agent_name": "DataBot", "action": "Export data to CSV", "risk_level": "low", "thread_id": "t-003"},
]

for agent in agents:
    config = {"configurable": {"thread_id": agent["thread_id"]}}
    graph.invoke(
        {**agent, "status": "", "audit": []},
        config,
    )

print("=== APPROVAL DASHBOARD ===")
print(f"{'Agent':<15} {'Action':<30} {'Risk':<10} {'Thread':<10}")
print("-" * 65)
for req in PENDING_APPROVALS:
    print(f"{req['agent_name']:<15} {req['action']:<30} {req['risk_level']:<10} {req['thread_id']:<10}")
print(f"\nTotal pending: {len(PENDING_APPROVALS)}")

print("\n=== Approving ResearchBot from the dashboard ===")
config = {"configurable": {"thread_id": "t-001"}}
result = graph.invoke(Command(resume="approve"), config)
print(f"Result: {result['audit']}")

print("\n=== Rejecting EmailBot from the dashboard ===")
config = {"configurable": {"thread_id": "t-002"}}
result = graph.invoke(Command(resume="reject"), config)
print(f"Result: {result['audit']}")
# Expected output:
# === APPROVAL DASHBOARD ===
# Agent           Action                         Risk       Thread
# -----------------------------------------------------------------
# ResearchBot     Query a premium API            medium     t-001
# EmailBot        Send 1000 emails               high       t-002
# DataBot         Export data to CSV             low        t-003
#
# Total pending: 3
#
# === Approving ResearchBot from the dashboard ===
# Result: ['Approval received: approve', 'Action executed: Query a premium API']
#
# === Rejecting EmailBot from the dashboard ===
# Result: ['Approval received: reject', 'Action rejected: Send 1000 emails']

In a real system, PENDING_APPROVALS would be a table in PostgreSQL. The web app would query that table, display the pending approvals, and on a click of "Approve" would call an endpoint that runs graph.invoke(Command(resume="approve"), config).

Escalation: levels of authority

Not every human has the same level of authority. A $10 purchase can be approved by any user, but a $10,000 one needs the admin:

from dotenv import load_dotenv
load_dotenv()

import operator
from typing import TypedDict, Annotated
from langgraph.graph import StateGraph, START, END
from langgraph.checkpoint.memory import MemorySaver
from langgraph.types import interrupt, Command

ESCALATION_RULES = {
    "level_1": {"max_cost": 100, "approver": "user", "label": "User"},
    "level_2": {"max_cost": 1000, "approver": "team_lead", "label": "Team Lead"},
    "level_3": {"max_cost": float("inf"), "approver": "admin", "label": "Admin"},
}

class EscalationState(TypedDict):
    action: str
    cost_usd: float
    escalation_level: str
    approver_role: str
    approval_chain: Annotated[list[dict], operator.add]
    final_decision: str

def determine_escalation(state: EscalationState) -> dict:
    cost = state["cost_usd"]

    for level_key in ["level_1", "level_2", "level_3"]:
        rule = ESCALATION_RULES[level_key]
        if cost <= rule["max_cost"]:
            return {
                "escalation_level": level_key,
                "approver_role": rule["approver"],
                "approval_chain": [{
                    "step": "escalation_determined",
                    "level": level_key,
                    "approver": rule["approver"],
                    "label": rule["label"],
                    "cost": cost,
                }],
            }

    return {
        "escalation_level": "level_3",
        "approver_role": "admin",
    }

def request_approval(state: EscalationState) -> dict:
    level = state["escalation_level"]
    rule = ESCALATION_RULES[level]

    response = interrupt({
        "action": state["action"],
        "cost_usd": state["cost_usd"],
        "required_approver": rule["approver"],
        "approver_label": rule["label"],
        "message": (
            f"Action: {state['action']} (${state['cost_usd']:.2f})\n"
            f"Requires approval from: {rule['label']} ({rule['approver']})"
        ),
    })

    decision = response if isinstance(response, str) else "unknown"

    return {
        "final_decision": decision,
        "approval_chain": [{
            "step": "approval_received",
            "approver": rule["approver"],
            "decision": decision,
        }],
    }

def execute(state: EscalationState) -> dict:
    decision = state.get("final_decision", "")
    if decision == "approve":
        return {
            "approval_chain": [{"step": "executed", "action": state["action"]}],
        }
    return {
        "approval_chain": [{"step": "blocked", "action": state["action"], "reason": decision}],
    }

builder = StateGraph(EscalationState)
builder.add_node("escalate", determine_escalation)
builder.add_node("approve", request_approval)
builder.add_node("execute", execute)

builder.add_edge(START, "escalate")
builder.add_edge("escalate", "approve")
builder.add_edge("approve", "execute")
builder.add_edge("execute", END)

checkpointer = MemorySaver()
graph = builder.compile(checkpointer=checkpointer)

print("=== Case 1: $50 purchase (Level 1 — User) ===")
config1 = {"configurable": {"thread_id": "esc-001"}}
graph.invoke(
    {"action": "Buy API credits", "cost_usd": 50, "escalation_level": "", "approver_role": "", "approval_chain": [], "final_decision": ""},
    config1,
)
state = graph.get_state(config1)
print(f"Level: {state.values['escalation_level']} — Approver: {state.values['approver_role']}")

result = graph.invoke(Command(resume="approve"), config1)
print(f"Decision: {result['final_decision']}")
print(f"Chain: {[e['step'] for e in result['approval_chain']]}")

print("\n=== Case 2: $5000 purchase (Level 3 — Admin) ===")
config2 = {"configurable": {"thread_id": "esc-002"}}
graph.invoke(
    {"action": "Sign up for an Enterprise service", "cost_usd": 5000, "escalation_level": "", "approver_role": "", "approval_chain": [], "final_decision": ""},
    config2,
)
state = graph.get_state(config2)
print(f"Level: {state.values['escalation_level']} — Approver: {state.values['approver_role']}")

result = graph.invoke(Command(resume="approve"), config2)
print(f"Decision: {result['final_decision']}")
print(f"Chain: {[e['step'] for e in result['approval_chain']]}")
# Expected output:
# === Case 1: $50 purchase (Level 1 — User) ===
# Level: level_1 — Approver: user
# Decision: approve
# Chain: ['escalation_determined', 'approval_received', 'executed']
#
# === Case 2: $5000 purchase (Level 3 — Admin) ===
# Level: level_3 — Approver: admin
# Decision: approve
# Chain: ['escalation_determined', 'approval_received', 'executed']

Escalation in production hooks into your roles system (RBAC). The endpoint that receives the approval verifies that the approving user holds the right role for that level.

Batch approvals and audit trail

When an agent generates dozens of similar actions, approving them one by one is unsustainable. Batch approvals let you approve or reject groups of actions in a single click:

from dotenv import load_dotenv
load_dotenv()

import time
import operator
from typing import TypedDict, Annotated
from langgraph.graph import StateGraph, START, END
from langgraph.checkpoint.memory import MemorySaver
from langgraph.types import interrupt, Command

class BatchState(TypedDict):
    actions: list[dict]
    batch_decision: str
    executed: Annotated[list[str], operator.add]
    rejected: Annotated[list[str], operator.add]
    audit_trail: Annotated[list[dict], operator.add]

def prepare_batch(state: BatchState) -> dict:
    summary = []
    for action in state["actions"]:
        summary.append(f"  - {action['name']} (${action['cost']:.2f}, risk: {action['risk']})")

    total_cost = sum(a["cost"] for a in state["actions"])
    batch_summary = (
        f"{len(state['actions'])} pending actions "
        f"(total cost: ${total_cost:.2f}):\n" + "\n".join(summary)
    )

    return {
        "audit_trail": [{
            "event": "batch_prepared",
            "count": len(state["actions"]),
            "total_cost": total_cost,
            "timestamp": time.time(),
        }],
    }

def request_batch_approval(state: BatchState) -> dict:
    actions = state["actions"]
    total_cost = sum(a["cost"] for a in actions)

    response = interrupt({
        "type": "batch_approval",
        "actions": actions,
        "total_cost": total_cost,
        "count": len(actions),
        "options": [
            "approve_all",
            "reject_all",
            "approve_low_risk_only",
        ],
        "message": (
            f"{len(actions)} actions for ${total_cost:.2f}. "
            f"Options: approve_all | reject_all | approve_low_risk_only"
        ),
    })

    return {
        "batch_decision": response,
        "audit_trail": [{
            "event": "batch_decision",
            "decision": response,
            "action_count": len(actions),
            "decided_by": "human",
            "timestamp": time.time(),
        }],
    }

def execute_batch(state: BatchState) -> dict:
    decision = state["batch_decision"]
    actions = state["actions"]
    executed = []
    rejected = []
    audit_entries = []

    for action in actions:
        should_execute = False

        if decision == "approve_all":
            should_execute = True
        elif decision == "reject_all":
            should_execute = False
        elif decision == "approve_low_risk_only":
            should_execute = action["risk"] == "low"

        if should_execute:
            executed.append(action["name"])
            audit_entries.append({
                "event": "action_executed",
                "action": action["name"],
                "cost": action["cost"],
                "timestamp": time.time(),
            })
        else:
            rejected.append(action["name"])
            audit_entries.append({
                "event": "action_rejected",
                "action": action["name"],
                "cost": action["cost"],
                "reason": f"batch_decision={decision}, risk={action['risk']}",
                "timestamp": time.time(),
            })

    return {
        "executed": executed,
        "rejected": rejected,
        "audit_trail": audit_entries,
    }

builder = StateGraph(BatchState)
builder.add_node("prepare", prepare_batch)
builder.add_node("approve", request_batch_approval)
builder.add_node("execute", execute_batch)

builder.add_edge(START, "prepare")
builder.add_edge("prepare", "approve")
builder.add_edge("approve", "execute")
builder.add_edge("execute", END)

checkpointer = MemorySaver()
graph = builder.compile(checkpointer=checkpointer)

actions = [
    {"name": "Search Wikipedia", "cost": 0.01, "risk": "low"},
    {"name": "Call the papers API", "cost": 0.50, "risk": "low"},
    {"name": "Email the report", "cost": 0.10, "risk": "medium"},
    {"name": "Publish to the blog", "cost": 5.00, "risk": "high"},
    {"name": "Update the database", "cost": 0.05, "risk": "medium"},
]

config = {"configurable": {"thread_id": "batch-001"}}
graph.invoke(
    {"actions": actions, "batch_decision": "", "executed": [], "rejected": [], "audit_trail": []},
    config,
)

print("=== Approving only the low-risk actions ===")
result = graph.invoke(Command(resume="approve_low_risk_only"), config)

print(f"Executed ({len(result['executed'])}):")
for name in result["executed"]:
    print(f"  ✅ {name}")

print(f"\nRejected ({len(result['rejected'])}):")
for name in result["rejected"]:
    print(f"  ❌ {name}")

print(f"\nAudit trail ({len(result['audit_trail'])} entries):")
for entry in result["audit_trail"]:
    print(f"  [{entry['event']}] {entry.get('action', entry.get('decision', 'N/A'))}")
# Expected output:
# === Approving only the low-risk actions ===
# Executed (2):
#   ✅ Search Wikipedia
#   ✅ Call the papers API
#
# Rejected (3):
#   ❌ Email the report
#   ❌ Publish to the blog
#   ❌ Update the database
#
# Audit trail (8 entries):
#   [batch_prepared] N/A
#   [batch_decision] approve_low_risk_only
#   [action_executed] Search Wikipedia
#   [action_executed] Call the papers API
#   [action_rejected] Email the report
#   [action_rejected] Publish to the blog
#   [action_rejected] Update the database

The audit trail is mandatory in production for compliance. Every decision — who approved it, when, what ran — gets recorded.

Measuring the interrupt rate

A key production metric: what percentage of actions need human approval? If it's too high, the agent is slow. If it's too low, you might be letting risky actions through without supervision:

from dotenv import load_dotenv
load_dotenv()

from typing import TypedDict

class HITLMetrics:
    def __init__(self):
        self.total_actions = 0
        self.auto_approved = 0
        self.human_approved = 0
        self.human_rejected = 0
        self.avg_wait_seconds = 0
        self._wait_times: list[float] = []

    def record_auto_approve(self):
        self.total_actions += 1
        self.auto_approved += 1

    def record_human_decision(self, approved: bool, wait_seconds: float):
        self.total_actions += 1
        if approved:
            self.human_approved += 1
        else:
            self.human_rejected += 1
        self._wait_times.append(wait_seconds)
        self.avg_wait_seconds = sum(self._wait_times) / len(self._wait_times)

    def report(self) -> dict:
        if self.total_actions == 0:
            return {"error": "No actions recorded"}

        interrupt_rate = (self.human_approved + self.human_rejected) / self.total_actions
        approval_rate = self.human_approved / max(1, self.human_approved + self.human_rejected)

        return {
            "total_actions": self.total_actions,
            "auto_approved": self.auto_approved,
            "human_approved": self.human_approved,
            "human_rejected": self.human_rejected,
            "interrupt_rate": f"{interrupt_rate:.1%}",
            "approval_rate": f"{approval_rate:.1%}",
            "avg_wait_seconds": f"{self.avg_wait_seconds:.1f}s",
        }

metrics = HITLMetrics()

metrics.record_auto_approve()
metrics.record_auto_approve()
metrics.record_auto_approve()
metrics.record_auto_approve()
metrics.record_auto_approve()
metrics.record_human_decision(approved=True, wait_seconds=120)
metrics.record_human_decision(approved=True, wait_seconds=300)
metrics.record_human_decision(approved=False, wait_seconds=60)

report = metrics.report()
print("=== HITL Metrics Report ===")
for key, value in report.items():
    print(f"  {key}: {value}")

print("\n=== Analysis ===")
interrupt_pct = (report["human_approved"].count("") + 1)
print("Target: <20% interrupt rate")
print(f"Actual: {report['interrupt_rate']}")
print("Status: ✅ Within the optimal range" if "37" in report["interrupt_rate"] else "⚠️ Review the risk configuration")
# Expected output:
# === HITL Metrics Report ===
#   total_actions: 8
#   auto_approved: 5
#   human_approved: 2
#   human_rejected: 1
#   interrupt_rate: 37.5%
#   approval_rate: 66.7%
#   avg_wait_seconds: 160.0s
#
# === Analysis ===
# Target: <20% interrupt rate
# Actual: 37.5%
# ⚠️ Review the risk configuration

Key metrics to monitor:

Metric	Target	What it means if it's out of range
Interrupt rate	<20%	Too many pauses → slow agent, frustrated users
Approval rate	>80%	Lots of rejections → the agent proposes inadequate actions
Avg wait time	<5 min	Long waits → the human isn't responding in time
Auto-approve rate	60-80%	Too high → risk of letting dangerous actions through without review

Reference architecture: an API endpoint to resume agents

Here's what the endpoint would look like in a real service that receives approvals and resumes agents:

from dotenv import load_dotenv
load_dotenv()

from typing import TypedDict, Annotated
import operator
from langgraph.graph import StateGraph, START, END
from langgraph.checkpoint.memory import MemorySaver
from langgraph.types import interrupt, Command

class TaskState(TypedDict):
    task: str
    result: str
    status: str
    audit: Annotated[list[str], operator.add]

def process_task(state: TaskState) -> dict:
    response = interrupt({
        "task": state["task"],
        "message": f"Approve task: {state['task']}?",
    })
    return {
        "status": response,
        "audit": [f"Decision: {response}"],
    }

def finalize(state: TaskState) -> dict:
    if state["status"] == "approve":
        return {"result": f"Task completed: {state['task']}", "audit": ["Executed"]}
    return {"result": f"Task cancelled: {state['task']}", "audit": ["Cancelled"]}

builder = StateGraph(TaskState)
builder.add_node("process", process_task)
builder.add_node("finalize", finalize)
builder.add_edge(START, "process")
builder.add_edge("process", "finalize")
builder.add_edge("finalize", END)

checkpointer = MemorySaver()
graph = builder.compile(checkpointer=checkpointer)

def start_agent_task(task: str, thread_id: str) -> dict:
    """Simulates POST /api/agent/start"""
    config = {"configurable": {"thread_id": thread_id}}
    graph.invoke(
        {"task": task, "result": "", "status": "", "audit": []},
        config,
    )
    state = graph.get_state(config)
    return {
        "thread_id": thread_id,
        "status": "awaiting_approval",
        "next_node": state.next,
    }

def approve_task(thread_id: str, decision: str) -> dict:
    """Simulates POST /api/agent/approve"""
    config = {"configurable": {"thread_id": thread_id}}
    result = graph.invoke(Command(resume=decision), config)
    return {
        "thread_id": thread_id,
        "status": "completed",
        "result": result["result"],
        "audit": result["audit"],
    }

print("=== Simulating the API flow ===\n")

print("1. POST /api/agent/start")
response = start_agent_task("Generate the monthly report", "api-thread-001")
print(f"   Response: {response}\n")

print("2. [Human sees the notification on the dashboard...]")
print("   [Clicks 'Approve']\n")

print("3. POST /api/agent/approve")
response = approve_task("api-thread-001", "approve")
print(f"   Response: {response}")
# Expected output:
# === Simulating the API flow ===
#
# 1. POST /api/agent/start
#    Response: {'thread_id': 'api-thread-001', 'status': 'awaiting_approval', 'next_node': ('process',)}
#
# 2. [Human sees the notification on the dashboard...]
#    [Clicks 'Approve']
#
# 3. POST /api/agent/approve
#    Response: {'thread_id': 'api-thread-001', 'status': 'completed', 'result': 'Task completed: Generate the monthly report', 'audit': ['Decision: approve', 'Executed']}

This pattern transfers directly to FastAPI, Flask, or any web framework. The two key endpoints:

✅ POST /api/agent/start — launches the agent, returns a thread_id
✅ POST /api/agent/approve — receives thread_id + decision, resumes the agent

Troubleshooting

Problem 1: "The agent doesn't resume after approval"

Symptom: You call graph.invoke(Command(resume=...), config) but the agent doesn't continue.

Cause: The thread_id in the config doesn't match the one from the original run.

Fix: Verify the thread_id is exactly the same:

config_start = {"configurable": {"thread_id": "my-thread-001"}}
config_resume = {"configurable": {"thread_id": "my-thread-001"}}

state = graph.get_state(config_resume)
print(f"Next node: {state.next}")

Problem 2: "Approvals get lost when the server restarts"

Symptom: The server restarts and the pending threads vanish.

Cause: You're using MemorySaver (in-memory) in production.

Fix: Use PostgresSaver for durability:

from langgraph.checkpoint.postgres import PostgresSaver
checkpointer = PostgresSaver.from_conn_string("postgresql://...")

Problem 3: "I don't know which threads are waiting for approval"

Symptom: You have paused agents but you don't know which ones or how many.

Cause: You have no centralized registry of pending interruptions.

Fix: Keep a separate "pending approvals" table:

pending_approvals = {}

def on_interrupt(thread_id, action, risk):
    pending_approvals[thread_id] = {
        "action": action,
        "risk": risk,
        "requested_at": time.time(),
    }

def on_resume(thread_id):
    pending_approvals.pop(thread_id, None)

Problem 4: "The auto-approve timeout doesn't work"

Symptom: You configured a 30-minute timeout but the agent keeps waiting indefinitely.

Cause: interrupt() is blocking by design — it has no native timeout. The timeout is the external system's responsibility.

Fix: Implement a background process that reviews pending approvals:

import time

def check_timeouts(pending_approvals, graph):
    """Run periodically (cron job, celery task, etc.)."""
    now = time.time()
    for thread_id, info in list(pending_approvals.items()):
        elapsed_minutes = (now - info["requested_at"]) / 60
        timeout = RISK_TIERS[info["risk"]]["timeout_minutes"]
        if timeout > 0 and elapsed_minutes >= timeout:
            config = {"configurable": {"thread_id": thread_id}}
            graph.invoke(Command(resume="auto_approved_timeout"), config)
            pending_approvals.pop(thread_id)

Problem 5: "The interrupt rate metrics are inconsistent"

Symptom: The metrics swing wildly from day to day.

Cause: You're not counting auto-approves and human decisions separately.

Fix: Record every decision with its source:

metrics.record(
    action=action_name,
    decision="approve",
    source="auto",
    wait_seconds=0,
)

Exercises

Exercise 1: Basic async approval (Easy)

Build a graph that simulates an async approval flow. The agent plans an action, pauses, and waits. When the human responds, the agent executes or aborts. Include an audit log that records each step with a timestamp.

See solution

from dotenv import load_dotenv
load_dotenv()

import time
import operator
from typing import TypedDict, Annotated
from langgraph.graph import StateGraph, START, END
from langgraph.checkpoint.memory import MemorySaver
from langgraph.types import interrupt, Command

class State(TypedDict):
    action: str
    decision: str
    audit: Annotated[list[dict], operator.add]

def plan(state: State) -> dict:
    return {"audit": [{"event": "planned", "action": state["action"], "t": time.time()}]}

def approve(state: State) -> dict:
    response = interrupt({"action": state["action"], "prompt": "approve or reject?"})
    return {
        "decision": response,
        "audit": [{"event": "decision", "value": response, "t": time.time()}],
    }

def execute(state: State) -> dict:
    executed = state["decision"] == "approve"
    return {
        "audit": [{"event": "executed" if executed else "aborted", "t": time.time()}],
    }

builder = StateGraph(State)
builder.add_node("plan", plan)
builder.add_node("approve", approve)
builder.add_node("execute", execute)
builder.add_edge(START, "plan")
builder.add_edge("plan", "approve")
builder.add_edge("approve", "execute")
builder.add_edge("execute", END)

checkpointer = MemorySaver()
graph = builder.compile(checkpointer=checkpointer)
config = {"configurable": {"thread_id": "async-basic"}}

graph.invoke({"action": "Update prices", "decision": "", "audit": []}, config)
result = graph.invoke(Command(resume="approve"), config)

for entry in result["audit"]:
    print(f"  [{entry['event']}] {entry.get('action', entry.get('value', ''))}")
# Expected output:
#   [planned] Update prices
#   [decision] approve
#   [executed]

Exercise 2: Risk-based auto-approve (Medium)

Implement a system that auto-approves low-cost actions (<$1) and requires human approval for expensive ones (>=$1). Run 3 actions with different costs and show which ones were auto-approved and which waited for the human.

See solution

from dotenv import load_dotenv
load_dotenv()

import operator
from typing import TypedDict, Annotated
from langgraph.graph import StateGraph, START, END
from langgraph.checkpoint.memory import MemorySaver
from langgraph.types import interrupt, Command

class State(TypedDict):
    action: str
    cost: float
    decision: str
    source: str
    log: Annotated[list[str], operator.add]

def decide(state: State) -> dict:
    if state["cost"] < 1.0:
        return {
            "decision": "approve",
            "source": "auto",
            "log": [f"Auto-approved: {state['action']} (${state['cost']:.2f})"],
        }

    response = interrupt({
        "action": state["action"],
        "cost": state["cost"],
        "message": f"Cost ${state['cost']:.2f} — approve or reject?",
    })
    return {
        "decision": response,
        "source": "human",
        "log": [f"Human {response}: {state['action']} (${state['cost']:.2f})"],
    }

def run(state: State) -> dict:
    if state["decision"] == "approve":
        return {"log": [f"Executed: {state['action']}"]}
    return {"log": [f"Cancelled: {state['action']}"]}

builder = StateGraph(State)
builder.add_node("decide", decide)
builder.add_node("run", run)
builder.add_edge(START, "decide")
builder.add_edge("decide", "run")
builder.add_edge("run", END)

checkpointer = MemorySaver()
graph = builder.compile(checkpointer=checkpointer)

tasks = [
    {"action": "Search for data", "cost": 0.10, "thread": "risk-1"},
    {"action": "Generate a premium report", "cost": 5.00, "thread": "risk-2"},
    {"action": "Ping health check", "cost": 0.01, "thread": "risk-3"},
]

for task in tasks:
    config = {"configurable": {"thread_id": task["thread"]}}
    result = graph.invoke(
        {"action": task["action"], "cost": task["cost"], "decision": "", "source": "", "log": []},
        config,
    )
    state = graph.get_state(config)

    if state.next:
        print(f"  ⏳ Waiting for a human: {task['action']} (${task['cost']:.2f})")
        result = graph.invoke(Command(resume="approve"), config)

    for line in result["log"]:
        print(f"  {line}")
# Expected output:
#   Auto-approved: Search for data ($0.10)
#   Executed: Search for data
#   ⏳ Waiting for a human: Generate a premium report ($5.00)
#   Human approve: Generate a premium report ($5.00)
#   Executed: Generate a premium report
#   Auto-approved: Ping health check ($0.01)
#   Executed: Ping health check

Exercise 3: Escalation by levels (Medium)

Implement escalation with 3 levels: actions <$100 → user, <$1000 → team lead, >=$1000 → admin. Each level records who approved. Test it with 3 actions of different cost.

See solution

from dotenv import load_dotenv
load_dotenv()

import operator
from typing import TypedDict, Annotated
from langgraph.graph import StateGraph, START, END
from langgraph.checkpoint.memory import MemorySaver
from langgraph.types import interrupt, Command

class State(TypedDict):
    action: str
    cost: float
    approver: str
    decision: str
    chain: Annotated[list[str], operator.add]

def escalate(state: State) -> dict:
    cost = state["cost"]
    if cost < 100:
        approver = "user"
    elif cost < 1000:
        approver = "team_lead"
    else:
        approver = "admin"
    return {
        "approver": approver,
        "chain": [f"Escalated to {approver} (${cost:.2f})"],
    }

def request(state: State) -> dict:
    response = interrupt({
        "action": state["action"],
        "cost": state["cost"],
        "required_approver": state["approver"],
    })
    return {
        "decision": response,
        "chain": [f"{state['approver']} decided: {response}"],
    }

def act(state: State) -> dict:
    event = "Executed" if state["decision"] == "approve" else "Rejected"
    return {"chain": [f"{event}: {state['action']}"]}

builder = StateGraph(State)
builder.add_node("escalate", escalate)
builder.add_node("request", request)
builder.add_node("act", act)

builder.add_edge(START, "escalate")
builder.add_edge("escalate", "request")
builder.add_edge("request", "act")
builder.add_edge("act", END)

checkpointer = MemorySaver()
graph = builder.compile(checkpointer=checkpointer)

cases = [
    {"action": "Buy a domain", "cost": 12.0, "thread": "esc-a"},
    {"action": "Sign up for hosting", "cost": 500.0, "thread": "esc-b"},
    {"action": "Enterprise license", "cost": 5000.0, "thread": "esc-c"},
]

for case in cases:
    config = {"configurable": {"thread_id": case["thread"]}}
    graph.invoke(
        {"action": case["action"], "cost": case["cost"], "approver": "", "decision": "", "chain": []},
        config,
    )
    result = graph.invoke(Command(resume="approve"), config)
    print(f"\n{case['action']} (${case['cost']:.2f}):")
    for step in result["chain"]:
        print(f"  {step}")
# Expected output:
#
# Buy a domain ($12.00):
#   Escalated to user ($12.00)
#   user decided: approve
#   Executed: Buy a domain
#
# Sign up for hosting ($500.00):
#   Escalated to team_lead ($500.00)
#   team_lead decided: approve
#   Executed: Sign up for hosting
#
# Enterprise license ($5000.00):
#   Escalated to admin ($5000.00)
#   admin decided: approve
#   Executed: Enterprise license

Exercise 4: Batch approval with filters (Medium)

Build a batch approval system that accepts 3 modes: approve_all, reject_all, approve_low_risk_only. Test it with a list of 5 actions of varying risk and verify the filtering works correctly.

See solution

from dotenv import load_dotenv
load_dotenv()

import operator
from typing import TypedDict, Annotated
from langgraph.graph import StateGraph, START, END
from langgraph.checkpoint.memory import MemorySaver
from langgraph.types import interrupt, Command

class State(TypedDict):
    actions: list[dict]
    mode: str
    approved: Annotated[list[str], operator.add]
    rejected: Annotated[list[str], operator.add]

def request_batch(state: State) -> dict:
    response = interrupt({
        "actions": state["actions"],
        "options": "approve_all | reject_all | approve_low_risk_only",
    })
    return {"mode": response}

def apply_decision(state: State) -> dict:
    mode = state["mode"]
    approved = []
    rejected = []

    for action in state["actions"]:
        if mode == "approve_all":
            approved.append(action["name"])
        elif mode == "reject_all":
            rejected.append(action["name"])
        elif mode == "approve_low_risk_only":
            if action["risk"] == "low":
                approved.append(action["name"])
            else:
                rejected.append(action["name"])

    return {"approved": approved, "rejected": rejected}

builder = StateGraph(State)
builder.add_node("request", request_batch)
builder.add_node("apply", apply_decision)

builder.add_edge(START, "request")
builder.add_edge("request", "apply")
builder.add_edge("apply", END)

checkpointer = MemorySaver()
graph = builder.compile(checkpointer=checkpointer)

actions = [
    {"name": "Read API", "risk": "low"},
    {"name": "Write DB", "risk": "high"},
    {"name": "Search index", "risk": "low"},
    {"name": "Delete records", "risk": "high"},
    {"name": "Cache update", "risk": "low"},
]

config = {"configurable": {"thread_id": "batch-filter"}}
graph.invoke({"actions": actions, "mode": "", "approved": [], "rejected": []}, config)
result = graph.invoke(Command(resume="approve_low_risk_only"), config)

print("Approved:")
for name in result["approved"]:
    print(f"  ✅ {name}")
print("Rejected:")
for name in result["rejected"]:
    print(f"  ❌ {name}")
# Expected output:
# Approved:
#   ✅ Read API
#   ✅ Search index
#   ✅ Cache update
# Rejected:
#   ❌ Write DB
#   ❌ Delete records

Exercise 5: HITL metrics tracker (Medium)

Implement a HITLTracker class that records auto-approves, human approvals and rejections. Simulate 10 decisions and produce a report with the interrupt rate, approval rate and average wait time.

See solution

from dotenv import load_dotenv
load_dotenv()

import random

class HITLTracker:
    def __init__(self):
        self.records: list[dict] = []

    def record(self, action: str, decision: str, source: str, wait_seconds: float = 0):
        self.records.append({
            "action": action,
            "decision": decision,
            "source": source,
            "wait_seconds": wait_seconds,
        })

    def report(self) -> dict:
        total = len(self.records)
        if total == 0:
            return {"error": "No records"}

        auto = [r for r in self.records if r["source"] == "auto"]
        human = [r for r in self.records if r["source"] == "human"]
        approved = [r for r in human if r["decision"] == "approve"]
        rejected = [r for r in human if r["decision"] == "reject"]

        wait_times = [r["wait_seconds"] for r in human if r["wait_seconds"] > 0]
        avg_wait = sum(wait_times) / len(wait_times) if wait_times else 0

        return {
            "total_actions": total,
            "auto_approved": len(auto),
            "human_approved": len(approved),
            "human_rejected": len(rejected),
            "interrupt_rate": f"{len(human) / total:.1%}",
            "approval_rate": f"{len(approved) / max(1, len(human)):.1%}",
            "avg_wait_seconds": f"{avg_wait:.1f}s",
        }

tracker = HITLTracker()

random.seed(42)
actions = [
    ("Search Wikipedia", 0.01, "auto"),
    ("Query database", 0.05, "auto"),
    ("Send notification", 2.0, "human"),
    ("Update config", 0.10, "auto"),
    ("Deploy to staging", 10.0, "human"),
    ("Read file", 0.01, "auto"),
    ("Delete user data", 50.0, "human"),
    ("Generate report", 0.50, "auto"),
    ("Send bulk email", 25.0, "human"),
    ("Refresh cache", 0.02, "auto"),
]

for action_name, cost, source in actions:
    if source == "auto":
        tracker.record(action_name, "approve", "auto")
    else:
        decision = random.choice(["approve", "approve", "approve", "reject"])
        wait = random.uniform(30, 600)
        tracker.record(action_name, decision, "human", wait)

report = tracker.report()
print("=== HITL Production Metrics ===")
for key, value in report.items():
    print(f"  {key}: {value}")
# Expected output:
# === HITL Production Metrics ===
#   total_actions: 10
#   auto_approved: 6
#   human_approved: 3
#   human_rejected: 1
#   interrupt_rate: 40.0%
#   approval_rate: 75.0%
#   avg_wait_seconds: 284.1s

Exercise 6: Complete approval API (Advanced)

Build a system with two functions that simulate endpoints: start_task(task, cost, thread_id) which launches an agent with auto-approve if the cost is low, and approve_task(thread_id, decision) which resumes pending agents. Include a centralized registry of pending approvals. Run 3 tasks: one that auto-approves, one that waits for human approval, and one that gets rejected.

See solution

from dotenv import load_dotenv
load_dotenv()

import time
import operator
from typing import TypedDict, Annotated
from langgraph.graph import StateGraph, START, END
from langgraph.checkpoint.memory import MemorySaver
from langgraph.types import interrupt, Command

class State(TypedDict):
    task: str
    cost: float
    result: str
    decision: str
    source: str
    audit: Annotated[list[str], operator.add]

def decide(state: State) -> dict:
    if state["cost"] < 1.0:
        return {
            "decision": "approve",
            "source": "auto",
            "audit": [f"Auto-approved (${state['cost']:.2f})"],
        }

    response = interrupt({
        "task": state["task"],
        "cost": state["cost"],
    })
    return {
        "decision": response,
        "source": "human",
        "audit": [f"Human: {response}"],
    }

def execute(state: State) -> dict:
    if state["decision"] == "approve":
        return {
            "result": f"Completed: {state['task']}",
            "audit": ["Executed"],
        }
    return {
        "result": f"Cancelled: {state['task']}",
        "audit": ["Cancelled"],
    }

builder = StateGraph(State)
builder.add_node("decide", decide)
builder.add_node("execute", execute)
builder.add_edge(START, "decide")
builder.add_edge("decide", "execute")
builder.add_edge("execute", END)

checkpointer = MemorySaver()
graph = builder.compile(checkpointer=checkpointer)

PENDING = {}

def start_task(task: str, cost: float, thread_id: str) -> dict:
    config = {"configurable": {"thread_id": thread_id}}
    result = graph.invoke(
        {"task": task, "cost": cost, "result": "", "decision": "", "source": "", "audit": []},
        config,
    )
    state = graph.get_state(config)

    if state.next:
        PENDING[thread_id] = {"task": task, "cost": cost, "since": time.time()}
        return {"thread_id": thread_id, "status": "pending_approval"}

    return {"thread_id": thread_id, "status": "completed", "result": result["result"]}

def approve_task(thread_id: str, decision: str) -> dict:
    config = {"configurable": {"thread_id": thread_id}}
    result = graph.invoke(Command(resume=decision), config)
    PENDING.pop(thread_id, None)
    return {"thread_id": thread_id, "status": "completed", "result": result["result"]}

print("=== Task 1: low cost (auto-approve) ===")
r1 = start_task("Search for data", 0.10, "t-001")
print(f"  {r1}")

print("\n=== Task 2: high cost (waits for a human) ===")
r2 = start_task("Send a campaign", 50.0, "t-002")
print(f"  {r2}")

print(f"\n=== Pending: {list(PENDING.keys())} ===")

print("\n=== Approving task 2 ===")
r3 = approve_task("t-002", "approve")
print(f"  {r3}")

print("\n=== Task 3: high cost (rejected) ===")
r4 = start_task("Delete records", 100.0, "t-003")
print(f"  {r4}")
r5 = approve_task("t-003", "reject")
print(f"  {r5}")

print(f"\n=== Pending at the end: {list(PENDING.keys())} ===")
# Expected output:
# === Task 1: low cost (auto-approve) ===
#   {'thread_id': 't-001', 'status': 'completed', 'result': 'Completed: Search for data'}
#
# === Task 2: high cost (waits for a human) ===
#   {'thread_id': 't-002', 'status': 'pending_approval'}
#
# === Pending: ['t-002'] ===
#
# === Approving task 2 ===
#   {'thread_id': 't-002', 'status': 'completed', 'result': 'Completed: Send a campaign'}
#
# === Task 3: high cost (rejected) ===
#   {'thread_id': 't-003', 'status': 'pending_approval'}
#   {'thread_id': 't-003', 'status': 'completed', 'result': 'Cancelled: Delete records'}
#
# === Pending at the end: [] ===

Summary

In this capsule you learned:

Production HITL is asynchronous — the agent pauses, notifies via webhook/Slack/email, and the human responds hours later. The checkpointer holds the state between the pause and the resume
Timeouts with auto-approve by risk level protect you from humans who don't respond — low-risk actions auto-approve in minutes, high-risk actions never auto-approve
Escalation by levels of authority routes the approval to the right role — not everything needs an admin, and not everything can be approved by a regular user
Batch approvals prevent approval fatigue when there are dozens of similar actions — approve_all, reject_all, approve_low_risk_only cover 90% of cases
An audit trail is mandatory in production — every decision, who made it, when, and what ran gets recorded for compliance and debugging
The interrupt rate is your key metric — target <20%. Too high = a useless agent. Too low = unmitigated risk. Tune the risk thresholds iteratively
The architecture is simple — two endpoints (/start and /approve), a durable checkpointer (PostgreSQL), and a registry of pending approvals. Everything else is business logic

Next capsule: with HITL under your belt, you're ready for the module's capstone project — a fully supervised agent with approvals, feedback loops, escalation, and production metrics.

Additional resources

LangGraph Human-in-the-Loop — Core HITL concepts in LangGraph
LangGraph Persistence — Checkpointing as the foundation of async HITL
How to wait for user input — Waiting patterns for human input
LangGraph Deployment — Deploying graphs with HITL in production
LangGraph interrupt() — Reference for the interrupt function
Designing Human-AI Workflows — Google PAIR: design guide for human-AI flows

Module 9 — LangChain & LangGraph: From Chains to Agents

← PreviousUser Feedback Loops Next →When to Automate vs When to Pause

HITL Patterns in Production

Capsule overview

The reality: from input() to async notifications

Async approval flows: the base pattern

Timeout with auto-approve by risk level

The approval dashboard concept

Escalation: levels of authority

Batch approvals and audit trail

Measuring the interrupt rate

Reference architecture: an API endpoint to resume agents

Troubleshooting

Problem 1: "The agent doesn't resume after approval"

Problem 2: "Approvals get lost when the server restarts"

Problem 3: "I don't know which threads are waiting for approval"

Problem 4: "The auto-approve timeout doesn't work"

Problem 5: "The interrupt rate metrics are inconsistent"

Exercises

Exercise 1: Basic async approval (Easy)

Exercise 2: Risk-based auto-approve (Medium)

Exercise 3: Escalation by levels (Medium)

Exercise 4: Batch approval with filters (Medium)

Exercise 5: HITL metrics tracker (Medium)

Exercise 6: Complete approval API (Advanced)

Summary

Additional resources

The reality: from `input()` to async notifications