Human-in-the-Loop AI: 4 LangGraph Patterns That Keep Humans in Control

Why Fully Autonomous Agents Fail in Production

The failure mode of fully autonomous agents is not that they misbehave randomly — it is that they behave correctly according to their training but incorrectly according to business context they were never given. The database agent above was correct: the index optimisation was valid. What it lacked was awareness of the maintenance window policy, the business calendar, and the unwritten rule that production database changes require a DBA sign-off.

No amount of prompt engineering encodes that organisational context reliably. Business policies change. Exceptional circumstances arise. An agent running at 3 AM cannot ask a follow-up question. Human-in-the-loop patterns solve this by acknowledging a fundamental truth: AI agents are excellent at identifying what to do; humans are still essential for deciding when and whether to do it in specific context.

The data supports this. Enterprise teams that implemented HITL patterns for high-risk actions saw a 91% reduction in agent-caused incidents compared to fully autonomous deployments. The overhead cost? Roughly 8–15% of total agent actions require human review — the other 85–92% proceed autonomously, at full speed.

Pattern 1: Approval Gate

The approval gate is the simplest HITL pattern: the agent builds a plan and then pauses before executing any destructive, irreversible, or high-impact action. A human reviews the plan, approves or rejects it, and the agent either proceeds or backtracks.

In LangGraph, this is implemented using interrupt_before on the node that executes the dangerous action. The graph compiles with the interrupt configured, and when execution reaches that node, it raises an NodeInterrupt exception — which LangGraph catches, persists the state, and surfaces to the caller.

from typing import TypedDict, Optional
from langgraph.graph import StateGraph, END
from langgraph.checkpoint.memory import MemorySaver

class DeploymentState(TypedDict):
    task: str
    deployment_plan: Optional[str]
    human_approved: bool
    deployment_result: Optional[str]

def plan_deployment(state: DeploymentState) -> dict:
    """Agent builds the deployment plan — safe, read-only analysis."""
    plan = f"""
    Deployment Plan for: {state['task']}
    1. Run database migration scripts (estimated: 4 min table lock)
    2. Deploy new container image to production namespace
    3. Update Kubernetes service selector
    4. Validate health checks pass (30s window)
    
    RISK ASSESSMENT: Medium — table lock during migration.
    RECOMMENDED WINDOW: Off-peak (02:00–04:00 IST)
    """
    return {"deployment_plan": plan, "human_approved": False}

def execute_deployment(state: DeploymentState) -> dict:
    """
    Executes the deployment. This node is gated behind interrupt_before.
    WHY: deployment is irreversible once migration runs — human must confirm.
    """
    if not state["human_approved"]:
        # This should never be reached due to interrupt, but defensive check
        raise ValueError("Deployment attempted without human approval")

    # ... actual deployment logic here
    return {"deployment_result": "Deployment completed successfully at 02:15 IST"}

# Build the graph
builder = StateGraph(DeploymentState)
builder.add_node("plan", plan_deployment)
builder.add_node("execute", execute_deployment)
builder.add_edge("__start__", "plan")
builder.add_edge("plan", "execute")
builder.add_edge("execute", END)

# CheckpointSaver persists state across the interruption
# Use PostgresSaver in production for durability across restarts
checkpointer = MemorySaver()

# interrupt_before="execute" pauses graph BEFORE the execute node runs
graph = builder.compile(checkpointer=checkpointer, interrupt_before=["execute"])

# --- PHASE 1: Run until interrupt ---
config = {"configurable": {"thread_id": "deploy-001"}}
result = graph.invoke({"task": "Optimise orders table index", "human_approved": False}, config)

print("Agent paused. Review the plan:")
print(graph.get_state(config).values["deployment_plan"])

# --- PHASE 2: Human approves and resumes ---
# Human reviews the plan, decides to schedule for off-peak window
# Update state with approval and preferred timing
graph.update_state(config, {"human_approved": True})

# Resume from checkpoint — execute node now runs with approved=True
final_result = graph.invoke(None, config)
print(final_result["deployment_result"])

Pattern 2: Interrupt-and-Resume with State Editing

The approval gate either approves or rejects a pre-built plan. The interrupt-and-resume pattern allows the human to modify the agent's state during the interruption — correcting errors, adding context, or redirecting the plan before the agent continues.

This is the most powerful HITL pattern for complex workflows. The agent does the hard reasoning work; the human does the contextual correction; the agent resumes with the corrected state and continues.

from typing import TypedDict, Optional, Annotated
import operator
from langgraph.graph import StateGraph, END
from langgraph.checkpoint.memory import MemorySaver

class EmailDraftState(TypedDict):
    recipient: str
    subject: str
    draft_body: str
    human_edits: Optional[str]         # populated by human during interrupt
    send_approved: bool
    sent_message_id: Optional[str]

def draft_email(state: EmailDraftState) -> dict:
    """Agent drafts email based on context — fast, automated."""
    draft = f"""
Dear {state['recipient']},

Following our discussion on enterprise Kubernetes training, I wanted to share
our Agentic AI workshop details. Rated 4.91/5.0 at Oracle, this 5-day
programme has delivered measurable ROI for teams at [CLIENT_NAME].

Would you be available for a 20-minute call next week?

Best regards,
Rajesh Gheware
    """
    return {"draft_body": draft, "send_approved": False}

def send_email(state: EmailDraftState) -> dict:
    """
    Sends the email. Gated behind interrupt_before.
    WHY: email is irreversible — once sent, cannot be unsent.
    The human interrupt allows editing recipient, subject, and body.
    """
    # In production: replace with actual SES send
    final_body = state.get("human_edits") or state["draft_body"]
    # send_email_via_ses(state["recipient"], state["subject"], final_body)
    return {"sent_message_id": "msg-001", "send_approved": True}

builder = StateGraph(EmailDraftState)
builder.add_node("draft", draft_email)
builder.add_node("send", send_email)
builder.add_edge("__start__", "draft")
builder.add_edge("draft", "send")
builder.add_edge("send", END)

checkpointer = MemorySaver()
graph = builder.compile(checkpointer=checkpointer, interrupt_before=["send"])

# Phase 1: Draft
config = {"configurable": {"thread_id": "email-rajiv-001"}}
graph.invoke({
    "recipient": "Rajiv Kumar, L&D Head",
    "subject": "Agentic AI Workshop — Oracle 4.91/5.0",
    "draft_body": "", "human_edits": None, "send_approved": False, "sent_message_id": None
}, config)

# Phase 2: Human reviews draft, edits personalisation
current_state = graph.get_state(config).values
print(f"Draft ready for review:\n{current_state['draft_body']}")

# Human updates the state with personalised edits
graph.update_state(config, {
    "human_edits": current_state["draft_body"].replace(
        "[CLIENT_NAME]", "Deutsche Bank and Standard Chartered"
    )
})

# Phase 3: Resume — send node runs with edited body
graph.invoke(None, config)
print(f"Email sent: {graph.get_state(config).values['sent_message_id']}")

Pattern 3: Confidence Threshold Escalation

Not every action needs human approval. The efficiency of HITL comes from routing selectively: only escalate when the agent's confidence in its decision falls below a threshold. High-confidence, well-understood actions proceed autonomously. Low-confidence or ambiguous decisions pause for human review.

In LangGraph, this is implemented as a conditional edge that checks a confidence score in the agent state and routes to either the action node (proceed) or an escalation node (pause for human):

from typing import TypedDict, Optional, Literal
from langgraph.graph import StateGraph, END
from langgraph.checkpoint.memory import MemorySaver

CONFIDENCE_THRESHOLD = 0.75  # Below this → escalate to human

class IncidentState(TypedDict):
    alert: str
    diagnosis: Optional[str]
    confidence: float           # 0.0 to 1.0
    remediation_plan: Optional[str]
    human_override: Optional[str]  # human can provide alternative plan
    resolved: bool

def diagnose_incident(state: IncidentState) -> dict:
    """Agent analyses the alert and produces a diagnosis + confidence score."""
    # In production: this calls the LLM with alert context
    # Simplified example with mocked confidence
    if "OOMKilled" in state["alert"]:
        return {
            "diagnosis": "Pod OOMKilled — memory limit too low for current workload",
            "confidence": 0.92,  # High confidence — well-known pattern
            "remediation_plan": "Increase memory limit from 512Mi to 1Gi"
        }
    else:
        return {
            "diagnosis": "Unknown failure pattern — network latency spike detected",
            "confidence": 0.61,  # Low confidence — novel pattern
            "remediation_plan": "Investigate network policy and pod-to-pod latency"
        }

def should_escalate(state: IncidentState) -> Literal["auto_remediate", "escalate_human"]:
    """
    Conditional edge: route based on agent confidence.
    WHY 0.75 threshold: below this, false positive rate rises above 15%
    — humans catch errors that save more time than the escalation costs.
    """
    if state["confidence"] >= CONFIDENCE_THRESHOLD:
        return "auto_remediate"
    return "escalate_human"

def auto_remediate(state: IncidentState) -> dict:
    """Executes remediation plan autonomously — only for high-confidence diagnoses."""
    plan = state.get("human_override") or state["remediation_plan"]
    print(f"AUTO-REMEDIATING: {plan}")
    # ... kubectl apply or Helm upgrade here
    return {"resolved": True}

def escalate_to_human(state: IncidentState) -> dict:
    """
    Low-confidence path: surfaces diagnosis to on-call engineer.
    Graph pauses here via interrupt_before in the escalation node.
    """
    print(f"ESCALATING TO HUMAN (confidence={state['confidence']:.0%})")
    print(f"Agent diagnosis: {state['diagnosis']}")
    print(f"Suggested plan: {state['remediation_plan']}")
    # In production: send PagerDuty alert with state context
    return {}

builder = StateGraph(IncidentState)
builder.add_node("diagnose", diagnose_incident)
builder.add_node("auto_remediate", auto_remediate)
builder.add_node("escalate_human", escalate_to_human)

builder.add_edge("__start__", "diagnose")
builder.add_conditional_edges("diagnose", should_escalate)
builder.add_edge("auto_remediate", END)

# Pause BEFORE executing escalate_human — human reviews + provides override
builder.add_edge("escalate_human", END)

checkpointer = MemorySaver()
graph = builder.compile(checkpointer=checkpointer, interrupt_before=["escalate_human"])

# High-confidence incident — auto-remediated without human
config1 = {"configurable": {"thread_id": "incident-001"}}
graph.invoke({"alert": "OOMKilled: payment-service pod", "confidence": 0.0, "resolved": False, "diagnosis": None, "remediation_plan": None, "human_override": None}, config1)

# Low-confidence incident — escalated to human
config2 = {"configurable": {"thread_id": "incident-002"}}
graph.invoke({"alert": "Latency spike detected on auth-service", "confidence": 0.0, "resolved": False, "diagnosis": None, "remediation_plan": None, "human_override": None}, config2)

# Human provides override plan and resumes
graph.update_state(config2, {"human_override": "Restart auth-service pods, check Redis connection pool"})
graph.invoke(None, config2)

Pattern 4: Async Review Loop

The previous three patterns are synchronous — the agent pauses and waits. The async review loop is different: the agent acts first, the human reviews asynchronously, and the agent learns from the feedback. This pattern is ideal for high-volume, low-risk actions where reviewing before acting would create unacceptable latency.

Classic use cases: social media posts reviewed after drafting but before publishing, code suggestions reviewed after generation but before commit, email drafts queued for human review in batch.

from typing import TypedDict, Optional, Annotated
import operator
from langgraph.graph import StateGraph, END
from langgraph.checkpoint.memory import MemorySaver
from datetime import datetime

class ContentState(TypedDict):
    topic: str
    draft_content: Optional[str]
    review_status: str   # "pending" | "approved" | "rejected" | "edited"
    reviewer_feedback: Optional[str]
    published_url: Optional[str]
    feedback_incorporated: Annotated[list[str], operator.add]  # running learning log

def generate_content(state: ContentState) -> dict:
    """Agent drafts the content — runs immediately, no waiting."""
    draft = f"""
# {state['topic']}

[Full blog post content generated by the agent based on current trends,
workshop curriculum, and Rajesh Gheware's practitioner voice...]

4 min read | Agentic AI | gheWARE Insights
    """
    return {
        "draft_content": draft,
        "review_status": "pending"
    }

def queue_for_review(state: ContentState) -> dict:
    """
    Queues content for async human review.
    WHY async: content creation is high-volume (4 posts/day).
    Blocking on human approval would serialise the content pipeline.
    Instead: create draft → queue → human reviews in batch → agent publishes approved.
    """
    # In production: insert into review queue (database, Notion, Slack message)
    print(f"[{datetime.now().strftime('%H:%M')}] Queued for review: {state['topic']}")
    print("Review URL: https://cms.gheware.com/review/pending")
    # Graph pauses here — resumes when human_approves() is called
    return {}

def publish_content(state: ContentState) -> dict:
    """Publishes approved content."""
    if state["review_status"] != "approved":
        return {"published_url": None}
    # ... actual GitHub Pages publish
    return {"published_url": f"https://devops.gheware.com/blog/posts/{state['topic'].lower().replace(' ', '-')}-2026.html"}

def incorporate_feedback(state: ContentState) -> dict:
    """
    Learns from reviewer feedback for future content generation.
    WHY this node: accumulating feedback in state creates a living style guide
    that improves subsequent drafts without full re-training.
    """
    if state.get("reviewer_feedback"):
        return {"feedback_incorporated": [state["reviewer_feedback"]]}
    return {}

builder = StateGraph(ContentState)
builder.add_node("generate", generate_content)
builder.add_node("review_queue", queue_for_review)
builder.add_node("publish", publish_content)
builder.add_node("learn", incorporate_feedback)

builder.add_edge("__start__", "generate")
builder.add_edge("generate", "review_queue")
builder.add_edge("review_queue", "publish")  # resumes here after approval
builder.add_edge("publish", "learn")
builder.add_edge("learn", END)

checkpointer = MemorySaver()
# interrupt_after review_queue: human reviews, then workflow continues to publish
graph = builder.compile(checkpointer=checkpointer, interrupt_after=["review_queue"])

config = {"configurable": {"thread_id": "content-001"}}
graph.invoke({"topic": "LangGraph StateGraph for Beginners", "draft_content": None,
              "review_status": "pending", "reviewer_feedback": None,
              "published_url": None, "feedback_incorporated": []}, config)

# --- Hours later: human reviews via CMS and approves ---
graph.update_state(config, {
    "review_status": "approved",
    "reviewer_feedback": "Good post. Add more code examples in section 3. Use 'production' not 'prod' in headings."
})
graph.invoke(None, config)
print(f"Published: {graph.get_state(config).values['published_url']}")

When to Use Each HITL Pattern

Conclusion: HITL Is Architecture, Not Compromise

The organisations that deploy AI agents most successfully in 2026 are not the ones that went fastest to full autonomy. They are the ones that designed human oversight into the architecture from the first sprint — and then gradually expanded the agent's autonomous operating envelope as trust was established through monitored performance.

Start with approval gates on every destructive action. Add confidence threshold routing once you have 30 days of production data and understand where the agent's accuracy falls off. Move to async review for content and low-risk operations once human reviewers trust the agent's output quality. By month three, 85% of your agent's actions are fully autonomous. By month six, that number is 92%. And every autonomous action is based on demonstrated trust, not optimistic assumption.

In Day 3 of the Agentic AI Workshop, we build these four patterns from scratch in live lab sessions — wiring up LangGraph StateGraphs with real checkpoint persistence, testing interrupt flows, and simulating human approval workflows. Participants leave with production-ready HITL code they understand end to end, not just boilerplate they copy-pasted from a tutorial.