Build a LangChain Agent with Human-in-the-Loop Approval

Every time I demo an autonomous agent to someone non-technical, the first question is always the same: "But what stops it from doing something wrong?"

It's a fair question. An agent that can send emails, delete files, call APIs, and make purchases is genuinely powerful — and genuinely dangerous if it misunderstands the intent. Human-in-the-loop (HITL) is the pattern that answers that question with actual code instead of vibes.

The idea is simple: before the agent executes a high-stakes action, it pauses and asks a human to approve or reject. Only after approval does it proceed. The implementation is more interesting.

This guide covers the LangGraph-based approach (the right way), a simpler manual approach for lightweight cases, and patterns for different approval delivery methods.

Why HITL Matters More Than You Think

There's a tendency to treat human oversight as a temporary measure — something you add during testing and remove when the agent gets "good enough." I'd push back on this.

Some actions should always require human approval regardless of agent accuracy:

Sending emails to external parties
Any financial transaction
Deleting records from a database
Publishing content publicly
Calling external APIs that have side effects

The cost of approval friction is low. The cost of an agent accidentally emailing the wrong person or deleting the wrong data is not.

Setting Up LangGraph with Checkpointing

HITL requires persistent state — the agent needs to pause and resume without losing its progress. LangGraph's checkpointer handles this.

pip install langgraph langchain-openai

from langgraph.graph import StateGraph, END
from langgraph.checkpoint.sqlite import SqliteSaver
from langchain_openai import ChatOpenAI
from langchain_core.messages import HumanMessage, AIMessage
from langchain_core.tools import tool
from typing import TypedDict, Annotated, List
import operator

# Define state
class AgentState(TypedDict):
    messages: Annotated[List, operator.add]
    pending_action: str | None
    approved: bool | None

# Tools — some are safe, some need approval
@tool
def search_web(query: str) -> str:
    """Search for information on the web."""
    return f"Search results for: {query}"

@tool
def send_email(to: str, subject: str, body: str) -> str:
    """Send an email. REQUIRES HUMAN APPROVAL before execution."""
    return f"Email sent to {to} with subject '{subject}'"

@tool
def delete_record(record_id: str) -> str:
    """Delete a database record. REQUIRES HUMAN APPROVAL before execution."""
    return f"Record {record_id} deleted"

# Mark which tools need approval
TOOLS_REQUIRING_APPROVAL = {"send_email", "delete_record"}

llm = ChatOpenAI(model="gpt-4o-mini", temperature=0)
tools = [search_web, send_email, delete_record]
llm_with_tools = llm.bind_tools(tools)
tool_map = {t.name: t for t in tools}

Building the HITL Graph

from langgraph.graph import StateGraph, END
from langgraph.prebuilt import ToolNode
from langchain_core.messages import ToolMessage

def should_continue(state: AgentState) -> str:
    last_message = state["messages"][-1]
    if not hasattr(last_message, "tool_calls") or not last_message.tool_calls:
        return "end"

    # Check if any tool calls need approval
    for tc in last_message.tool_calls:
        if tc["name"] in TOOLS_REQUIRING_APPROVAL:
            return "needs_approval"
    return "execute_tools"

def call_model(state: AgentState) -> AgentState:
    response = llm_with_tools.invoke(state["messages"])
    return {"messages": [response]}

def execute_approved_tools(state: AgentState) -> AgentState:
    last_message = state["messages"][-1]
    tool_results = []
    for tc in last_message.tool_calls:
        result = tool_map[tc["name"]].invoke(tc["args"])
        tool_results.append(
            ToolMessage(content=str(result), tool_call_id=tc["id"])
        )
    return {"messages": tool_results}

def request_approval(state: AgentState) -> AgentState:
    last_message = state["messages"][-1]
    actions_summary = []
    for tc in last_message.tool_calls:
        if tc["name"] in TOOLS_REQUIRING_APPROVAL:
            actions_summary.append(f"- {tc['name']}({tc['args']})")
    pending = "\n".join(actions_summary)
    print(f"\n⚠️  APPROVAL REQUIRED:\n{pending}\n")
    return {"pending_action": pending}

# Build the graph
workflow = StateGraph(AgentState)
workflow.add_node("agent", call_model)
workflow.add_node("execute_tools", execute_approved_tools)
workflow.add_node("request_approval", request_approval)

workflow.set_entry_point("agent")
workflow.add_conditional_edges(
    "agent",
    should_continue,
    {
        "execute_tools": "execute_tools",
        "needs_approval": "request_approval",
        "end": END,
    }
)
workflow.add_edge("execute_tools", "agent")
# After approval, execution resumes at execute_tools
workflow.add_edge("request_approval", "execute_tools")

# Compile with SQLite checkpointer for state persistence
checkpointer = SqliteSaver.from_conn_string(":memory:")
graph = workflow.compile(
    checkpointer=checkpointer,
    interrupt_before=["execute_tools"]  # Always pause before tool execution
)

Using interrupt_before for Proper Pause/Resume

The cleaner LangGraph approach uses interrupt_before to halt execution at specific nodes:

from langgraph.graph import StateGraph, END
from langgraph.checkpoint.sqlite import SqliteSaver

# Same nodes as before, but compiled differently
graph = workflow.compile(
    checkpointer=checkpointer,
    interrupt_before=["request_approval"]  # Pause before the approval node
)

thread_id = "user-session-123"
config = {"configurable": {"thread_id": thread_id}}

# First run — will pause at request_approval
events = graph.stream(
    {"messages": [HumanMessage("Send an email to alice@example.com saying the project is done")]},
    config,
    stream_mode="values"
)

for event in events:
    if event.get("messages"):
        print(event["messages"][-1].content[:200])

# Check current state
snapshot = graph.get_state(config)
print("Graph interrupted:", snapshot.next)  # ('request_approval',)
print("Pending tool calls:", snapshot.values["messages"][-1].tool_calls)

Implementing the Approval Interface

CLI Approval

def run_with_cli_approval(user_input: str, thread_id: str) -> str:
    config = {"configurable": {"thread_id": thread_id}}

    # Initial run
    for event in graph.stream(
        {"messages": [HumanMessage(user_input)]},
        config,
        stream_mode="values"
    ):
        pass

    # Check if paused
    snapshot = graph.get_state(config)
    if not snapshot.next:
        return snapshot.values["messages"][-1].content

    # Show what needs approval
    last_msg = snapshot.values["messages"][-1]
    print("\n⚠️  Agent wants to execute:")
    for tc in last_msg.tool_calls:
        print(f"  {tc['name']}({tc['args']})")

    approval = input("\nApprove? (yes/no): ").strip().lower()

    if approval == "yes":
        # Resume execution
        for event in graph.stream(None, config, stream_mode="values"):
            pass
        return graph.get_state(config).values["messages"][-1].content
    else:
        return "Action cancelled by user."

Web/API Approval

For async systems, you pause the agent and return a task ID to the client. The client polls or waits for a webhook, approves or rejects, and you resume:

from fastapi import FastAPI
from pydantic import BaseModel

app = FastAPI()
pending_approvals: dict = {}

class ApprovalRequest(BaseModel):
    task_id: str
    approved: bool

@app.post("/agent/run")
async def run_agent(user_input: str, user_id: str):
    thread_id = f"thread-{user_id}"
    config = {"configurable": {"thread_id": thread_id}}

    # Run until interrupted
    async for event in graph.astream(
        {"messages": [HumanMessage(user_input)]},
        config,
        stream_mode="values"
    ):
        pass

    snapshot = graph.get_state(config)
    if snapshot.next:
        # Needs approval
        last_msg = snapshot.values["messages"][-1]
        task_id = f"approval-{thread_id}"
        pending_approvals[task_id] = thread_id
        return {
            "status": "pending_approval",
            "task_id": task_id,
            "actions": [{"tool": tc["name"], "args": tc["args"]} for tc in last_msg.tool_calls]
        }

    return {
        "status": "completed",
        "response": snapshot.values["messages"][-1].content
    }

@app.post("/agent/approve")
async def approve_action(req: ApprovalRequest):
    thread_id = pending_approvals.get(req.task_id)
    if not thread_id:
        return {"error": "Task not found"}

    config = {"configurable": {"thread_id": thread_id}}

    if req.approved:
        async for event in graph.astream(None, config, stream_mode="values"):
            pass
        final_state = graph.get_state(config)
        del pending_approvals[req.task_id]
        return {"status": "completed", "response": final_state.values["messages"][-1].content}
    else:
        del pending_approvals[req.task_id]
        return {"status": "cancelled"}

Approval via Slack

Many teams already use Slack for notifications — it's natural to send approval requests there:

from slack_sdk import WebClient

slack = WebClient(token=os.environ["SLACK_BOT_TOKEN"])

def send_slack_approval(channel: str, task_id: str, actions: list) -> None:
    action_text = "\n".join([f"• `{a['tool']}` with args: `{a['args']}`" for a in actions])
    slack.chat_postMessage(
        channel=channel,
        text=f"⚠️ Agent approval needed (task: {task_id}):\n{action_text}",
        blocks=[
            {
                "type": "section",
                "text": {"type": "mrkdwn", "text": f"*Agent wants to execute:*\n{action_text}"}
            },
            {
                "type": "actions",
                "elements": [
                    {
                        "type": "button",
                        "text": {"type": "plain_text", "text": "✅ Approve"},
                        "style": "primary",
                        "action_id": "approve",
                        "value": task_id
                    },
                    {
                        "type": "button",
                        "text": {"type": "plain_text", "text": "❌ Reject"},
                        "style": "danger",
                        "action_id": "reject",
                        "value": task_id
                    }
                ]
            }
        ]
    )

Audit Logging

Every approval decision should be logged:

import json
from datetime import datetime

def log_approval_decision(task_id: str, user_id: str, actions: list, approved: bool, reason: str = ""):
    log_entry = {
        "timestamp": datetime.utcnow().isoformat(),
        "task_id": task_id,
        "user_id": user_id,
        "approved": approved,
        "reason": reason,
        "actions": actions
    }
    with open("approval_audit.jsonl", "a") as f:
        f.write(json.dumps(log_entry) + "\n")
    return log_entry

Comparison: Approval Approaches

Approach	Latency	Setup Complexity	Best For
CLI prompt	Manual	Minutes	Local testing
Web API (polling)	Seconds-minutes	Medium	Web applications
Slack interactive	Minutes	Medium	Team workflows
Email approval	Minutes-hours	Low	Low-urgency tasks
Mobile push	Seconds	High	Time-sensitive ops

For this agent to run reliably, you need a proper checkpointer. SQLite works for single-server deployments; see the Build AI agent with LangChain guide for PostgreSQL checkpointing patterns at scale.

The AI agent memory and planning guide explains how agents track state between turns — the same concepts apply here since HITL is fundamentally a state management problem. And CrewAI tutorial shows another approach to multi-agent oversight where one agent reviews another's actions.

Conclusion

Human-in-the-loop is not a crutch for imperfect agents. It's a deliberate design choice that acknowledges some actions are too consequential for full automation — at least until you have extensive production data on the agent's accuracy.

The three things that make HITL work in practice: use LangGraph's interrupt_before for clean pause/resume semantics, persist state in a proper checkpointer (not in-memory), and always log approval decisions for audit purposes.

Start with CLI approval during development, add Slack or email notifications for staging, and build a proper web interface for production. The approval interface will keep improving as you understand where the agent makes mistakes — and you'll be glad you built it from the start.

For the next step in building production agents, the Deploy AI model to production guide covers hosting, monitoring, and scaling patterns.

Frequently Asked Questions

Why would you add human approval to an AI agent? Autonomous agents can make mistakes — wrong API calls, accidental deletions, incorrect email sends. Human-in-the-loop adds a checkpoint before irreversible actions. This is especially important for agents that interact with external systems (email, databases, payment APIs), manage sensitive data, or operate in regulated industries where audit trails are required.

How does interrupt_before work in LangGraph? LangGraph's interrupt_before pauses graph execution before a specified node runs. The state is persisted in a checkpointer, and execution waits until you call graph.invoke() again on the same thread_id. The agent doesn't timeout or lose state — it literally waits in a suspended state until you resume it.

Can I implement human-in-the-loop without LangGraph? Yes, but it's more work. With plain LangChain you'd intercept the agent's action list before execution, present them for review, then manually call the tools and continue. LangGraph's interrupt mechanism handles the state persistence and resumption cleanly, which is why it's the recommended approach for anything beyond simple demos.