Langgraph Human In Loop

LangGraph Human-in-the-Loop

Pause workflows for human intervention and approval.

Basic Interrupt

workflow = StateGraph(State)
workflow.add_node("draft", generate_draft)
workflow.add_node("review", human_review)
workflow.add_node("publish", publish_content)

# Interrupt before review
app = workflow.compile(interrupt_before=["review"])

# Step 1: Generate draft (stops at review)
config = {"configurable": {"thread_id": "doc-123"}}
result = app.invoke({"topic": "AI"}, config=config)
# Workflow pauses here

Dynamic interrupt() Function (2026 Best Practice)

Modern approach using interrupt() within node logic:

from langgraph.types import interrupt, Command

def approval_node(state: State):
    """Dynamic interrupt based on conditions."""
    # Only interrupt for high-risk actions
    if state["risk_level"] == "high":
        response = interrupt({
            "question": "High-risk action detected. Approve?",
            "action": state["proposed_action"],
            "risk_level": state["risk_level"],
            "details": state["action_details"]
        })

        if not response.get("approved"):
            return {"status": "rejected", "action": None}

    # Low risk or approved - proceed
    return {"status": "approved", "action": state["proposed_action"]}

Resume After Approval

# Step 2: Human reviews and updates state
state = app.get_state(config)
print(f"Draft: {state.values['draft']}")

# Human decision
state.values["approved"] = True
state.values["feedback"] = "Looks good"
app.update_state(config, state.values)

# Step 3: Resume workflow
result = app.invoke(None, config=config)  # Continues to publish

Command(resume=) Pattern (2026 Best Practice)

from langgraph.types import Command

config = {"configurable": {"thread_id": "workflow-123"}}

# Initial invoke - stops at interrupt
result = graph.invoke(initial_state, config)

# Check for interrupt
if "__interrupt__" in result:
    interrupt_info = result["__interrupt__"][0].value
    print(f"Action: {interrupt_info['action']}")
    print(f"Question: {interrupt_info['question']}")

    # Get user decision
    user_response = {"approved": True, "feedback": "Looks good"}

    # Resume with Command
    final = graph.invoke(Command(resume=user_response), config)

Approval Gate Node

def approval_gate(state: WorkflowState) -> WorkflowState:
    """Check if human approved."""
    if not state.get("human_reviewed"):
        # Will pause here due to interrupt_before
        return state

    if state["approved"]:
        state["next"] = "publish"
    else:
        state["next"] = "revise"

    return state

workflow.add_node("approval_gate", approval_gate)

# Pause before this node
app = workflow.compile(interrupt_before=["approval_gate"])

Feedback Loop Pattern

import uuid

async def run_with_feedback(initial_state: dict):
    """Run until human approves."""
    config = {"configurable": {"thread_id": str(uuid.uuid4())}}

    while True:
        # Run until interrupt
        result = app.invoke(initial_state, config=config)

        # Check for interrupt
        if "__interrupt__" not in result:
            return result  # Completed without interrupt

        interrupt_info = result["__interrupt__"][0].value
        print(f"Output: {interrupt_info.get('output', 'N/A')}")
        feedback = input("Approve? (yes/no/feedback): ")

        if feedback.lower() == "yes":
            return app.invoke(Command(resume={"approved": True}), config=config)
        elif feedback.lower() == "no":
            return {"status": "rejected"}
        else:
            # Incorporate feedback and retry
            initial_state = None
            result = app.invoke(
                Command(resume={"approved": False, "feedback": feedback}),
                config=config
            )

Input Validation Loop

from langgraph.types import interrupt

def get_valid_age(state: State):
    """Repeatedly prompt until valid input."""
    prompt = "What is your age?"

    while True:
        answer = interrupt(prompt)

        # Validate
        if isinstance(answer, int) and 0 < answer < 150:
            return {"age": answer}

        # Invalid - update prompt and continue
        prompt = f"'{answer}' is not valid. Please enter a number between 1 and 150."

API Integration

from fastapi import FastAPI, HTTPException

app = FastAPI()

@app.post("/workflows/{workflow_id}/approve")
async def approve_workflow(workflow_id: str, approved: bool, feedback: str = ""):
    """API endpoint for human approval."""
    config = {"configurable": {"thread_id": workflow_id}}

    try:
        state = langgraph_app.get_state(config)
    except Exception:
        raise HTTPException(404, "Workflow not found")

    # Update state with human decision
    state.values["approved"] = approved
    state.values["feedback"] = feedback
    state.values["human_reviewed"] = True
    langgraph_app.update_state(config, state.values)

    # Resume workflow
    result = langgraph_app.invoke(None, config=config)

    return {"status": "completed", "result": result}

Multiple Approval Points

# Interrupt at multiple points
app = workflow.compile(
    interrupt_before=["first_review", "final_review"]
)

# First review
result = app.invoke(initial_state, config=config)
# ... human approves first review ...
app.update_state(config, {"first_approved": True})

# Continue to second review
result = app.invoke(None, config=config)
# ... human approves final review ...
app.update_state(config, {"final_approved": True})

# Complete workflow
result = app.invoke(None, config=config)

Key Decisions

Critical Rules

DO:

• Place side effects AFTER interrupt calls
• Make pre-interrupt side effects idempotent (upsert vs create)
• Keep interrupt call order consistent across executions
• Pass simple, JSON-serializable values to interrupt()

DON'T:

• Wrap interrupt in bare try/except (catches the interrupt exception)
• Conditionally skip interrupt calls (breaks determinism)
• Pass functions or class instances to interrupt()
• Create non-idempotent records before interrupts (duplicates on resume)

Common Mistakes

• No timeout (workflows hang forever)
• No notification (humans don't know to review)
• Losing checkpoint (can't resume)
• No reject path (only approve works)
• Wrapping interrupt() in try/except (breaks the mechanism)
• Non-deterministic interrupt call order (breaks resumption)

Evaluations

See references/evaluations.md for test cases.

Related Skills

• langgraph-checkpoints - Persist state across human review pauses
• langgraph-routing - Route based on approval/rejection decisions
• langgraph-tools - Add approval gates before dangerous tool execution
• langgraph-supervisor - Human approval in supervisor routing
• langgraph-streaming - Stream status while waiting for human input
• api-design-framework - Design review API endpoints

Capability Details

interrupt-before

Keywords: interrupt, pause, stop, before, gate Solves:

• How do I pause a workflow for approval?
• Add human review before a step
• Interrupt workflow execution

resume-workflow

Keywords: resume, continue, approve, proceed, update_state Solves:

• How do I resume after human approval?
• Continue workflow after review
• Update state and proceed

approval-patterns

Keywords: approval, approve, reject, decision, gate Solves:

• How do I implement approval workflows?
• Add approval gate to pipeline
• Handle approve/reject decisions

feedback-integration

Keywords: feedback, comment, review, notes, human input Solves:

• How do I collect human feedback?
• Integrate reviewer comments
• Capture feedback in workflow state

interactive-supervision

Keywords: supervise, monitor, interactive, control, override Solves:

• How do I supervise agent execution?
• Add human oversight to agents
• Override agent decisions

state-inspection

Keywords: get_state, inspect, view, current state, debug Solves:

• How do I inspect workflow state?
• View current state at interrupt
• Debug paused workflows

Human-in-the-Loop Checklist

Design

• [ ] Identify approval points
• [ ] Define what requires human input
• [ ] Plan timeout handling
• [ ] Design approval UI/interface

Implementation

• [ ] Use interrupt_before or interrupt_after
• [ ] Save state before interrupt
• [ ] Resume from checkpoint
• [ ] Handle approval/rejection

Interruption Points

graph.add_node("sensitive_action", action_node)
graph.add_edge("analysis", "sensitive_action")  # Interrupt here

# In config
config = {"interrupt_before": ["sensitive_action"]}

User Experience

• [ ] Clear prompt for human
• [ ] Show relevant context
• [ ] Provide approve/reject options
• [ ] Allow modification if needed

Timeout Handling

• [ ] Set reasonable timeout
• [ ] Notify on timeout approach
• [ ] Define default action
• [ ] Log abandoned workflows

Testing

• [ ] Test approval flow
• [ ] Test rejection flow
• [ ] Test timeout behavior
• [ ] Test resume from checkpoint

API Integration for Human Review

Expose REST endpoints for human-in-the-loop workflows.

Implementation

from fastapi import FastAPI, HTTPException
from pydantic import BaseModel

app = FastAPI()

class ApprovalRequest(BaseModel):
    approved: bool
    feedback: str = ""
    reviewer_id: str

class WorkflowStatus(BaseModel):
    workflow_id: str
    status: str
    current_node: str
    awaiting_review: bool

@app.get("/workflows/{workflow_id}/status")
async def get_workflow_status(workflow_id: str) -> WorkflowStatus:
    """Get current workflow state for review UI."""
    config = {"configurable": {"thread_id": workflow_id}}
    try:
        state = langgraph_app.get_state(config)
        return WorkflowStatus(
            workflow_id=workflow_id,
            status="pending_review" if state.next else "completed",
            current_node=state.next[0] if state.next else "end",
            awaiting_review=state.values.get("awaiting_review", False)
        )
    except Exception:
        raise HTTPException(404, "Workflow not found")

@app.post("/workflows/{workflow_id}/approve")
async def approve_workflow(workflow_id: str, request: ApprovalRequest):
    """Submit human approval decision."""
    config = {"configurable": {"thread_id": workflow_id}}
    state = langgraph_app.get_state(config)

    state.values.update({
        "approved": request.approved,
        "feedback": request.feedback,
        "human_reviewed": True,
        "reviewer_id": request.reviewer_id
    })
    langgraph_app.update_state(config, state.values)

    result = await langgraph_app.ainvoke(None, config=config)
    return {"status": "completed", "result": result}

When to Use

• Web-based review interfaces
• Mobile approval workflows
• Integration with external systems
• Async human review processes

Anti-patterns

• No authentication on approval endpoints
• Missing audit logging
• No idempotency for approval calls
• Blocking API calls (use background tasks)

Approval Gate Pattern

Route workflow based on human approval decisions.

Implementation

from langgraph.graph import StateGraph, END

def approval_gate(state: WorkflowState) -> dict:
    """Check human approval status and route accordingly."""
    if not state.get("human_reviewed"):
        # State will be updated by human via API
        return {"awaiting_review": True}

    if state["approved"]:
        return {"next": "publish", "awaiting_review": False}
    elif state.get("feedback"):
        return {"next": "revise", "awaiting_review": False}
    else:
        return {"next": END, "status": "rejected"}

workflow = StateGraph(WorkflowState)
workflow.add_node("generate", generate_node)
workflow.add_node("approval_gate", approval_gate)
workflow.add_node("revise", revise_node)
workflow.add_node("publish", publish_node)

workflow.add_edge("generate", "approval_gate")
workflow.add_edge("revise", "approval_gate")  # Re-review after revision

workflow.add_conditional_edges(
    "approval_gate",
    lambda s: s.get("next", "approval_gate"),
    {"publish": "publish", "revise": "revise", END: END}
)

app = workflow.compile(interrupt_before=["approval_gate"])

When to Use

• Approve/reject/revise workflows
• Multi-stage approval processes
• Iterative refinement with feedback
• Gated publishing pipelines

Anti-patterns

• No revise path (only approve/reject)
• No max revision limit (infinite loops)
• Approval without viewing content
• Missing audit trail of decisions

Evaluation Test Cases

Test 1: Approval Gate

{
  "skills": ["langgraph-human-in-loop"],
  "query": "Add human approval before executing a dangerous operation",
  "expected_behavior": [
    "Uses interrupt() function from langgraph.types",
    "Passes approval request data to interrupt()",
    "Resumes with Command(resume=response)",
    "Handles approval and rejection cases"
  ]
}

Test 2: Edit Before Continue

{
  "skills": ["langgraph-human-in-loop"],
  "query": "Let human edit AI-generated content before proceeding",
  "expected_behavior": [
    "Shows generated content in interrupt payload",
    "Human can modify or approve",
    "Resume uses human-edited version",
    "Original preserved if human approves as-is"
  ]
}

Test 3: Validation Loop

{
  "skills": ["langgraph-human-in-loop"],
  "query": "Implement review loop until human approves quality",
  "expected_behavior": [
    "Tracks review_count in state",
    "interrupt() on each review cycle",
    "Routes back to generation if rejected",
    "Proceeds to next step when approved"
  ]
}

Feedback Loop Pattern

Iterate with human feedback until approval.

Implementation

import uuid_utils

async def run_with_feedback_loop(
    app,
    initial_state: dict,
    max_iterations: int = 5
) -> dict:
    """Run workflow with iterative human feedback."""
    config = {"configurable": {"thread_id": str(uuid_utils.uuid7())}}

    for iteration in range(max_iterations):
        # Run until interrupt
        result = app.invoke(
            initial_state if iteration == 0 else None,
            config=config
        )

        # Get state for review
        state = app.get_state(config)
        print(f"\n--- Iteration {iteration + 1} ---")
        print(f"Output: {state.values.get('output', 'N/A')}")

        # Collect human feedback
        feedback = input("Approve? (yes/no/[feedback]): ").strip()

        if feedback.lower() == "yes":
            state.values["approved"] = True
            app.update_state(config, state.values)
            return app.invoke(None, config=config)

        if feedback.lower() == "no":
            return {"status": "rejected", "iteration": iteration + 1}

        # Incorporate feedback and retry
        state.values["feedback"] = feedback
        state.values["iteration"] = iteration + 1
        app.update_state(config, state.values)

    return {"status": "max_iterations_reached", "final_state": state.values}

When to Use

• Creative content refinement
• Iterative document editing
• AI-assisted writing workflows
• Quality improvement loops

Anti-patterns

• No max iteration limit
• Ignoring previous feedback in prompts
• No progress indication
• Lost context between iterations

Interrupt and Resume Pattern

Pause workflow execution for human intervention.

Implementation

from langgraph.graph import StateGraph

workflow = StateGraph(WorkflowState)
workflow.add_node("generate", generate_content)
workflow.add_node("review", human_review_node)
workflow.add_node("publish", publish_content)

# Compile with interrupt point
app = workflow.compile(interrupt_before=["review"])

# Step 1: Run until interrupt
config = {"configurable": {"thread_id": "doc-123"}}
result = app.invoke({"topic": "AI Safety"}, config=config)
# Workflow pauses at 'review' node

# Step 2: Get current state for human review
state = app.get_state(config)
print(f"Content to review: {state.values['draft']}")

# Step 3: Human updates state
state.values["approved"] = True
state.values["feedback"] = "Looks good, minor typo on line 3"
app.update_state(config, state.values)

# Step 4: Resume workflow
final_result = app.invoke(None, config=config)

When to Use

• Content approval before publishing
• High-stakes decisions requiring oversight
• Quality gates in production pipelines
• Compliance review requirements

Anti-patterns

• No checkpointer configured (cannot resume)
• Forgetting to call update_state before resume
• No timeout for abandoned reviews
• Missing notification to reviewers