1. Introduction
What is LangGraph?
LangGraph is a framework for building stateful, multi-actor applications with Large Language Models (LLMs). While LangChain is great for chaining prompts, LangGraph introduces cycles (loops), controllability, and persistence. It allows you to model complex workflows as graphs where nodes are actions (agents/tools) and edges are the logic connecting them.
Why State Management Matters
In traditional software, state is managed via databases or session objects. In AI, state is the context. Without state management, an LLM is just a stateless function: you pass a prompt, it returns a completion, and it forgets everything. State management allows agents to remember past actions, current goals, and intermediate results.
Stateless vs. Stateful Agents
Stateless Agent: Forgets everything after a single turn. If the server crashes, the workflow dies. You cannot pause for human approval.
Stateful Agent: Remembers context across turns. If the server crashes, it resumes exactly where it left off. It can pause, wait for a human, and continue.
Challenges in Production AI Systems
Context Window Limits: LLMs forget if the conversation gets too long.
Determinism: LLMs are probabilistic; workflows need guaranteed execution paths.
Fault Tolerance: LLM APIs timeout. Servers crash. Workflows must recover.
Human-in-the-Loop (HITL): High-stakes decisions require human approval, requiring the workflow to "pause" and "resume".
2. Real-Time Business Use Case: Multi-Agent Insurance Claim Processing
The Business Problem
An insurance company receives thousands of claims daily. Processing them manually is slow and error-prone. Fully automating them is risky due to fraud. We need a system that automates the mundane, detects fraud, and routes high-risk claims to humans.
Architecture & Agent Responsibilities
Document Intake Agent: Receives PDFs/images, extracts metadata.
OCR Agent: Converts images to text.
Validation Agent: Checks if all required fields (policy number, date) are present.
Fraud Detection Agent: Analyzes text/patterns, outputs a fraud score (0-100).
Risk Assessment Agent: Calculates financial risk and payout estimate.
Human Review Agent: (HITL) Pauses the workflow if fraud score > 80. Waits for adjuster approval.
Decision Agent: Finalizes the claim (Approved/Rejected) and triggers payment.
3. Understanding LangGraph State
State is the heart of LangGraph. It is the single source of truth passed between all nodes.
State Schema (TypedDict)
LangGraph uses Python's TypedDict to define state, ensuring type safety.
1. Simple State
from typing import TypedDict, Annotated
from langgraph.graph.message import add_messages
class SimpleState(TypedDict):
messages: Annotated[list, add_messages] # Reducer appends new messages
2. Nested State
class ClaimData(TypedDict):
claim_id: str
documents: list[str]
fraud_score: float
class NestedState(TypedDict):
messages: Annotated[list, add_messages]
claim: ClaimData
3. Enterprise State
In production, you need metadata for routing, auditing, and multi-tenancy.
class EnterpriseState(TypedDict):
messages: Annotated[list, add_messages]
claim: ClaimData
metadata: dict # Contains user_id, tenant_id, trace_id
current_step: str
requires_human_review: bool
Why Reducers Matter: Notice Annotated[list, add_messages]. Without a reducer, a new node returning {"messages": [new_msg]} would overwrite the list. The reducer appends to it.
4. Agent Memory Types
Memory in AI is not a single concept. It is layered.
A. Short-Term Memory (Thread State)
What it is: The conversation history and current workflow state for a single execution (thread).
Where it's stored: LangGraph Checkpointer (Postgres/SQLite).
Lifespan: Dies when the thread is deleted.
B. Long-Term Memory (Cross-Thread)
What it is: User preferences, historical claims, learned facts.
Where it's stored: Vector DB (pgvector) or Relational DB (Postgres JSONB).
Lifespan: Permanent.
C. Working Memory (Context Window)
What it is: The specific slice of state passed to the LLM in a single prompt.
Where it's stored: LLM Provider's RAM.
Lifespan: Single LLM call.
D. Shared Team Memory (Graph State)
Memory Hierarchy Diagram
5. Checkpointing Fundamentals
What is Checkpointing?
Checkpointing is the process of saving the entire Graph State to a persistent storage medium (disk or database) after every node execution.
Why Checkpoints are Needed
Failure Recovery: If node 3 fails, you don't restart from node 1. You resume from node 3.
Human-in-the-Loop: The workflow pauses, saves state to DB, and waits. Hours later, a human approves, and the workflow resumes.
Time Travel: You can revert to a previous checkpoint to debug or re-run with different parameters.
Long-Running Workflows: Workflows that take days (waiting for external APIs or humans) cannot live in RAM.
6. Implementing Checkpointing (Local Testing)
Let's build a basic graph with an in-memory checkpointer.
Installation
pip install langgraph langchain-openai langchain-core
Step-by-Step Code
from typing import TypedDict, Annotated
from langgraph.graph import StateGraph, START, END
from langgraph.graph.message import add_messages
from langgraph.checkpoint.memory import MemorySaver
# 1. Define State
class State(TypedDict):
messages: Annotated[list, add_messages]
claim_status: str
# 2. Define Nodes
def process_claim(state: State):
print("Processing claim...")
return {"claim_status": "processed", "messages": [{"role": "assistant", "content": "Claim processed."}]}
def finalize_claim(state: State):
print("Finalizing claim...")
return {"claim_status": "finalized", "messages": [{"role": "assistant", "content": "Claim finalized."}]}
# 3. Build Graph
builder = StateGraph(State)
builder.add_node("process_claim", process_claim)
builder.add_node("finalize_claim", finalize_claim)
builder.add_edge(START, "process_claim")
builder.add_edge("process_claim", "finalize_claim")
builder.add_edge("finalize_claim", END)
# 4. Add Checkpointer
memory = MemorySaver()
graph = builder.compile(checkpointer=memory)
# 5. Execute
config = {"configurable": {"thread_id": "claim-123"}}
result = graph.invoke({"messages": [], "claim_status": "new"}, config)
print(result["claim_status"]) # Output: finalized
Explanation:
MemorySaver(): Stores state in RAM. Great for local dev, bad for production.
thread_id: The unique identifier for this specific workflow run. If you run it again with claim-123, it resumes from the end.
7. Production Checkpointing
For production, RAM is not enough. We need persistent storage.
SQLite Checkpointer
Good for single-server deployments or local testing that requires persistence.
pip install langgraph-checkpoint-sqlite
from langgraph.checkpoint.sqlite import SqliteSaver
import sqlite3
conn = sqlite3.connect("checkpoints.sqlite", check_same_thread=False)
checkpointer = SqliteSaver(conn)
graph = builder.compile(checkpointer=checkpointer)
PostgreSQL Checkpointer (Enterprise Standard)
Best for distributed, multi-server production environments.
pip install langgraph-checkpoint-postgres psycopg
Note: Modern LangGraph uses psycopg (v3), not psycopg2.
from langgraph.checkpoint.postgres import PostgresSaver
from psycopg_pool import ConnectionPool
# Create connection pool
pool = ConnectionPool(
conninfo="postgresql://user:password@localhost:5432/langgraph",
max_size=10,
)
checkpointer = PostgresSaver(pool)
# IMPORTANT: You must run setup to create the tables
checkpointer.setup()
graph = builder.compile(checkpointer=checkpointer)
VS Code Project Structure
8. Human-in-the-Loop Workflow
In modern LangGraph (v0.2+), we use interrupt() to pause and Command(resume=...) to continue.
The Scenario
If Fraud Score > 80, pause and wait for human.
Code Implementation
from langgraph.types import interrupt, Command
def fraud_detection_node(state: State):
# Simulate LLM call
fraud_score = 85.0
state["claim"]["fraud_score"] = fraud_score
if fraud_score > 80:
# PAUSE WORKFLOW
human_feedback = interrupt({
"message": "High fraud score detected. Please review.",
"fraud_score": fraud_score
})
# When resumed, human_feedback contains the data passed via Command(resume=...)
if not human_feedback.get("approved"):
state["claim_status"] = "rejected_by_human"
return state
state["claim_status"] = "approved_by_fraud_check"
return state
# Compile graph
graph = builder.compile(checkpointer=PostgresSaver(pool))
config = {"configurable": {"thread_id": "claim-999"}}
# 1. Run until interrupt
result = graph.invoke({"messages": [], "claim": {"fraud_score": 0}}, config)
# Graph pauses here. State is saved to Postgres.
# ... Hours later, human reviews in UI ...
# 2. Resume workflow
human_decision = {"approved": True, "reviewer_id": "agent_smith"}
final_result = graph.invoke(Command(resume=human_decision), config)
Execution Flow
9. Recovery After Failure
Checkpoints make your system invincible to transient failures.
Simulating Failures
import random
def flaky_llm_call_node(state: State):
# Simulate random failures
if random.choice([True, False]):
raise ConnectionError("LLM API Timeout!")
return {"messages": [{"role": "assistant", "content": "Success"}]}
How Recovery Works
Because LangGraph saves state after every node, if flaky_llm_call_node crashes, the state up to that point is safe.
# First run: Crashes
try:
graph.invoke(initial_state, config)
except Exception as e:
print(f"Failed: {e}")
# Second run: Resumes automatically!
# LangGraph sees the thread_id, loads the last checkpoint,
# and re-executes ONLY the failed node.
result = graph.invoke(None, config)
Case Studies:
Server Crash: Pod dies. K8s restarts it. API receives request with thread_id. LangGraph loads from Postgres and resumes.
Agent Exception: Tool throws error. Catch it, log it, and re-invoke.
Network Outage: LLM API is down. Retry logic inside the node handles it. If it exhausts retries, it throws. Re-invoke later.
LLM Timeout: Handled exactly like network outage.
10. Multi-Agent Memory Sharing
Agents don't need complex message-passing protocols. They share the Graph State.
# Agent A writes
def agent_a(state: State):
state["shared_data"] = {"extracted_text": "Policy #12345"}
return state
# Agent B reads
def agent_b(state: State):
text = state["shared_data"]["extracted_text"]
print(f"Agent B read: {text}")
return state
# Agent C updates
def agent_c(state: State):
state["shared_data"]["validated"] = True
return state
Rule of Thumb: Use the state for structured data sharing. Use messages for conversational context.
11. Production Architecture
Here is the enterprise-grade architecture for this system
Layer Breakdown
API Layer: Handles HTTP, validates JWTs, translates requests to LangGraph invocations.
Orchestration: Runs the LangGraph workers. Uses async to handle high concurrency.
Checkpoint Store: Postgres. Stores state, checkpoints, and long-term memory (JSONB).
Vector DB: Stores policy documents for RAG during the Risk Assessment phase.
LLM Gateway: Abstracts OpenAI/Claude/Gemini, handles fallbacks and load balancing.
12. Monitoring and Observability
You cannot manage what you cannot measure.
LangSmith Integration
LangSmith is the official observability platform for LangChain/LangGraph.
import os
os.environ["LANGCHAIN_TRACING_V2"] = "true"
os.environ["LANGCHAIN_API_KEY"] = "your-langsmith-key"
os.environ["LANGCHAIN_PROJECT"] = "insurance-claims-prod"
What to Monitor
Tracing: See the exact prompt, LLM response, and tool calls for every node.
Latency: Track time per node. If OCR takes 10s, you need to optimize.
Error Rates: Track LLM hallucinations or tool failures.
Audit Trails: Because of checkpointing, you have a perfect, immutable log of every state transition for compliance.
13. Scaling to Thousands of Workflows
When you move from 10 to 10,000 concurrent claims:
Concurrent Execution: Use Python's asyncio and LangGraph's ainvoke.
Distributed Checkpoints: Postgres becomes a bottleneck. Use PgBouncer for connection pooling.
Database Scaling: Partition your Postgres checkpoint tables by tenant_id or created_at.
Redis Caching: Use Redis to cache frequent LLM responses (semantic caching) or store ephemeral rate-limit counters.
Worker Scaling: Deploy LangGraph workers as stateless containers (Docker/K8s). The state lives in Postgres, so you can scale workers horizontally to infinity.
14. Common Interview Questions
1. What is state in LangGraph?
State is a TypedDict that acts as the single source of truth for the workflow. It is passed to every node, and nodes return dictionaries that update the state via reducers.
2. Why checkpointing?
Checkpointing persists the state to disk/DB after every step. It enables fault tolerance (crash recovery), human-in-the-loop (pausing), and time-travel debugging.
3. Difference between memory and state?
State is the current context of a single workflow run (short-term). Memory usually refers to long-term storage (like a Vector DB) that persists across multiple different workflow runs and threads.
4. How does workflow recovery work?
When a node fails, the exception bubbles up. Because the previous node's output was checkpointed, re-invoking the graph with the same thread_id loads the last checkpoint and re-executes only the failed node.
5. How do you avoid infinite loops?
Use the recursion_limit parameter in graph.invoke(). Also, design your graph with clear conditional edges that eventually route to an END node.
6. How do agents communicate?
They don't message each other directly. They communicate implicitly by reading and writing to the shared Graph State.
7. How does human approval work?
Using the interrupt() function inside a node. This pauses execution and saves the state. The frontend shows an approval UI. When approved, the backend calls graph.invoke(Command(resume=data)) to continue.
8. How do you scale LangGraph?
Keep the workers stateless. Store checkpoints in a managed, highly available database like PostgreSQL. Use connection pooling (PgBouncer), scale workers horizontally via Kubernetes, and use async execution.
15. Complete End-to-End Project
Here is a runnable, simplified version of the project.
Folder Structure
langgraph-insurance/
├── .env
├── requirements.txt
└── main.py
requirements.txt
langgraph>=0.2.0
langchain-openai>=0.1.0
langchain-core>=0.2.0
python-dotenv
.env
OPENAI_API_KEY=sk-...
LANGCHAIN_API_KEY=ls-...
LANGCHAIN_TRACING_V2=true
main.py
import os
from typing import TypedDict, Annotated
from dotenv import load_dotenv
from langchain_openai import ChatOpenAI
from langgraph.graph import StateGraph, START, END
from langgraph.graph.message import add_messages
from langgraph.checkpoint.memory import MemorySaver
from langgraph.types import interrupt, Command
load_dotenv()
# 1. State Definition
class ClaimData(TypedDict):
claim_id: str
fraud_score: float
status: str
class InsuranceState(TypedDict):
messages: Annotated[list, add_messages]
claim: ClaimData
# 2. Initialize LLM
llm = ChatOpenAI(model="gpt-4o-mini", temperature=0)
# 3. Define Nodes
def intake_node(state: InsuranceState):
print(f"[Intake] Processing claim {state['claim']['claim_id']}")
return {"messages": [{"role": "assistant", "content": "Intake complete."}]}
def fraud_node(state: InsuranceState):
print("[Fraud] Analyzing for fraud...")
# Simulating a high fraud score to trigger HITL
score = 85.0
return {"claim": {**state["claim"], "fraud_score": score}}
def human_review_node(state: InsuranceState):
print("[Human Review] Pausing for human approval...")
# PAUSE WORKFLOW
feedback = interrupt({
"prompt": f"Claim {state['claim']['claim_id']} has fraud score {state['claim']['fraud_score']}. Approve?",
})
# RESUME WORKFLOW
if feedback.get("approved"):
return {"claim": {**state["claim"], "status": "approved"}}
else:
return {"claim": {**state["claim"], "status": "rejected"}}
def decision_node(state: InsuranceState):
status = state["claim"]["status"]
print(f"[Decision] Final status: {status}")
return {"messages": [{"role": "assistant", "content": f"Workflow finished. Status: {status}"}]}
# 4. Build Graph
builder = StateGraph(InsuranceState)
builder.add_node("intake", intake_node)
builder.add_node("fraud", fraud_node)
builder.add_node("human_review", human_review_node)
builder.add_node("decision", decision_node)
builder.add_edge(START, "intake")
builder.add_edge("intake", "fraud")
builder.add_edge("fraud", "human_review")
builder.add_edge("human_review", "decision")
builder.add_edge("decision", END)
# 5. Compile with Checkpointer
checkpointer = MemorySaver() # Use PostgresSaver in prod
graph = builder.compile(checkpointer=checkpointer)
# 6. Execution Flow
if __name__ == "__main__":
config = {"configurable": {"thread_id": "claim-001"}}
initial_state = {
"messages": [],
"claim": {"claim_id": "CLM-999", "fraud_score": 0.0, "status": "pending"}
}
print("--- Starting Workflow ---")
# This will pause at human_review_node
result = graph.invoke(initial_state, config)
print(f"Paused. Current status: {result['claim']['status']}")
print("\n--- Simulating Human Approval ---")
# Resume the workflow
final_result = graph.invoke(Command(resume={"approved": True}), config)
print(f"Finished. Final status: {final_result['claim']['status']}")
Building production AI is not just about prompting an LLM; it is about building resilient software systems around the LLM.
Agent Memory & State: The brain of your system. Use TypedDict and reducers to manage context efficiently.
Checkpointing: The safety net. Persisting state to Postgres ensures your system survives crashes and enables long-running processes.
Recovery: The superpower. Checkpointing turns fragile LLM scripts into fault-tolerant enterprise applications.
Human Review: The guardrail. Using interrupt() and Command allows seamless integration of human judgment into automated pipelines.
Production Deployment: Requires observability (LangSmith), scaling (connection pooling, async workers), and strict state management.
By mastering LangGraph's state and checkpointing mechanisms, you transition from building AI prototypes to engineering enterprise-grade AI systems.
Join the conversation! Your thoughts help the community grow.