Human-in-the-Loop (HITL) & Approval Workflows

1. The Necessity of Human-in-the-Loop in Production

Fully autonomous agents without human guardrails represent severe operational liability. An unconstrained agent could:

• Accidentally execute DROP TABLE customers; in a production database.
• Send an offensive or legally binding email to a key enterprise client.
• Authorize a fraudulent wire transfer due to an indirect prompt injection attack.

Human-in-the-Loop (HITL) establishes a safety contract: the agent autonomously executes read-only research steps, but pauses and yields control whenever an action requires human authorization.

2. How LangGraph Interrupts Work

LangGraph provides the native interrupt() primitive and graph-level breakpoints:

3. Production Python Implementation: Financial Transfer Approval

Here is a practical Python script demonstrating how to pause execution before a financial transfer and resume upon human sign-off:

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

class WireState(TypedDict):
    recipient: str
    amount: float
    status: str
    admin_notes: str

# Node 1: Review and Request Human Approval
def verify_transfer_node(state: WireState):
    amount = state["amount"]
    recipient = state["recipient"]
    
    # Critical threshold: Transfers >= $1,000 require human authorization!
    if amount >= 1000.0:
        # Halt execution and return prompt to the human reviewer
        human_decision = interrupt({
            "action": "EXECUTE_WIRE_TRANSFER",
            "details": f"Authorize wire of USD {amount:.2f} to {recipient}?",
            "requires_role": "FINANCIAL_CONTROLLER"
        })
        
        # When graph resumes, human_decision contains the operator's response
        if not human_decision.get("approved"):
            return {
                "status": "REJECTED_BY_HUMAN",
                "admin_notes": human_decision.get("reason", "No reason provided")
            }
            
    return {"status": "AUTHORIZED", "admin_notes": "Approved by financial controller."}

# Node 2: Execute Transfer
def execute_transfer_node(state: WireState):
    if state["status"] == "AUTHORIZED":
        print(f"--> SUCCEEDED: Sent USD {state['amount']} to {state['recipient']}")
        return {"status": "COMPLETED"}
    else:
        print(f"--> ABORTED: Transfer cancelled due to status: {state['status']}")
        return {"status": "ABORTED"}

# Assemble Graph
workflow = StateGraph(WireState)
workflow.add_node("verify", verify_transfer_node)
workflow.add_node("execute", execute_transfer_node)

workflow.set_entry_point("verify")
workflow.add_edge("verify", "execute")
workflow.add_edge("execute", END)

# Checkpointer is mandatory for HITL to freeze state in place!
checkpointer = MemorySaver()
app = workflow.compile(checkpointer=checkpointer)

# --- Runtime Execution ---
thread_config = {"configurable": {"thread_id": "tx_99812"}}

# 1. Trigger transfer
initial_state = {"recipient": "Acme Corp", "amount": 5000.0, "status": "PENDING", "admin_notes": ""}
result = app.invoke(initial_state, config=thread_config)

print("Current Graph Status:", result)  # Graph has paused at interrupt!

# 2. Human Controller Reviews and Approves
print("\nHuman Controller approving the transaction...")
resumed_result = app.invoke(
    Command(resume={"approved": True, "reason": "Verified invoice #1042."}),
    config=thread_config
)

print("Final Execution State:", resumed_result)