LangGraph Stateful RAG-Agent with Human-in-the-Loop

Tier: Advanced · Workflow area: Agentic systems

This playbook builds a RAG-agent that routes each query to retrieve / answer / escalate, and interrupts for human approval when its confidence is low — the pattern behind support bots, compliance assistants, and any agent whose wrong answer costs more than a pause.

When to use

• •
  A RAG assistant where some fraction of answers must be reviewed by a human before delivery (support, legal, medical, finance).
• •
  Agent runs that must survive process restarts — long tasks, worker deployments, retries.
• •
  LangGraph is named in the job descriptions you are targeting and you need a defensible working example.

For a linear retrieve-then-answer pipeline with no routing and no pauses, plain RAG (your pb from the retrieval module) is simpler — a graph earns its complexity only when control flow branches or pauses.

Prerequisites

• •
  A working baseline RAG stack (retriever + LLM) from earlier playbooks.
• •
  pip install langgraph langgraph-checkpoint-postgres psycopg[binary] and a Postgres instance (Docker is fine); langgraph.checkpoint.memory.InMemorySaver works for development.
• •
  An LLM API key; cheap-tier models are fine for the router node.

The workflow

1. 1.

   Define state and nodes. State is a typed dict: question, docs, draft, confidence, final. Nodes: router (classify: needs retrieval / answerable directly / out-of-scope), retrieve, generate (produce draft + a 0–1 confidence from a rubric or logprob-free self-grade), human_review, respond, escalate.

2. 2.

   Wire the graph with conditional edges.

   Copy

   from typing import TypedDict
   from langgraph.graph import StateGraph, START, END
   from langgraph.types import interrupt, Command
   from langgraph.checkpoint.postgres import PostgresSaver
   
   class State(TypedDict, total=False):
       question: str; docs: list; draft: str; confidence: float; final: str
   
   def human_review(state: State):
       decision = interrupt({          # pauses run, persists state
           "draft": state["draft"],
           "confidence": state["confidence"],
           "ask": "approve / edit / escalate",
       })
       if decision["action"] == "edit":
           return {"final": decision["text"]}
       if decision["action"] == "escalate":
           return {"final": "ESCALATED"}
       return {"final": state["draft"]}
   
   builder = StateGraph(State)
   # ... add_node for router/retrieve/generate/human_review/respond/escalate
   builder.add_conditional_edges("router", route_fn,
       {"retrieve": "retrieve", "answer": "generate", "escalate": "escalate"})
   builder.add_conditional_edges("generate",
       lambda s: "human_review" if s["confidence"] < 0.6 else "respond",
       {"human_review": "human_review", "respond": "respond"})
   
   with PostgresSaver.from_conn_string(DB_URI) as checkpointer:
       checkpointer.setup()   # first run only
       graph = builder.compile(checkpointer=checkpointer)
   

3. 3.

   Run with a thread id; detect the pause. config = {"configurable": {"thread_id": "case-123"}}; result = graph.invoke({"question": q}, config). If the result contains an __interrupt__ entry, the run is paused — surface the payload to your reviewer UI (even a CLI prompt for the lab).

4. 4.

   Resume with the human decision. graph.invoke(Command(resume={"action": "approve"}), config) — the value passed to Command(resume=...) becomes the return value of the interrupt() call inside the node, and execution continues from the checkpoint. Kill and restart the process between pause and resume to prove durability.

5. 5.

   Tune the confidence threshold empirically. Run 30 labeled questions; plot review-rate vs error-rate at thresholds 0.4–0.8 and pick the knee. The threshold is a product decision — document it.

Tool variants

• Checkpointer:InMemorySaver for dev; PostgresSaver for anything shared; SQLite saver for single-machine tools.
• LangSmith (managed) adds tracing and a review UI over the same graph; self-host Arize Phoenix for the free path (pb-49).
• Claude Agent SDK / OpenAI Agents SDK cover the agent loop but durable interrupt-and-resume with pluggable checkpointers is the LangGraph-specific muscle this playbook exercises.

Verification

• •
  A low-confidence question pauses; the process is killed and restarted; resume completes correctly from Postgres state.
• •
  The router sends at least one query down each of the three paths on your test set.
• •
  Threshold experiment table exists: review-rate and error-rate at three or more thresholds.
• •
  Two runs with different thread_ids do not leak state into each other.

Pitfalls

• No checkpointer.interrupt() requires one — without it the graph cannot pause and resume. This is the most common setup error.
• Side effects before the interrupt. Code earlier in the node re-executes on resume; keep nodes idempotent or put side effects after the interrupt() call.
• Self-reported confidence taken at face value. Calibrate it against labels (pb from the evals module) before it gates anything real.
• Reusing one thread_id for all users. Thread id is the conversation key; collisions merge strangers' histories.

Restricted-environment variant

Everything self-hosts: LangGraph and the Postgres checkpointer are OSS, the retriever is your local vector store, and the LLM can be a vLLM-served open-weight model behind an OpenAI-compatible endpoint. For client engagements, keep checkpoint tables in the client's own Postgres — checkpoints contain full state including retrieved document text.

Success criteria

• •
  Working graph with conditional routing, a durable HITL pause, and a documented threshold choice.
• •
  Kill-and-resume demonstrated with evidence (logs or a short recording).
• •
  Review-rate vs error-rate trade-off table a stakeholder could act on.
• •
  You can explain to an interviewer why checkpointing, not prompting, is LangGraph's differentiator.