Phase 4: Multi-Agent Systems · 70 min · Python · LangGraph · Anthropic SDK
LangGraph for Stateful Multi-Agent Workflows
A workflow that can't resume from failure isn't production-ready.
What you will learn
- Build a LangGraph StateGraph with typed state, nodes, and conditional edges
- Implement checkpointing with a SQLite persister for fault-tolerant workflows
- Add interrupt/resume logic for human-in-the-loop approval gates
The Problem
Ad-hoc multi-agent systems have a critical flaw: they hold all state in memory. One server restart, one API timeout, one process kill — and the entire workflow must restart from scratch. For tasks that take minutes or hours, this is unacceptable. LangGraph solves this with a graph-based execution model that persists state after every node, enabling checkpointing, resumption, and human-in-the-loop approval gates.
LangGraph is not just a convenience library. It's an architectural decision: your workflow's state is a first-class citizen that survives failures.
LangGraph core concepts
StateGraph: a directed graph where each node is a function that receives the current state and returns updates. Nodes don't communicate directly — they read from state and write back to state.
State: a TypedDict (or Pydantic model) that flows through all nodes. Think of it as a shared whiteboard: every node can read anything on it, and write its results to specific fields.
Nodes: Python functions or agent calls. Each node signature is (state: State) -> dict. The dict it returns is merged into the state.
Edges: connections between nodes. A normal edge always flows forward. A conditional edge routes to different next nodes based on the current state.
Checkpointer: a persistence backend (SQLite, Postgres) that saves state after each node completes. This is what enables resumption after failure.
Typed state design
The State TypedDict is the blueprint of your workflow's memory. Design it upfront — changing it mid-project is painful.
from typing import TypedDict, Optional, Annotated
from langgraph.graph.message import add_messages
class ReportState(TypedDict):
task: str # Original user request (read-only after set)
research_results: str # Output of research node
draft: str # Output of draft node
review_feedback: str # Output of review node
final_report: str # Output of formatter node
revision_count: int # How many times we've revised
approval_status: Optional[bool] # Set by human-in-the-loop
Key design principles:
- Fields are written by specific nodes — document which node owns which field
revision_count prevents infinite revision loopsapproval_status: Optional[bool] enables the human-in-the-loop gate- Use
Annotated[list, add_messages] for message lists that should accumulate (append-only)
A LangGraph workflow fails on step 7 of 12 due to an API timeout. With a Postgres checkpointer, what happens when you re-run?
Checkpointers save state after EVERY node execution. Step 6 completed successfully — its state was persisted. On re-run with the same thread_id, LangGraph loads the checkpoint from after step 6 and resumes from step 7. This is the primary value of checkpointing: idempotent resumption from any failure point without re-doing completed work.
Checkpointing and human-in-the-loop
Setting up checkpointing:
from langgraph.checkpoint.sqlite import SqliteSaver
checkpointer = SqliteSaver.from_conn_string("workflow.db")
graph = workflow.compile(checkpointer=checkpointer)
# Run with a thread ID — each thread has its own checkpoint history
config = {"configurable": {"thread_id": "report-job-001"}}
result = graph.invoke(initial_state, config=config)
# If it fails partway through, re-run with the SAME thread_id:
result = graph.invoke(initial_state, config=config) # resumes from last checkpoint
Human-in-the-loop with interrupt():
LangGraph's interrupt() pauses graph execution at a specific node and returns control to the caller. The caller can present the current state to a human, collect input, and then call graph.invoke() with the human's decision to resume.
from langgraph.types import interrupt, Command
def send_node(state: ReportState) -> dict:
# Pause here and wait for human approval
decision = interrupt({
"action": "approve_send",
"draft_preview": state["final_report"][:500]
})
if decision == "approve":
# do the actual send
return {"sent": True}
else:
return {"approval_status": False, "review_feedback": "Human rejected."}
# In the caller:
result = graph.invoke(state, config=config)
# result is the interrupt payload — show it to the human
human_decision = input("Approve? (yes/no): ")
# Resume from where it paused:
result = graph.invoke(Command(resume=human_decision), config=config)
You need human approval before an agent sends an email. Using LangGraph's interrupt(), what is the correct flow?
LangGraph's interrupt() is designed exactly for this. It pauses execution at a specific point, surfaces current state to the caller, and waits. The caller can do anything during the pause — show a UI, send a notification, wait for async input. When ready, Command(resume=decision) continues from exactly where it paused. This is the production human-in-the-loop pattern.
Build
Build it
LangGraph itself needs its runtime to try live, but the checkpoint-and-resume idea can be illustrated with a plain Python dict standing in for a persistence backend.
# Minimal checkpoint/resume simulation (no dependencies)
checkpoint_store = {}
def run_step(thread_id, step_name, step_fn, state):
key = (thread_id, step_name)
if key in checkpoint_store:
print(f" [skip] '{step_name}' already checkpointed, reusing result")
return checkpoint_store[key]
result = step_fn(state)
checkpoint_store[key] = result
print(f" [run] '{step_name}' executed and checkpointed")
return result
steps = [
("research", lambda s: {**s, "research": "3 key facts"}),
("draft", lambda s: {**s, "draft": "a 200-word report"}),
("review", lambda s: 1 / 0), # simulate a crash on this step
]
thread_id = "report-001"
state = {"task": "transformer attention"}
for name, fn in steps:
try:
state = run_step(thread_id, name, fn, state)
except ZeroDivisionError:
print(f" [fail] '{name}' crashed — will resume from last checkpoint next run")
break
print(f"\nCheckpoints saved so far: {[k[1] for k in checkpoint_store]}")
The code below implements a 4-node checkpointed research-to-report workflow with a human approval gate.
"""
lesson_04_04_langgraph.py
LangGraph stateful workflow: research → draft → review → (conditional) send
with SQLite checkpointing and human-in-the-loop interrupt at the send node.
"""
import os
import sqlite3
from typing import TypedDict, Optional
from langgraph.graph import StateGraph, START, END
from langgraph.checkpoint.sqlite import SqliteSaver
from langgraph.types import interrupt, Command
import anthropic
ANTHROPIC_KEY = os.getenv("ANTHROPIC_API_KEY")
client = anthropic.Anthropic(api_key=ANTHROPIC_KEY) if ANTHROPIC_KEY else None
MAX_REVISIONS = 3
class ReportState(TypedDict):
task: str
research_results: str
draft: str
review_feedback: str
final_report: str
revision_count: int
approval_status: Optional[bool]
def _llm(prompt: str, system: str = "", max_tokens: int = 500) -> str:
if not client:
return f"[Mock LLM response for: {prompt[:60]}...]"
response = client.messages.create(
model="claude-haiku-4-5",
max_tokens=max_tokens,
system=system or "You are a helpful AI assistant.",
messages=[{"role": "user", "content": prompt}],
)
return response.content[0].text
def research_node(state: ReportState) -> dict:
print(f" [research] Researching: {state['task']}")
result = _llm(
f"Research this topic and list 5 key facts: {state['task']}",
system="You are a research specialist. Be concise and factual.",
)
return {"research_results": result}
def draft_node(state: ReportState) -> dict:
print(f" [draft] Writing draft (revision {state.get('revision_count', 0) + 1})...")
feedback = state.get("review_feedback", "")
prompt = (
f"Write a 200-word report on '{state['task']}' using this research:\n"
f"{state['research_results']}"
)
if feedback and state.get("revision_count", 0) > 0:
prompt += f"\n\nAddress this feedback:\n{feedback}"
draft = _llm(prompt, system="You are a technical writer. Be clear and structured.")
return {
"draft": draft,
"revision_count": state.get("revision_count", 0) + 1,
}
def review_node(state: ReportState) -> dict:
print(f" [review] Reviewing draft...")
review = _llm(
f"Review this report for accuracy, clarity, and completeness. "
f"Start with APPROVED or NEEDS_REVISION.\n\n{state['draft']}",
system="You are an editor. Be specific about any issues.",
)
approved = "APPROVED" in review.upper()
return {
"review_feedback": review,
"approval_status": approved if state.get("revision_count", 0) < MAX_REVISIONS else True,
}
def send_node(state: ReportState) -> dict:
print(f" [send] Requesting human approval...")
decision = interrupt({
"action": "approve_report",
"preview": state["draft"][:400],
"revision_count": state.get("revision_count", 1),
})
if str(decision).lower() in ("yes", "approve", "y"):
print(" [send] APPROVED by human. Sending report.")
return {"final_report": state["draft"], "approval_status": True}
else:
print(" [send] REJECTED by human.")
return {"approval_status": False, "review_feedback": f"Human rejected: {decision}"}
def should_revise(state: ReportState) -> str:
"""Conditional edge: route based on review verdict and revision count."""
if state.get("approval_status") and state.get("revision_count", 0) <= MAX_REVISIONS:
return "send"
elif state.get("revision_count", 0) >= MAX_REVISIONS:
return "send" # Force to send after max revisions
else:
return "draft" # Revise
def build_graph(checkpointer) -> StateGraph:
workflow = StateGraph(ReportState)
workflow.add_node("research", research_node)
workflow.add_node("draft", draft_node)
workflow.add_node("review", review_node)
workflow.add_node("send", send_node)
workflow.add_edge(START, "research")
workflow.add_edge("research", "draft")
workflow.add_edge("draft", "review")
workflow.add_conditional_edges("review", should_revise, {"send": "send", "draft": "draft"})
workflow.add_edge("send", END)
return workflow.compile(checkpointer=checkpointer)
if __name__ == "__main__":
import json
with SqliteSaver.from_conn_string("workflow_checkpoint.db") as checkpointer:
graph = build_graph(checkpointer)
config = {"configurable": {"thread_id": "report-001"}}
initial_state: ReportState = {
"task": "The impact of transformer architecture on modern NLP",
"research_results": "",
"draft": "",
"review_feedback": "",
"final_report": "",
"revision_count": 0,
"approval_status": None,
}
print("Running workflow...")
result = graph.invoke(initial_state, config=config)
# If interrupted, handle human approval
if "__interrupt__" in str(result):
print("\nWorkflow paused for human approval.")
print(f"Preview: {result.get('draft', '')[:300]}...")
decision = input("\nApprove this report? (yes/no): ").strip() or "yes"
result = graph.invoke(Command(resume=decision), config=config)
print("\n" + "=" * 60)
print("WORKFLOW COMPLETE")
print("=" * 60)
print(f"Final report:\n{result.get('final_report', result.get('draft', 'N/A'))[:500]}")
print(f"\nRevisions: {result.get('revision_count', 0)}")
print(f"Approved: {result.get('approval_status', False)}")
Unlock the full lesson
You've read the first 2 sections. The rest of this lesson covers Check Yourself — plus a hands-on lab, quiz, and project artifact.
Create a free account — the opening phases of 29 of 35 courses are free, no credit card.
Browse all courses · View pricing · DeVenture Academy