How to Implement Human-in-the-Loop (HITL) Approval Gates in LangGraph with PostgreSQL Checkpointers

Introduction

Somewhere between "cool demo" and "production system," every AI agent hits the same wall: someone eventually asks, "What happens if it does something we didn't want it to do?"

Maybe your agent can send emails. Maybe it can issue refunds. Maybe it can run DELETE statements against a customer database because a support ticket implied it should. In a demo, that's fine — you're watching. In production, at 2 AM, with no one watching, that's a liability.

This is exactly the problem Human-in-the-Loop (HITL) design solves. Instead of trusting an LLM to always make the right call, you build a checkpoint into the workflow — a literal pause — where a human reviews the agent's intended action before it executes. If the action is safe, a human approves it and the agent continues exactly where it left off. If it's not, a human rejects it, and the agent either stops or replans.

LangGraph, the low-level orchestration framework from the LangChain team, was built with this exact pattern in mind. It treats your agent as a graph of nodes and edges, and — critically — it treats the state of that graph as something that can be persisted, interrupted, inspected, edited, and resumed. Pair that with a PostgreSQL-backed checkpointer, and you get something a lot more serious than a toy: a durable, auditable, production-ready approval workflow that survives server restarts, scales across workers, and gives you a full history of every decision an agent almost made.

In this tutorial, we're going to build exactly that. By the end, you'll have a working agent that:

Detects when it's about to call a sensitive tool
Pauses execution using interrupt_before (and the newer dynamic interrupt() primitive)
Persists its full state to PostgreSQL via AsyncPostgresSaver
Waits — indefinitely, if needed — for a human decision
Resumes safely using Command(resume=...) once approval or rejection arrives

Let's get into it.

Why Human-in-the-Loop Matters in Production AI Agents

Before writing any code, it's worth being precise about why this pattern exists, because it shapes every design decision that follows.

Autonomous agents fail in a specific and annoying way: they don't crash, they comply confidently with the wrong interpretation. An agent asked to "clean up inactive user accounts" might reasonably decide that means deleting them, when the actual intent was to flag them. There's no exception thrown, no stack trace — just a quietly wrong action executed with full conviction.

HITL gates address three concrete risks:

Irreversibility — Some actions (sending an email, charging a card, deleting a row) can't be undone. A pause before execution is the last safety net.
Ambiguity — LLMs are probabilistic. When a tool call carries real-world consequences, a deterministic human checkpoint removes the ambiguity entirely.
Auditability — Regulated industries (finance, healthcare, legal) often require a documented human sign-off before certain automated actions can occur. A checkpoint that's persisted to a real database gives you that audit trail for free.

The trick is implementing this without turning your agent into a fragile, single-process script that dies the moment the human doesn't respond within the same request lifecycle. That's where durable checkpointing comes in.

Core Concepts: Checkpointers, Threads, and Interrupts

LangGraph's HITL model rests on three pillars. Understanding them individually makes the code that follows feel almost obvious.

What Is a Checkpointer?

A checkpointer is LangGraph's persistence layer. Every time your graph executes a "super-step" (roughly: after each node runs), the checkpointer saves a snapshot of the graph's state. This includes the full message history, any custom state fields you've defined, and metadata about where execution currently sits in the graph.

LangGraph ships with several checkpointer backends:

MemorySaver — in-process, non-persistent, great for local testing
SqliteSaver / AsyncSqliteSaver — file-based, good for single-machine apps
PostgresSaver / AsyncPostgresSaver — production-grade, multi-worker safe, the focus of this article

For anything running behind a real API — where a human might not respond to an approval request for minutes, hours, or days — you need a checkpointer that outlives the process that created it. PostgreSQL is the natural choice: it's battle-tested, supports concurrent access safely, and most teams already run it.

Threads: The Unit of Conversation State

Every invocation of a LangGraph graph is tied to a thread_id. Think of a thread as a persistent "conversation" or "workflow instance." All checkpoints for a given thread are stored together, keyed by that ID.

This is the mechanism that makes asynchronous human approval possible at all. When an agent pauses, you don't keep an HTTP connection open waiting for a human. You simply stop. The thread's state sits safely in PostgreSQL. Later — from a completely different process, server, or even a different team's admin dashboard — you load that thread_id, inspect the paused state, and resume it.

Interrupts: Pausing Execution Mid-Graph

LangGraph gives you two complementary ways to pause a graph:

Static interrupts — declared at compile time via interrupt_before=[...] or interrupt_after=[...], naming specific nodes.
Dynamic interrupts — triggered at runtime from inside a node by calling interrupt(payload), which halts execution and surfaces payload to whoever is waiting on the graph.

Static interrupts are simple and declarative — perfect when you know in advance that a specific node (like execute_sensitive_tool) always requires approval. Dynamic interrupts are more flexible — you can decide inside the node logic whether this particular call warrants human review (e.g., only pause if the refund amount exceeds $500). We'll build both, but the static interrupt_before approach is our primary pattern since it maps directly to the tutorial's goal.

Setting Up Your Environment

Installing Dependencies

pip install langgraph langgraph-checkpoint-postgres langchain-openai psycopg[binary,pool] python-dotenv

A quick note on versions: langgraph-checkpoint-postgres is a separate package from core langgraph, and it depends on psycopg (v3), not the older psycopg2. Mixing the two is a common source of cryptic connection errors, so double-check your requirements.txt if you're migrating an existing project.

Spinning Up PostgreSQL

If you don't already have a Postgres instance handy, a local Docker container is the fastest path:

docker run --name langgraph-hitl-db \
  -e POSTGRES_USER=langgraph \
  -e POSTGRES_PASSWORD=langgraph \
  -e POSTGRES_DB=agent_checkpoints \
  -p 5432:5432 \
  -d postgres:16

Your connection string will look like this — keep it in an environment variable, never hardcoded:

# .env
DB_URI="postgresql://langgraph:langgraph@localhost:5432/agent_checkpoints?sslmode=disable"

Building the Agent Graph

Let's build something realistic: a support-operations agent that can look up customer accounts freely, but must get human sign-off before deleting one.

Defining State

from typing import Annotated, TypedDict
from langgraph.graph.message import add_messages
from langchain_core.messages import AnyMessage

class AgentState(TypedDict):
    messages: Annotated[list[AnyMessage], add_messages]

add_messages is a reducer — it tells LangGraph to append new messages to the list rather than overwrite it, which is what gives you conversational continuity across interrupts and resumes.

Defining Tools (Including a Sensitive One)

from langchain_core.tools import tool

@tool
def lookup_account(account_id: str) -> str:
    """Look up a customer account by ID. Safe, read-only operation."""
    # In reality: query your customer DB
    return f"Account {account_id}: status=active, plan=pro, balance=$0.00"

@tool
def delete_account(account_id: str) -> str:
    """Permanently delete a customer account. IRREVERSIBLE — requires approval."""
    # In reality: DELETE FROM accounts WHERE id = %s
    return f"Account {account_id} has been permanently deleted."

TOOLS = [lookup_account, delete_account]
SENSITIVE_TOOLS = {"delete_account"}

Marking sensitive tools explicitly, rather than inferring risk from the LLM's own judgment, is the whole point of this pattern — you don't want the model deciding when it needs permission.

Building the Graph Nodes

from langgraph.graph import StateGraph, START, END
from langgraph.prebuilt import ToolNode
from langchain_openai import ChatOpenAI

llm = ChatOpenAI(model="gpt-4.1", temperature=0).bind_tools(TOOLS)

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

def route_after_model(state: AgentState):
    last_message = state["messages"][-1]
    if not getattr(last_message, "tool_calls", None):
        return END
    # If any requested tool call is sensitive, route to the guarded tool node
    called = {tc["name"] for tc in last_message.tool_calls}
    if called & SENSITIVE_TOOLS:
        return "sensitive_tools"
    return "safe_tools"

builder = StateGraph(AgentState)
builder.add_node("agent", call_model)
builder.add_node("safe_tools", ToolNode([lookup_account]))
builder.add_node("sensitive_tools", ToolNode([delete_account]))

builder.add_edge(START, "agent")
builder.add_conditional_edges("agent", route_after_model, {
    "safe_tools": "safe_tools",
    "sensitive_tools": "sensitive_tools",
    END: END,
})
builder.add_edge("safe_tools", "agent")
builder.add_edge("sensitive_tools", "agent")

Note the split between safe_tools and sensitive_tools — two separate ToolNode instances. This gives us a clean, single interrupt target instead of having to inspect tool call names deep inside a shared node.

Configuring the AsyncPostgresSaver

This is the part that turns our graph from a script into a durable system. AsyncPostgresSaver handles connection pooling and schema setup for you, but you do need to run setup() once to create its tables.

import os
from psycopg_pool import AsyncConnectionPool
from langgraph.checkpoint.postgres.aio import AsyncPostgresSaver

DB_URI = os.environ["DB_URI"]

connection_kwargs = {
    "autocommit": True,
    "prepare_threshold": 0,
}

async def build_graph():
    pool = AsyncConnectionPool(
        conninfo=DB_URI,
        max_size=20,
        kwargs=connection_kwargs,
        open=False,
    )
    await pool.open()

    checkpointer = AsyncPostgresSaver(pool)
    await checkpointer.setup()  # idempotent — creates tables on first run only

    graph = builder.compile(
        checkpointer=checkpointer,
        interrupt_before=["sensitive_tools"],
    )
    return graph, pool

A few details worth pausing on:

autocommit=True and prepare_threshold=0 are recommended settings from the LangGraph docs when using a connection pool with AsyncPostgresSaver — they avoid prepared-statement conflicts under concurrent access.
setup() is idempotent. It's safe to call on every application boot; it only creates the checkpoints, checkpoint_blobs, and checkpoint_writes tables if they don't already exist. In a real deployment, you'd typically run this once via a migration step rather than on every process start, but it won't hurt to leave it in for smaller services.
The pool, not a single connection, is what you pass to AsyncPostgresSaver. This is what makes the checkpointer safe to use across concurrent requests in something like a FastAPI app.

Triggering interrupt_before on Sensitive Tool Calls

You've already seen the key line:

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

interrupt_before tells LangGraph: right before entering the named node, stop. Persist the checkpoint. Return control to the caller. The graph doesn't error out — it simply pauses in a well-defined, resumable state.

Let's trigger it:

import asyncio
from langchain_core.messages import HumanMessage

async def run_until_pause():
    graph, pool = await build_graph()
    config = {"configurable": {"thread_id": "support-ticket-4521"}}

    result = await graph.ainvoke(
        {"messages": [HumanMessage(content="Please delete account acct_9981, the customer requested closure.")]},
        config=config,
    )

    state = await graph.aget_state(config)
    print("Next node(s) to run:", state.next)
    # -> ('sensitive_tools',)

    pending_call = state.values["messages"][-1].tool_calls[0]
    print("Pending action:", pending_call["name"], pending_call["args"])
    # -> Pending action: delete_account {'account_id': 'acct_9981'}

    await pool.close()

asyncio.run(run_until_pause())

At this point, execution has genuinely stopped. There's no thread sleeping, no open connection waiting. The entire state of the conversation — including the exact tool call the model wants to make — lives in PostgreSQL under the key support-ticket-4521. You could shut the server down completely and it would still be there tomorrow.

The Modern Approach: Dynamic interrupt() Inside Nodes

interrupt_before is great when every call to a node needs review. But sometimes you want conditional pausing — say, only refunds over $500 need a human, while smaller ones sail through automatically. For that, LangGraph's dynamic interrupt() function (used from inside a node) is the more flexible 2026-standard approach.

from langgraph.types import interrupt

def guarded_delete_node(state: AgentState):
    tool_call = state["messages"][-1].tool_calls[0]
    account_id = tool_call["args"]["account_id"]

    decision = interrupt({
        "action": "delete_account",
        "account_id": account_id,
        "reason": "Irreversible action requires human approval",
    })

    if decision.get("approved"):
        result = delete_account.invoke({"account_id": account_id})
    else:
        result = f"Deletion of {account_id} was rejected by reviewer: {decision.get('reason', 'no reason given')}"

    return {"messages": [{"role": "tool", "content": result, "tool_call_id": tool_call["id"]}]}

Calling interrupt(payload) immediately halts the graph at that exact line — not just before the node, but mid-node — and surfaces payload to whatever is polling graph.aget_state(). When the graph is resumed with Command(resume=decision), execution picks back up inside that same function call, with interrupt() now returning decision instead of pausing. This gives you approval logic that lives right next to the business logic it's protecting, rather than being bolted on as a separate compiled-graph setting.

For this tutorial's core scenario, interrupt_before is the cleaner fit since we want every deletion reviewed — but it's worth knowing both tools exist, because most production systems end up using a mix of both.

Injecting Human Approval or Rejection

Now for the payoff: how does a human actually approve or reject the paused action?

The key primitive is Command, imported from langgraph.types. You don't mutate the state dictionary directly and hope for the best — you pass a Command object into ainvoke, using the same thread_id the graph paused on.

Approving the Action

from langgraph.types import Command

async def approve_pending_action(thread_id: str):
    graph, pool = await build_graph()
    config = {"configurable": {"thread_id": thread_id}}

    # Resuming with None input simply lets the graph proceed past the interrupt
    result = await graph.ainvoke(None, config=config)

    print(result["messages"][-1].content)
    await pool.close()

When your graph was paused via interrupt_before, resuming is as simple as calling ainvoke(None, config=config) again with the same thread ID — LangGraph knows exactly where it left off and continues into the sensitive_tools node.

Rejecting the Action

Rejection is more interesting, because you don't want to silently continue into the tool call — you want to redirect the agent. The cleanest way is to update the graph's state before resuming, replacing the pending tool call with a synthetic tool response that tells the model the action was denied.

from langchain_core.messages import ToolMessage

async def reject_pending_action(thread_id: str, reason: str):
    graph, pool = await build_graph()
    config = {"configurable": {"thread_id": thread_id}}

    state = await graph.aget_state(config)
    pending_call = state.values["messages"][-1].tool_calls[0]

    rejection_message = ToolMessage(
        content=f"Action rejected by human reviewer. Reason: {reason}",
        tool_call_id=pending_call["id"],
    )

    # Inject the rejection as if it were the tool's own output,
    # then jump straight back to the agent node instead of sensitive_tools
    await graph.aupdate_state(
        config,
        {"messages": [rejection_message]},
        as_node="sensitive_tools",
    )

    result = await graph.ainvoke(None, config=config)
    print(result["messages"][-1].content)
    await pool.close()

The as_node="sensitive_tools" argument tells LangGraph to treat this update as though it came from the sensitive_tools node — meaning the graph's edges will correctly route back to agent next, letting the model see the rejection and respond appropriately ("I wasn't able to delete that account — a reviewer declined the request because...") instead of blindly retrying.

If you're using the dynamic interrupt() pattern instead, rejection is even simpler — you just resume with a Command carrying the decision payload:

await graph.ainvoke(
    Command(resume={"approved": False, "reason": reason}),
    config=config,
)

Real-World Example: An Agent That Can Delete Production Records

Let's tie it together into something resembling an actual service. Imagine a FastAPI backend exposing three endpoints: start a task, list pending approvals, and resolve one.

from fastapi import FastAPI

app = FastAPI()
GRAPH = None  # initialized on startup, backed by the pool from build_graph()

@app.post("/tasks")
async def start_task(user_request: str, thread_id: str):
    config = {"configurable": {"thread_id": thread_id}}
    result = await GRAPH.ainvoke({"messages": [HumanMessage(content=user_request)]}, config=config)
    state = await GRAPH.aget_state(config)
    if state.next:
        return {"status": "pending_approval", "thread_id": thread_id}
    return {"status": "completed", "response": result["messages"][-1].content}

@app.get("/approvals/{thread_id}")
async def get_pending(thread_id: str):
    config = {"configurable": {"thread_id": thread_id}}
    state = await GRAPH.aget_state(config)
    pending = state.values["messages"][-1].tool_calls[0]
    return {"tool": pending["name"], "args": pending["args"]}

@app.post("/approvals/{thread_id}/decision")
async def resolve(thread_id: str, approved: bool, reason: str = ""):
    if approved:
        await approve_pending_action(thread_id)
    else:
        await reject_pending_action(thread_id, reason)
    return {"status": "resolved"}

This is roughly the shape of a real approval dashboard's backend: an operator sees a queue of pending, high-risk actions (populated by scanning threads where state.next is non-empty), reviews the intended tool_call.args, and clicks approve or reject — all backed by durable Postgres state that survives deploys, crashes, and long delays.

🚀 Pro Tips

Index your checkpoints table by app-level metadata. The raw checkpoints table is keyed by thread_id, but you'll usually want a separate lightweight table (e.g., pending_approvals) that stores thread_id, requested_at, tool_name, and status, so you can query "all pending approvals older than 1 hour" without deserializing checkpoint blobs.
Set a TTL policy for stale threads. Not every paused agent will be resolved. Build a cleanup job that expires threads sitting in interrupt state past a business-defined SLA (e.g., auto-reject after 48 hours).
Use aget_state_history() for audit trails. Every checkpoint is retained by default, so you can replay the entire decision path of a thread — including what the model considered before and after the human's input — which is invaluable for compliance reviews.
Namespace thread IDs meaningfully. Something like f"support:{ticket_id}:{account_id}" makes debugging and querying dramatically easier than random UUIDs.
Test interrupts with MemorySaver first. Iterate on your graph topology locally with the in-memory checkpointer, then swap in AsyncPostgresSaver only once the interrupt/resume logic is correct — it's a one-line change and saves you from debugging graph logic and connection pooling issues simultaneously.

Best Practices for Production HITL Systems

Keep sensitive tools structurally separate. As shown above, routing sensitive and safe tool calls to different nodes makes interrupt_before trivial to reason about — you never have to inspect tool names deep inside conditional logic at runtime.
Always resume with the original thread_id. It's the only link between the paused state and the human decision. Losing it means losing the ability to resume gracefully.
Log every approval and rejection with reviewer identity. aupdate_state accepts arbitrary state fields — add a reviewed_by field to your state schema and populate it on every resume for a proper audit log.
Design for asynchronous, not synchronous, approval. Don't hold an HTTP request open waiting for a human. Return a pending_approval status immediately and let the client poll or receive a webhook/notification when resolved.
Validate tool arguments before showing them to a human. Malformed or suspicious arguments (e.g., an account_id that doesn't match any known record) should be caught and flagged automatically rather than presented as routine approvals — reviewer fatigue is real, and a queue full of noise trains humans to rubber-stamp everything.
Version your graph carefully. If you change node names or graph topology, old checkpoints referencing interrupt_before=["sensitive_tools"] may not resume cleanly against a new graph definition. Treat graph structure changes with the same care as database schema migrations.

Common Mistakes to Avoid

Forgetting to call checkpointer.setup(). This silently fails in confusing ways — usually a relation "checkpoints" does not exist error the first time you try to persist state.
Using a single raw connection instead of a pool. Under any real concurrency, this becomes a bottleneck or a source of connection-state bugs. Always use AsyncConnectionPool.
Mixing sync and async checkpointer classes. PostgresSaver (sync) and AsyncPostgresSaver (async) are not interchangeable — calling sync methods on an async-configured graph (or vice versa) raises confusing runtime errors. Pick one and be consistent across your codebase.
Mutating state directly instead of using Command or aupdate_state. LangGraph's state transitions are meant to flow through its own APIs so that checkpoint history stays consistent — reaching into the underlying row directly breaks that guarantee.
Not handling the "no pending interrupt" case. If a human tries to approve a thread that already resolved (e.g., a double-click on an approval button), your resume call should check state.next first and short-circuit gracefully rather than throwing.
Assuming interrupt_before gives you the tool call in a human-readable form. The raw tool_calls payload is JSON — build a proper rendering layer for your approval UI rather than dumping raw args at reviewers.

📌 Key Takeaways

HITL approval gates convert irreversible, high-risk agent actions into reviewable checkpoints instead of blind executions.
AsyncPostgresSaver gives LangGraph durable, multi-worker-safe persistence — essential once approvals can take longer than a single request lifecycle.
interrupt_before is the simplest way to force a pause ahead of a specific node; the dynamic interrupt() function offers finer-grained, conditional control from inside node logic.
Resuming execution always flows through thread_id plus either a plain ainvoke(None, config=...) call or a Command(resume=...)/aupdate_state() call carrying the human's decision.
Treat your approval queue as a first-class product surface — reviewer fatigue and stale threads are real operational risks, not edge cases.

Conclusion

Giving an AI agent the ability to take real-world action is powerful — and genuinely risky the moment that action can't be undone. LangGraph's checkpointing model turns "pause and ask a human" from an awkward workaround into a first-class, durable pattern. By pairing interrupt_before (or dynamic interrupt()) with AsyncPostgresSaver, you get an agent that behaves less like a script gambling on good outcomes and more like a well-governed system: one that knows exactly which actions require sign-off, persists its intentions safely while it waits, and picks up precisely where it left off once a human weighs in.

This pattern scales from a single support-ops agent all the way up to multi-team approval pipelines, and it's the same underlying mechanism whether you're gating a refund, a database mutation, or a deployment. Once you've wired it up for one sensitive tool, adding the next one is mostly a matter of naming it in the right set.

Build the guardrail once. Trust the agent more, precisely because you no longer have to trust it blindly.