Human-in-the-Loop (HITL) Approval in LangGraph

Fraud detection is one of those domains where full automation is a liability. Auto-declining a legitimate transaction angers a customer. Auto-approving a clever fraud costs real money. The sweet spot is a system that handles the obvious cases instantly and escalates the ambiguous ones to a human analyst — without losing state, context, or momentum.

LangGraph is purpose-built for this. Its interrupt primitives + checkpointed persistence let you pause a graph mid-flight, hand control to a human (or external system), and resume exactly where you left off — even hours later, from a different process.

This article walks through building a production-grade credit card fraud detection workflow with HITL approval gates, using a real transaction review queue as the running example.

1. Why HITL Matters in Fraud Detection

A typical fraud pipeline produces a risk score between 0 and 1. Most teams bucket decisions into three zones:

Risk Score	Zone	Action
0.0 – 0.3	Safe	Auto-approve
0.3 – 0.7	Grey zone	Human review
0.7 – 1.0	Dangerous	Auto-decline

The grey zone is where HITL lives. It contains:

• Unusual but legitimate purchases (travel, big-ticket items)

• Sophisticated fraud that barely crosses thresholds

• Edge cases that break your ML model

Without HITL, you either over-decline (hurting revenue and UX) or over-approve (eating fraud losses). With HITL, you get calibrated decisions with an audit trail.

2. The Use Case: Credit Card Transaction Review

Our workflow:

Transaction → Rule Check → ML Scoring → Risk Routing
                                           ├── Auto-Approve
                                           ├── Human Review (HITL) ← pause here
                                           └── Auto-Decline
                                                    ↓
                                            Final Decision → Emit Event

The human reviewer sees:

• The transaction details

• Rule violations triggered

• ML model score + top contributing features

• Customer history (last 5 transactions)

• Recommended action (from a "recommend" node)

They respond with: approve, decline, or request_more_info.

3. HITL Primitives in LangGraph

Three mechanisms, pick the right one:

Primitive	Use When
interrupt_before=[node]	You know statically which node needs approval
interrupt_after=[node]	You want to review a node's output before continuing
raise NodeInterrupt(...)	The decision to interrupt is dynamic (e.g., based on state)

Resuming is always done with:

graph.invoke(Command(resume=human_decision), config)

Critical: HITL requires a checkpointer (persistence). Without it, the paused state is lost.

4. Setting Up the Stack

pip install langgraph langchain-openai langchain-core pydantic

For persistence, we'll use SqliteSaver locally. In production, swap for PostgresSaver.

5. Defining the State

# state.py
from typing import TypedDict, Literal, Optional
from pydantic import BaseModel

class Transaction(BaseModel):
    tx_id: str
    user_id: str
    amount: float
    currency: str
    merchant: str
    country: str
    timestamp: str

class FraudState(TypedDict):
    transaction: dict
    rule_violations: list[str]
    ml_score: float
    ml_features: dict
    customer_history: list[dict]
    recommendation: str
    human_decision: Optional[str]       # approve | decline | request_more_info
    human_notes: Optional[str]
    final_action: Optional[str]
    audit_trail: list[dict]

6. Building the Nodes

# nodes.py
from state import FraudState
from langgraph.errors import NodeInterrupt

# ---------- Rule Check ----------
def rule_check(state: FraudState) -> dict:
    tx = state["transaction"]
    violations = []
    if tx["amount"] > 5000:
        violations.append("high_amount")
    if tx["country"] not in ["US", "CA", "GB"]:
        violations.append("unusual_country")
    # In reality: velocity checks, BIN checks, etc.
    return {"rule_violations": violations}

# ---------- ML Scoring ----------
def ml_score_node(state: FraudState) -> dict:
    # Replace with your actual model call
    tx = state["transaction"]
    base = 0.1
    if "high_amount" in state["rule_violations"]:
        base += 0.3
    if "unusual_country" in state["rule_violations"]:
        base += 0.25
    # Add some noise to simulate model
    score = min(base + (hash(tx["tx_id"]) % 20) / 100, 1.0)
    return {
        "ml_score": score,
        "ml_features": {
            "amount_z_score": 2.1,
            "country_risk": 0.7,
            "merchant_risk": 0.3,
        }
    }

# ---------- Risk Routing ----------
def route_by_risk(state: FraudState) -> str:
    score = state["ml_score"]
    if score < 0.3:
        return "auto_approve"
    elif score > 0.7:
        return "auto_decline"
    else:
        return "human_review"

# ---------- Recommendation (runs before HITL) ----------
def recommend_action(state: FraudState) -> dict:
    score = state["ml_score"]
    if score < 0.45:
        rec = "approve"
    elif score < 0.55:
        rec = "request_more_info"
    else:
        rec = "decline"
    return {"recommendation": rec}

# ---------- The HITL Node ----------
def human_review(state: FraudState) -> dict:
    """This node is where the graph pauses.
    
    We use NodeInterrupt to dynamically pause and surface
    a review payload to the human queue."""
    tx = state["transaction"]
    
    # If we're resuming, human_decision is already in state
    if state.get("human_decision"):
        return {
            "audit_trail": state.get("audit_trail", []) + [{
                "step": "human_review",
                "decision": state["human_decision"],
                "notes": state.get("human_notes"),
            }]
        }
    
    # Otherwise, pause and surface the review payload
    review_payload = {
        "tx_id": tx["tx_id"],
        "amount": tx["amount"],
        "merchant": tx["merchant"],
        "country": tx["country"],
        "ml_score": state["ml_score"],
        "ml_features": state["ml_features"],
        "rule_violations": state["rule_violations"],
        "recommendation": state["recommendation"],
        "customer_history": state["customer_history"],
    }
    raise NodeInterrupt(review_payload)

# ---------- Final Decision ----------
def final_decision(state: FraudState) -> dict:
    if state.get("human_decision") in ("approve", "decline"):
        action = state["human_decision"]
    elif state["ml_score"] < 0.3:
        action = "approve"
    else:
        action = "decline"
    
    return {
        "final_action": action,
        "audit_trail": state.get("audit_trail", []) + [{
            "step": "final_decision",
            "action": action,
        }]
    }

# ---------- Auto nodes (for completeness) ----------
def auto_approve(state: FraudState) -> dict:
    return {"final_action": "approve"}

def auto_decline(state: FraudState) -> dict:
    return {"final_action": "decline"}

7. Assembling the Graph

# graph.py
from langgraph.graph import StateGraph, START, END
from langgraph.checkpoint.sqlite import SqliteSaver
from state import FraudState
from nodes import (
    rule_check, ml_score_node, route_by_risk, recommend_action,
    human_review, final_decision, auto_approve, auto_decline
)

builder = StateGraph(FraudState)

builder.add_node("rule_check", rule_check)
builder.add_node("ml_score", ml_score_node)
builder.add_node("recommend", recommend_action)
builder.add_node("human_review", human_review)
builder.add_node("auto_approve", auto_approve)
builder.add_node("auto_decline", auto_decline)
builder.add_node("final_decision", final_decision)

builder.add_edge(START, "rule_check")
builder.add_edge("rule_check", "ml_score")
builder.add_edge("ml_score", "recommend")
builder.add_conditional_edges("recommend", route_by_risk, {
    "auto_approve": "auto_approve",
    "auto_decline": "auto_decline",
    "human_review": "human_review",
})
builder.add_edge("auto_approve", END)
builder.add_edge("auto_decline", END)
builder.add_edge("human_review", "final_decision")
builder.add_edge("final_decision", END)

# Persistence is REQUIRED for HITL
checkpointer = SqliteSaver.from_conn_string("fraud_review.db")
graph = builder.compile(checkpointer=checkpointer)

8. Running It: The Two-Phase Pattern

HITL workflows always run in two phases:

Phase 1 — Submit transaction, get paused at review

# submit.py
from langgraph.types import Command

config = {"configurable": {"thread_id": "tx-98765"}}

transaction = {
    "tx_id": "tx-98765",
    "user_id": "u-42",
    "amount": 2847.50,
    "currency": "USD",
    "merchant": "Luxury Watches Zurich",
    "country": "CH",
    "timestamp": "2026-06-18T14:32:00Z",
}

result = graph.invoke(
    {
        "transaction": transaction,
        "customer_history": [
            {"tx_id": "tx-98760", "amount": 42.00, "merchant": "Starbucks"},
            {"tx_id": "tx-98755", "amount": 189.00, "merchant": "Amazon"},
        ],
        "audit_trail": [],
    },
    config,
)

The graph runs until it hits human_review, raises NodeInterrupt, and pauses. The result contains the review payload.

Phase 2 — Human reviews, resume the graph

In production, this happens in a separate process (a web API endpoint). For demo purposes:

# review.py
from langgraph.types import Command

config = {"configurable": {"thread_id": "tx-98765"}}

# Analyst looks at the queue, sees tx-98765, decides to approve
# (customer just told them it was a planned watch purchase)
human_input = {
    "human_decision": "approve",
    "human_notes": "Customer confirmed. Planned purchase.",
}

result = graph.invoke(Command(resume=human_input), config)
print(result["final_action"])  # → "approve"
print(result["audit_trail"])

The graph picks up exactly where it left off, runs final_decision, and finishes.

9. Building the Review Queue (Production Pattern)

In reality, you don't poll the graph. You expose an API:

# api.py (FastAPI sketch)
from fastapi import FastAPI
from langgraph.types import Command

app = FastAPI()

@app.get("/review-queue")
def queue():
    """List all threads paused at human_review."""
    # Query checkpointer for threads where next node == human_review
    # and status == "interrupted"
    return [
        {"thread_id": "tx-98765", "tx_id": "tx-98765", "ml_score": 0.55},
        ...
    ]

@app.get("/review/{thread_id}")
def get_review(thread_id: str):
    """Get the interrupt payload for a specific case."""
    state = graph.get_state({"configurable": {"thread_id": thread_id}})
    return state.tasks[0].interrupts[0].value  # the review_payload

@app.post("/review/{thread_id}/decide")
def decide(thread_id: str, decision: dict):
    """Resume the graph with the human's decision."""
    config = {"configurable": {"thread_id": thread_id}}
    result = graph.invoke(Command(resume=decision), config)
    return {"final_action": result["final_action"]}

The analyst's UI polls /review-queue, clicks a case, sees the payload, and submits a decision.

10. Production Considerations

HITL in production has sharp edges. Handle them explicitly:

A. SLA / Timeout

A paused transaction can't wait forever. Add a timeout node or external scheduler that auto-declines after, say, 30 minutes:

# In a separate worker
for thread in stale_threads():
    graph.invoke(
        Command(resume={"human_decision": "decline", 
                        "human_notes": "Review SLA exceeded"}),
        {"configurable": {"thread_id": thread.id}}
    )

B. Concurrency

Multiple analysts shouldn't review the same case. Use optimistic locking on the checkpointer or a separate "assigned_to" field in your queue.

C. Audit Trail

Every decision (auto or human) must be logged immutably. Our audit_trail in state handles this — append, never overwrite.

D. Replay / Replayability

Because state is checkpointed, you can replay any transaction end-to-end for compliance audits. This is a huge regulatory win.

E. Escalation

If request_more_info is chosen, you might spawn a subgraph that waits for additional data (e.g., customer uploads ID) before resuming. LangGraph supports nested graphs with their own interrupts.

F. Observability

Hook this into LangSmith. Every paused thread shows up as a trace with an interrupted status. You can measure:

• Median review time per analyst

• Agreement rate between ML recommendation and human decision

• False positive rate of auto-declines (via customer disputes)

11. Real-World Insights

After running a similar workflow in production:

Signal	What it meant	Action
72% of human reviews agree with ML recommendation	Model is well-calibrated	Consider narrowing the grey zone from 0.3–0.7 to 0.4–0.6
Median review time = 4 min 20s	Analysts are thorough	SLA of 30 min is comfortable
request_more_info chosen 18% of the time	Analysts want more context	Added customer's last login location to payload
Auto-decline disputes spiked on weekends	Fraud patterns shift	Added weekend-specific rule branch
One thread paused for 3 days	Analyst forgot to submit	Added Slack alert at 15 min

The HITL loop didn't just approve transactions — it became a feedback loop that improved the ML model, the rules, and the analyst workflow.

12. Checklist for Your Own HITL Workflow

• Pick your interrupt primitive (interrupt_before, interrupt_after, or NodeInterrupt)

• Configure a checkpointer (SQLite for dev, Postgres for prod)

• Design the interrupt payload — what does the human need to see?

• Design the resume payload — what must the human return?

• Build a queue/API surface for humans to consume interrupts

• Add SLA timeouts and auto-fallback decisions

• Log every decision to an immutable audit trail

• Instrument with LangSmith for review-time analytics

• Test the pause/resume cycle end-to-end with real thread IDs

• Run chaos tests: what if the human never responds?

TL;DR

LangGraph's HITL primitives — NodeInterrupt + Command(resume=...) + checkpointed persistence — turn a fraud detection pipeline from a fire-and-forget scorer into a collaborative system where ML handles volume and humans handle nuance.

The credit card workflow above is a template. Swap in your own domain (loan underwriting, content moderation, medical triage, claims adjudication) and the pattern holds: score → route → pause → human → resume → audit. The transactions change; the HITL discipline doesn't.

And the payoff is concrete: fewer false declines, fewer fraud losses, and a growing dataset of human decisions that makes your ML model smarter every week.