Human in the Loop AI Agents with LangGraph: Step by Step Guide

Letting an autonomous AI agent execute code, send emails, or modify database records without oversight is a major risk in production. Even the best models make mistakes, misunderstand ambiguous instructions, or fall victim to prompt injections.

To run agents safely in high stakes environments, you need Human in the Loop (HITL) controls.

LangGraph provides built in support for human approval workflows through persistent checkpointing and the interrupt() function. When the agent reaches a dangerous action, it pauses its execution graph, saves its exact memory state to a database, and waits for a human to approve, reject, or edit the action before continuing.

Here is a step by step guide to building human approval workflows in LangGraph.

sequenceDiagram
    autonumber
    actor User as Human Operator
    participant Agent as LangGraph Agent
    participant DB as SQLite / Postgres Checkpointer
    participant API as External Service (e.g. Bank API)

    User->>Agent: "Transfer $5,000 to vendor account #8812"
    Agent->>Agent: Plan action & prepare tool arguments
    Note over Agent: Detects sensitive action: Fund Transfer
    Agent->>DB: Save complete graph state (Thread ID: 104)
    Agent-->>User: Pause and request human approval
    User->>Agent: Review & approve action with updated note
    Agent->>DB: Resume execution from Thread ID: 104
    Agent->>API: Execute fund transfer
    API-->>Agent: Transfer confirmed
    Agent-->>User: "Transfer complete. Confirmation ID: TX-9921"

Core Concepts: Checkpointers and Interrupts

LangGraph handles human approval using two primary primitives:

1. Checkpointer (SqliteSaver or PostgresSaver): A persistence engine that writes the full state of your agent graph to disk after every node execution. Because state is persisted, the server process can pause for hours or restart completely without losing the conversation history.
2. The interrupt() Function: A special call inside any graph node that halts execution immediately, yields control back to the caller, and waits for external input.

Complete Python Implementation

Here is a working example of an AI financial assistant that requires human sign off before transferring money:

1. Installation

pip install langgraph langchain-openai langchain-core

2. Implementation: Building the Approval Graph

import os
from typing import Annotated, TypedDict, Literal
from langchain_core.messages import BaseMessage, HumanMessage, AIMessage
from langchain_openai import ChatOpenAI
from langgraph.graph import StateGraph, START, END
from langgraph.graph.message import add_messages
from langgraph.checkpoint.memory import MemorySaver
from langgraph.types import interrupt, Command

# 1. Define graph state
class AgentState(TypedDict):
    messages: Annotated[list[BaseMessage], add_messages]
    pending_transfer: dict | None

# 2. Initialize language model
model = ChatOpenAI(model="gpt-4o-mini", temperature=0.0)

# Node 1: Analyze user request
def parse_request_node(state: AgentState) -> dict:
    messages = state["messages"]
    last_message = messages[-1].content
    
    # In a real app, use structured outputs with Pydantic
    if "transfer" in last_message.lower():
        # Simulated parsed transaction details
        return {
            "pending_transfer": {
                "recipient": "Vendor Corp",
                "amount_usd": 5000.0,
                "status": "pending_review"
            },
            "messages": [AIMessage(content="Transfer details prepared. Requesting human authorization.")]
        }
    
    response = model.invoke(messages)
    return {"messages": [response], "pending_transfer": None}

# Node 2: Human Approval Gate
def human_approval_node(state: AgentState) -> Command[Literal["execute_transfer_node", "cancel_transfer_node"]]:
    transfer = state.get("pending_transfer")
    if not transfer:
        return Command(goto="cancel_transfer_node")

    # Halt graph execution and wait for human input
    human_decision = interrupt({
        "question": "Do you authorize this financial transaction?",
        "recipient": transfer["recipient"],
        "amount": transfer["amount_usd"]
    })
    
    # Check decision received when resumed
    if isinstance(human_decision, dict) and human_decision.get("approved") is True:
        return Command(
            goto="execute_transfer_node",
            update={"pending_transfer": {**transfer, "status": "approved", "notes": human_decision.get("notes", "")}}
        )
    else:
        return Command(
            goto="cancel_transfer_node",
            update={"pending_transfer": {**transfer, "status": "rejected"}}
        )

# Node 3: Execute the action
def execute_transfer_node(state: AgentState) -> dict:
    transfer = state["pending_transfer"]
    confirmation = f"Successfully transferred ${transfer['amount_usd']} to {transfer['recipient']}."
    return {
        "messages": [AIMessage(content=confirmation)],
        "pending_transfer": None
    }

# Node 4: Cancel the action
def cancel_transfer_node(state: AgentState) -> dict:
    return {
        "messages": [AIMessage(content="Transaction was rejected and cancelled by the human reviewer.")],
        "pending_transfer": None
    }

# 3. Assemble the Graph
builder = StateGraph(AgentState)

builder.add_node("parse_request", parse_request_node)
builder.add_node("human_approval", human_approval_node)
builder.add_node("execute_transfer_node", execute_transfer_node)
builder.add_node("cancel_transfer_node", cancel_transfer_node)

builder.add_edge(START, "parse_request")

# Conditional routing from parser
def route_after_parse(state: AgentState):
    if state.get("pending_transfer"):
        return "human_approval"
    return END

builder.add_conditional_edges("parse_request", route_after_parse, ["human_approval", END])

# Compile graph with persistence checkpointer
checkpointer = MemorySaver() # In production, use PostgresSaver
agent_app = builder.compile(checkpointer=checkpointer)

Running and Resuming the Agent

To test the human in the loop flow, you run the agent with a unique thread_id. When it pauses, you inspect the state, get input from your UI or CLI, and resume using Command(resume=...).

# Unique session identifier
thread_config = {"configurable": {"thread_id": "session-101"}}

print("--- Step 1: User sends initial request ---")
initial_input = {"messages": [HumanMessage(content="Please transfer $5000 to Vendor Corp")]}

# Run until the interrupt is reached
events = agent_app.invoke(initial_input, config=thread_config)

# Check state at interrupt
current_state = agent_app.get_state(thread_config)
print(f"Graph status: Next node is {current_state.next}")
print(f"Interrupt payload: {current_state.tasks[0].interrupts[0].value}")

print("\n--- Step 2: Human approves the request ---")
# Resume execution with approval payload
resume_command = Command(resume={"approved": True, "notes": "Approved by Finance Lead"})
final_events = agent_app.invoke(resume_command, config=thread_config)

for msg in final_events["messages"]:
    print(f"{msg.type.upper()}: {msg.content}")

Production Best Practices

1. Use PostgresSaver in Production: In memory checkpointers lose state when the container restarts. Use langgraph-checkpoint-postgres so approval workflows can wait days across distributed pods.
2. Audit Logs for All Human Decisions: Log the exact user identity, timestamp, and reason whenever an approval or rejection happens.
3. Set Expiration Timeouts: If a human does not review a pending task within a certain timeframe (like 24 hours), automatically fail or cancel the operation to prevent stale state execution.