Lab 12 · LangGraph Agent with Human-in-the-Loop

Low-confidence drafts pause the graph with interrupt(), wait for a human decision, and resume from a checkpoint — the durable-execution pattern that separates production agents from demo loops. Your deliverable includes a trace of a genuinely paused-and-resumed episode.

Time & tier

Advanced · ~90 minutes hands-on · API spend well under $1.

Setup

Python 3.11+, an Anthropic API key, and:

mkdir lab-12 && cd lab-12 && python -m venv .venv && source .venv/bin/activate
pip install -U langgraph anthropic pydantic
export ANTHROPIC_API_KEY="sk-ant-..."

Fully self-contained: the "knowledge base" is the ten Orbitly policy snippets from lab 09. Create corpus.py:

DOCS = [
    "Refunds are available within 14 days of purchase on Pro and Team plans. Free workspaces are not eligible for refunds.",
    "SAML SSO is available on the Team plan only. Pro and Free plans support Google sign-in but not SAML.",
    "Workspace admins can export all data as CSV or JSON. Exports are generated within 24 hours and download links expire after 7 days.",
    "Adding a seat mid-cycle is prorated to the day. Removing a seat takes effect at the next renewal; no partial credits are issued.",
    "The API allows 600 requests per minute per workspace. Exceeding the limit returns HTTP 429 with a Retry-After header.",
    "Deleted records are held in trash for 30 days, after which they are permanently purged and cannot be recovered.",
    "The uptime SLA is 99.9% for Team plans, measured monthly. Credits are 10% of the monthly fee per 0.1% below target.",
    "Trials last 21 days and include all Team features. No credit card is required to start a trial.",
    "Two-factor authentication can be enforced workspace-wide by an admin. Enforcement gives existing members a 7-day grace period.",
    "Each workspace includes 100 GB of storage on Pro and 1 TB on Team. Additional storage is billed at $5 per 100 GB per month.",
]

STOP = {"the", "is", "a", "an", "of", "do", "i", "can", "what", "how", "on", "in",
        "my", "to", "and", "are", "for", "does", "if", "long", "get", "need"}

def _tokens(text: str) -> set[str]:
    return {w.strip("?.,;:()").lower() for w in text.split()} - STOP

def retrieve(question: str, k: int = 2) -> list[str]:
    q = _tokens(question)
    scored = sorted(DOCS, key=lambda d: -len(q & _tokens(d)))
    return [d for d in scored[:k] if len(q & _tokens(d)) >= 1]

Token-overlap retrieval is deliberately crude — this lab is about graph control flow; swap in your lab-05/06 retriever later without touching the graph.

Guided steps

1. 1.

   Define state and the drafting node. Create agent.py. State is a typed dict the checkpointer persists at every step:

   Copy

   import os, anthropic
   from typing import TypedDict, Literal
   from pydantic import BaseModel
   from langgraph.graph import StateGraph, START, END
   from langgraph.types import interrupt, Command
   from langgraph.checkpoint.memory import InMemorySaver
   from corpus import retrieve
   
   client = anthropic.Anthropic(max_retries=3)
   MODEL = os.getenv("MODEL", "claude-opus-4-8")
   
   class AgentState(TypedDict, total=False):
       question: str
       docs: list[str]
       draft: str
       confidence: float
       route: str
       final: str
   
   class Draft(BaseModel):
       answer: str
       confidence: float          # 0-1: how fully the context supports the answer
       grounded: Literal["yes", "partial", "no"]
   
   def retrieve_node(state: AgentState) -> AgentState:
       return {"docs": retrieve(state["question"])}
   
   def draft_node(state: AgentState) -> AgentState:
       ctx = "\n".join(state["docs"]) or "(no relevant policy found)"
       resp = client.messages.parse(
           model=MODEL, max_tokens=400,
           system=("Answer the support question strictly from the policy context. "
                   "Report confidence as the fraction of your answer the context explicitly supports; "
                   "if the context does not cover the question, answer briefly, set confidence below 0.5, "
                   "and set grounded to 'no'."),
           messages=[{"role": "user", "content": f"<context>{ctx}</context>\n\n{state['question']}"}],
           output_format=Draft,
       )
       d = resp.parsed_output
       return {"draft": d.answer, "confidence": d.confidence}
   

2. 2.

   Add the router and the human gate. Routing is a pure function of state — testable without any model call. The gate calls interrupt(), which pauses the graph and surfaces a payload to whoever is driving it:

   Copy

   def route(state: AgentState) -> str:
       if not state["docs"]:
           return "escalate"
       return "finalize" if state["confidence"] >= 0.7 else "human_gate"
   
   def human_gate(state: AgentState) -> AgentState:
       decision = interrupt({
           "question": state["question"],
           "draft": state["draft"],
           "confidence": state["confidence"],
           "instructions": "Reply with {'approve': bool, 'edited_answer': str|None}",
       })
       if decision.get("approve"):
           return {"final": decision.get("edited_answer") or state["draft"], "route": "approved"}
       return {"final": "Escalated to a human agent — a specialist will reply within 4 business hours.",
               "route": "rejected"}
   
   def finalize(state: AgentState) -> AgentState:
       return {"final": state["draft"], "route": "auto"}
   
   def escalate(state: AgentState) -> AgentState:
       return {"final": "Escalated: no covering policy found. Ticket created.", "route": "no_context"}
   
   builder = StateGraph(AgentState)
   builder.add_node("retrieve", retrieve_node)
   builder.add_node("draft", draft_node)
   builder.add_node("human_gate", human_gate)
   builder.add_node("finalize", finalize)
   builder.add_node("escalate", escalate)
   builder.add_edge(START, "retrieve")
   builder.add_edge("retrieve", "draft")
   builder.add_conditional_edges("draft", route,
       {"finalize": "finalize", "human_gate": "human_gate", "escalate": "escalate"})
   for n in ["finalize", "human_gate", "escalate"]:
       builder.add_edge(n, END)
   
   graph = builder.compile(checkpointer=InMemorySaver())
   

   The checkpointer is what makes interrupt() resumable: every node's output is persisted per thread_id, so the resume replays into the exact paused frame.

3. 3.

   Drive it: the happy path. Create run.py:

   Copy

   from agent import graph
   
   cfg = {"configurable": {"thread_id": "t-happy"}}
   out = graph.invoke({"question": "How long is an export download link valid?"}, config=cfg)
   print(out["route"], "->", out["final"])          # expect: auto -> 7 days...
   

4. 4.

   The paused-and-resumed episode. Ask something the corpus half-covers, so confidence drops:

   Copy

   cfg = {"configurable": {"thread_id": "t-hitl"}}
   out = graph.invoke({"question": "Can I pay the storage overage fee in rupees by bank transfer?"}, config=cfg)
   
   if "__interrupt__" in out:
       payload = out["__interrupt__"][0].value
       print("PAUSED for review:", payload["confidence"], "\n", payload["draft"])
       # ... hours later, possibly from a different process reading the same checkpoint DB:
       from langgraph.types import Command
       resumed = graph.invoke(Command(resume={"approve": False, "edited_answer": None}), config=cfg)
       print(resumed["route"], "->", resumed["final"])   # rejected -> escalation message
   

   The resume value becomes the return value of the interrupt() call inside human_gate — the node continues mid-function as if it had never stopped. Run it again on a fresh thread and approve with an edited_answer to see the third outcome.

5. 5.

   Prove the state survived. Print the checkpoint history for the paused thread:

   Copy

   for snap in graph.get_state_history(cfg):
       print(snap.next, list(snap.values.keys()))
   

   You should see the pre-interrupt checkpoint with next=('human_gate',) and the post-resume one with the final answer. Capture this output plus the step-4 transcript into trace.md — that is your HITL evidence.

6. 6.

   Make it durable (production note). InMemorySaver dies with the process. Swap in SqliteSaver (pip install langgraph-checkpoint-sqlite) or the Postgres checkpointer and re-run step 4 with the pause and the resume in two separate process invocations — same thread_id, state loaded from disk. That two-process run is the difference between "supports HITL" and "supports HITL across a shift change."

Verification

• •
  High-confidence question completes with route == "auto" and never interrupts.
• •
  The low-confidence run returns an __interrupt__ payload containing draft and confidence; the process can exit and a second invocation resumes it.
• •
  Rejection produces the escalation message; approval with edited_answer returns the edited text.
• •
  get_state_history shows checkpoints spanning the pause (with next=('human_gate',)) and the resume.
• •
  The out-of-corpus question (retrieval returns nothing) routes straight to escalate without a gate.

Completion artifact

The lab-12 repo + trace.md showing one complete paused-and-resumed episode (interrupt payload, resume command, final state, checkpoint history). Self-grade (pass at 4+):

• Explicit routing: routing is a pure function with an explicit confidence threshold.
• Complete interrupt payload: the payload carries everything a reviewer needs to decide.
• Resume works: resume works from a separate invocation of the same thread.
• All routes shown: all three routes (auto / reviewed / escalated) demonstrated.
• Durability: durability path (Sqlite/Postgres) attempted or documented.

Stuck?

• interrupt raises instead of pausing: you compiled without a checkpointer, or invoked without configurable.thread_id — both are mandatory for HITL.
• Resume re-runs the whole graph: you passed a fresh input dict instead of Command(resume=...), or changed the thread_id.
• __interrupt__ missing: the draft came back above 0.7 — ask something further from the corpus, or temporarily raise the threshold to 0.9 to force the gate.
• parsed_output is None: check stop_reason; bump max_tokens.

Client-safe variant

For client deployments the human gate is not a demo feature — it is the control your statement of work should promise: any customer-facing answer below threshold is held for a named reviewer, and the checkpoint database (Postgres, in the client's VPC) is the audit log of who approved what and when. Route model calls through the client's sanctioned endpoint (Bedrock via AnthropicBedrockMantle — the graph code is unchanged; only the client constructor differs) and keep checkpoints, drafts, and reviewer decisions in-tenancy, since drafts may quote customer data. Set the confidence threshold with the client from measured data (lab 09's calibration approach), and document it as a tunable policy, not a magic number.