LangGraph supervisor template with budget guards, retries, and resumable human approval

A developer released a production-shaped LangGraph multi-agent template that wires five critical production features—structured routing, parallel-branch reducers, per-node retry policies, hard budget ceilings, and resumable human approval—into a single runnable graph. The template uses a Literal type for routing to prevent hallucinated node names and enforces hop and token budgets in code rather than delegating to the model.

| """ | | | supervisor graph.py - a production-shaped LangGraph multi-agent template. | | | Most LangGraph examples stop at "supervisor routes to workers." That part is easy. | | | The parts that actually break in production are scattered across a dozen doc pages: | | | 1. Routing that can't hallucinate a node name - structured output over a Literal | | | 2. Parallel branches clobbering shared state - a reducer on every multi-writer field | | | 3. One flaky call killing a 40-step run - per-node RetryPolicy | | | 4. A supervisor loop quietly burning $200 - hop + token budget enforced in code | | | 5. Crash recovery and human sign-off - checkpointer + interrupt | | | This file wires all five into one runnable graph. It is a skeleton, not a demo: | | | swap the worker prompts and you have a real pipeline. | | | Tested against: | | | langgraph 0.3.x, langchain-anthropic 0.3.x, langgraph-checkpoint-sqlite 2.x | | | Setup: | | | pip install langgraph langchain-anthropic langgraph-checkpoint-sqlite pydantic | | | export ANTHROPIC API KEY=... | | | python supervisor graph.py | | | """ | | | from future import annotations | | | import operator | | | import os | | | from typing import Annotated, Literal, TypedDict | | | from langchain anthropic import ChatAnthropic | | | from langchain core.messages import AIMessage, AnyMessage, HumanMessage, SystemMessage | | | from langgraph.checkpoint.sqlite import SqliteSaver | | | from langgraph.graph import END, START, StateGraph | | | from langgraph.graph.message import add messages | | | from langgraph.types import Command, RetryPolicy, interrupt | | | from pydantic import BaseModel, Field | | | Nothing below is model-specific. Point this anywhere. | | | MODEL = os.environ.get "AGENT MODEL", "claude-sonnet-4-5" | | | Hard ceilings. These are checked in code, never delegated to the model -- | | | a supervisor asked to reason about its own budget will reason its way into | | | one more loop every single time. | | | MAX HOPS = 12 | | | MAX TOKENS = 120 000 | | | WORKERS = "researcher", "analyst", "writer" | | | -------------------------------------------------------------------------- | | | State | | | -------------------------------------------------------------------------- | | | class Finding TypedDict : | | | agent: str | | | content: str | | | class AgentState TypedDict : | | | """Rule of thumb: any field more than one node can write in the same | | | superstep needs a reducer, or LangGraph raises InvalidUpdateError the first | | | time you fan out. draft and approved have exactly one writer each, so | | | they're safe as last-write-wins.""" | | | messages: Annotated list AnyMessage , add messages | | | findings: Annotated list Finding , operator.add | | | tokens used: Annotated int, operator.add | | | hops: Annotated int, operator.add | | | task: str | | | draft: str | | | approved: bool | | | class Route BaseModel : | | | """The Literal is the entire point. The model physically cannot emit a node | | | name that isn't in the graph, which kills the single most common failure | | | mode in string-parsed supervisors.""" | | | next: Literal "researcher", "analyst", "writer", "FINISH" | | | reason: str = Field description="One sentence. This is what you read in traces at 2am." | | | def text msg: AIMessage - str: | | | """Anthropic returns content as a block list the moment tools or extended | | | thinking are in play. Never index msg.content 0 -- flatten it.""" | | | if isinstance msg.content, str : | | | return msg.content | | | return "\n".join | | | b.get "text", "" | | | for b in msg.content | | | if isinstance b, dict and b.get "type" == "text" | | | | | | def digest findings: list Finding - str: | | | if not findings: | | | return " nothing yet -- you're first " | | | return "\n\n".join f" {f 'agent' } \n{f 'content' }" for f in findings | | | -------------------------------------------------------------------------- | | | Workers | | | -------------------------------------------------------------------------- | | | WORKER PROMPTS = { | | | "researcher": | | | "You gather raw material. Surface concrete facts, numbers, and sources. " | | | "Do not interpret, do not recommend. If you don't know something, say so plainly." | | | , | | | "analyst": | | | "You pressure-test what the researcher found. Name the weakest assumption, " | | | "the missing counter-evidence, and the one thing that would change the conclusion." | | | , | | | "writer": | | | "You produce the deliverable. Plain language, no preamble, no summary of " | | | "your own process. Lead with the answer." | | | , | | | } | | | def make worker name: str : | | | llm = ChatAnthropic model=MODEL, max tokens=2048 | | | def worker state: AgentState - dict: | | | reply = llm.invoke | | | SystemMessage WORKER PROMPTS name , | | | HumanMessage | | | f"Task: {state 'task' }\n\n" | | | f"What the team has so far:\n{ digest state 'findings' }" | | | , | | | | | | body = text reply | | | usage = reply.usage metadata or {} | | | out: dict = { | | | "findings": Finding agent=name, content=body , | | | "messages": AIMessage content=body, name=name , | | | "tokens used": usage.get "total tokens", 0 , | | | } | | | if name == "writer": | | | out "draft" = body | | | return out | | | return worker | | | -------------------------------------------------------------------------- | | | Supervisor | | | -------------------------------------------------------------------------- | | | def supervisor | | | state: AgentState, | | | - Command Literal "researcher", "analyst", "writer", "approval", " end " : | | | Guard first, LLM second. Cheaper and unbypassable. | | | if state "hops" = MAX HOPS or state "tokens used" = MAX TOKENS: | | | return Command | | | goto="approval" if state.get "draft" else END, | | | update={"messages": SystemMessage "Budget ceiling hit. Wrapping up." }, | | | | | | router = ChatAnthropic model=MODEL, max tokens=512 .with structured output Route | | | route: Route = router.invoke | | | SystemMessage | | | "You coordinate three specialists: researcher gathers , analyst " | | | " critiques , writer produces the deliverable . Pick who goes next. " | | | "Choose FINISH only once a draft exists and the analyst has seen it. " | | | "Do not send work to an agent whose input hasn't changed since last turn." | | | , | | | HumanMessage | | | f"Task: {state 'task' }\n" | | | f"Hops used: {state 'hops' }/{MAX HOPS}\n\n" | | | f"Transcript:\n{ digest state 'findings' }" | | | , | | | | | | if route.next == "FINISH": | | | return Command goto="approval" if state.get "draft" else "writer" | | | return Command | | | goto=route.next, | | | update={"hops": 1, "messages": SystemMessage f"- {route.next}: {route.reason}" }, | | | | | | -------------------------------------------------------------------------- | | | Human gate | | | -------------------------------------------------------------------------- | | | def approval state: AgentState - Command Literal "supervisor", " end " : | | | GOTCHA: on resume, LangGraph re-executes this node from the top. Every | | | line above interrupt runs twice. Keep DB writes, emails, and payments | | | strictly below it. | | | decision = interrupt { | | | "draft": state.get "draft", "" , | | | "findings": len state "findings" , | | | "tokens used": state "tokens used" , | | | "how to resume": "{'approved': true} to ship, or {'approved': false, 'notes': '...'}", | | | } | | | if decision.get "approved" : | | | return Command goto=END, update={"approved": True} | | | return Command | | | goto="supervisor", | | | update={"messages": HumanMessage f"Revision requested: {decision.get 'notes', '' }" }, | | | | | | -------------------------------------------------------------------------- | | | Wiring | | | -------------------------------------------------------------------------- | | | def build graph checkpointer : | | | b = StateGraph AgentState | | | b.add node "supervisor", supervisor | | | b.add node "approval", approval | | | for name in WORKERS: | | | b.add node | | | name, | | | make worker name , | | | On langgraph < 0.2.60 this kwarg is retry= instead. | | | retry policy=RetryPolicy | | | max attempts=3, | | | initial interval=1.0, | | | backoff factor=2.0, | | | retry on= TimeoutError, ConnectionError , | | | , | | | | | | b.add edge START, "supervisor" | | | for name in WORKERS: | | | b.add edge name, "supervisor" workers always report back | | | No edges out of supervisor or approval -- Command goto=... carries the | | | routing. The Command Literal ... return annotations are what let | | | graph.get graph .draw mermaid still render those hops. | | | return b.compile checkpointer=checkpointer | | | def main - None: | | | task = "Should a two-person team self-host Postgres or pay for a managed instance?" | | | cfg = {"configurable": {"thread id": "demo-001"}} | | | seed: AgentState = { | | | "messages": , | | | "findings": , | | | "tokens used": 0, | | | "hops": 0, | | | "task": task, | | | "draft": "", | | | "approved": False, | | | } | | | with SqliteSaver.from conn string "checkpoints.sqlite" as saver: | | | graph = build graph saver | | | for update in graph.stream seed, cfg, stream mode="updates" : | | | for node, payload in update.items : | | | print f" {node} {str payload :160 }" | | | snapshot = graph.get state cfg | | | if snapshot.next: parked on the interrupt | | | payload = snapshot.tasks 0 .interrupts 0 .value | | | print "\n--- awaiting human ---" | | | print payload "draft" :800 | | | print f" {payload 'tokens used' :,} tokens, {payload 'findings' } findings \n" | | | Kill the process here and rerun with the same thread id: it picks | | | up exactly at this line. That is the whole value of the checkpointer. | | | for update in graph.stream Command resume={"approved": True} , cfg, stream mode="updates" : | | | for node, p in update.items : | | | print f" {node} {str p :160 }" | | | print "\nfinal:", graph.get state cfg .values "draft" :800 | | | if name == " main ": | | | main | | | -------------------------------------------------------------------------- | | | Things that cost me time, in rough order of pain: | | | | | | - InvalidUpdateError on fan-out almost always means a shared state field is | | | missing a reducer, not that your graph shape is wrong. | | | - interrupt replays the node from the top. Side effects go after it. | | | - Checkpoints are keyed on thread id. Reusing one silently resumes an old | | | run; generate a fresh id per job unless resumption is the point. | | | - RetryPolicy wraps the node, not the LLM call, so a retried node re-emits | | | its full state update. Keep nodes idempotent. | | | - MemorySaver is fine for tests and useless in prod -- it dies with the | | | process. Sqlite for single-box, Postgres for anything real. | | | - recursion limit in config is a separate ceiling from your hop budget. Set | | | both; they fail differently and you want to know which one tripped. | | | -------------------------------------------------------------------------- |