LangGraph: LangChain's framework for building stateful, multi-actor agent applications

The Problem

Every agent framework that ships today handles the easy case well: a single LLM call, maybe a tool or two, return a result. The hard case — an agent that must loop, branch, pause for human input, survive a process crash, and resume from where it left off — is where every framework except LangGraph falls apart.

The root issue is architectural. Most frameworks model agent execution as a directed acyclic graph (DAG): a fixed sequence of steps that runs from start to finish. Real agent workflows are cyclic. An agent calls a tool, gets a result, calls another tool, loops back to re-evaluate, branches based on the result, and maybe pauses for a human to review before continuing. A DAG cannot express this. You end up bolting on while-loops, manual state tracking, and fragile error recovery that works in demos and breaks in production.

The real-world cost is measurable. A mid-size AI team running 5,000 agent workflows per month without cyclic orchestration loses 25-35% of executions to unrecoverable errors, context-window overflows from unbounded loops, and state corruption from partial failures.

Metric	Before (DAG-based orchestration)	After (LangGraph cyclic graphs)
Workflow completion rate	65%	97%
Error recovery (automatic)	0% (manual restart)	89% (checkpoint replay)
Human-in-the-loop support	Not supported	Native interrupt() API
State persistence	In-memory only	Postgres-backed checkpoints
Time-travel debugging	Not possible	Full replay from any checkpoint
Lines of orchestration code	280+	~80
Developer setup time	3-5 days	~2 hours

Why this matters: The difference between a demo agent and a production agent is not model quality — it is the ability to survive failures, accept human input mid-execution, and resume without data loss. LangGraph is the only open-source framework that makes these guarantees a first-class part of the graph definition, not an afterthought bolted onto a DAG.

The Investigation

The root cause of production agent failures is not the LLM. It is the execution model. Every popular multi-agent framework (CrewAI, AutoGen, the OpenAI Agents SDK) models workflows as a DAG: a fixed sequence of steps that runs once. This works for 80% of use cases. For the remaining 20% — the ones that involve loops, conditional branches, human approval gates, or long-running stateful processes — the DAG model forces developers to build their own state machine on top of a framework that was not designed for it.

What this means: A DAG-based agent that needs to retry a failed tool call must either re-run the entire pipeline (wasting tokens and time) or implement a retry loop outside the framework’s execution model (creating a maintenance burden). A LangGraph agent simply adds an edge from the error handler back to the tool-calling node. The graph runtime handles the loop, the checkpointing, and the state management.

LangGraph’s architecture is inspired by Google’s Pregel graph processing system and Apache Beam. The key insight is that agent workflows are graphs with cycles, not DAGs. A graph with cycles can express:

• Retry loops: Tool call fails -> route to error handler -> route back to retry
• Human-in-the-loop: Agent drafts output -> pause -> human reviews -> resume with edits
• Map-reduce fan-out: One node fans out to N parallel workers -> workers complete -> fan-in node aggregates results
• Evaluator-optimizer loops: Generate output -> evaluate quality -> loop back for revision if below threshold

LangChain’s internal benchmarks across 50,000 production agent runs show that cyclic graphs complete 97% of workflows vs 65% for equivalent DAG-based pipelines. The 32-point gap comes almost entirely from automatic recovery: LangGraph replays from the last checkpoint on crash, while DAG-based frameworks require manual restart from the beginning.

What this means: If your agent workflow has any of these patterns — retry, human approval, parallel fan-out, quality loops — LangGraph is not just better, it is the only framework that models them natively. Every other framework requires you to build these patterns yourself, outside the execution model, where they are harder to test, harder to debug, and harder to maintain.

Performance benchmarks on identical AWS m6i.4xlarge instances with GPT-4o show LangGraph’s median latency at 14.1s for research tasks (vs CrewAI’s 18.4s) and 8.3s for code review (vs CrewAI’s 9.1s). Token overhead versus raw API calls is +9% for LangGraph, +18% for CrewAI, and +31% for AutoGen. LangGraph is the fastest and most token-efficient framework in every benchmark.

The Solution

LangGraph is a low-level orchestration framework (MIT license, 35,000+ GitHub stars, 280+ contributors) that models agent workflows as stateful graphs with cycles. You define nodes (the work) and edges (the control flow), and the runtime handles state persistence, checkpointing, streaming, and error recovery.

                    +--------------------------------------------------+
                    |              StateGraph (Typed State)             |
                    |  Annotated[list, add] | Annotated[dict, merge]     |
                    +--------------------------------------------------+
                       |          |          |          |
            +----------+  +------+  +------+  +------+
            | Node A     |  | Node B    |  | Node C    |
            | Research   |  | Draft     |  | Review    |
            +------------+  +-----------+  +-----------+
                       |          |          |
                  +----+----+  +-+----+  +---+----+
                  | ToolNode|  | Tool  |  | Tool    |
                  | Search  |  | Write |  | Linter  |
                  +---------+  +------+  +--------+
                       |          |          |
                  +----+----+  +-+----+  +---+----+
                  | Cond.   |  | Cond. |  | Cond.  |
                  | Edge    |  | Edge  |  | Edge   |
                  +---------+  +------+  +--------+
                       |          |          |
                  +----+----+  +-+----+  +---+----+
                  | Check-  |  | Check|  | Check  |
                  | pointer  |  | point|  | point  |
                  +---------+  +------+  +--------+
                       |          |          |
                  +----+----+  +-+----+  +---+----+
                  | Postgres|  | Lang |  | Lang   |
                  | Saver   |  | Smith|  | Smith  |
                  +---------+  +------+  +--------+

Here’s what each piece does:

• StateGraph — the core abstraction. A graph with typed state, nodes, and edges. State is a TypedDict with reducer annotations that control how state updates are applied (append to lists, merge dicts, overwrite scalars).
• Nodes — Python functions that take state and return state updates. Each node is a unit of work: an LLM call, a tool execution, a conditional check, a human approval gate.
• Edges — control flow between nodes. Three types: normal edges (always traverse), conditional edges (branch based on state), and entry/exit points.
• ToolNode — a built-in node that executes tool calls returned by an LLM. Handles parallel tool execution, error handling, and result formatting.
• Checkpointer — persists graph state after every node execution. MemorySaver for development, PostgresSaver for production. Enables time-travel debugging, crash recovery, and human-in-the-loop.
• Interrupt — pauses graph execution and waits for human input. The graph state is checkpointed at the pause point. Resuming injects the human’s decision and continues from the exact same state.
• Send API — the primitive for map-reduce fan-out. A node returns Send(destination, state) for each parallel branch, and the runtime executes them concurrently.

Production-Grade Code Walkthrough

Here is a complete, production-grade content pipeline using LangGraph with Postgres-backed checkpoints, human-in-the-loop approval, and conditional routing:

import os
import operator
from typing import Annotated, TypedDict, Literal
from langgraph.graph import StateGraph, START, END
from langgraph.checkpoint.postgres import PostgresSaver
from langgraph.types import interrupt, Command, Send
from langchain_openai import ChatOpenAI
from langchain_core.messages import HumanMessage, SystemMessage
from langchain_community.tools import DuckDuckGoSearchRun

# ---------------------------------------------------------------------------
# State schema — TypedDict with reducer annotations
# ---------------------------------------------------------------------------

class ContentState(TypedDict):
    topic: str
    research_brief: Annotated[list[str], operator.add]  # append-only
    draft: str
    review_score: float
    review_feedback: str
    final_article: str
    approved: bool
    revision_count: int
    last_error: str

# ---------------------------------------------------------------------------
# LLM setup
# ---------------------------------------------------------------------------

llm = ChatOpenAI(
    model=os.getenv("OPENAI_MODEL", "gpt-4o"),
    temperature=0.3,
)
search = DuckDuckGoSearchRun()

# ---------------------------------------------------------------------------
# Nodes — each is a pure function: state in, state update out
# ---------------------------------------------------------------------------

def research_node(state: ContentState) -> dict:
    """Research the topic and append findings to the research brief."""
    prompt = (
        f"Research the topic: '{state['topic']}'\n"
        "Provide 5-7 key findings with sources. Focus on 2025-2026 data."
    )
    response = llm.invoke([SystemMessage(content=prompt)])
    return {"research_brief": [response.content]}

def draft_node(state: ContentState) -> dict:
    """Draft an article from the accumulated research brief."""
    brief = "\n\n".join(state["research_brief"])
    prompt = (
        f"Write a 1000-word article on '{state['topic']}' "
        f"using this research:\n\n{brief}\n\n"
        "Include an executive summary and actionable takeaways."
    )
    response = llm.invoke([SystemMessage(content=prompt)])
    return {"draft": response.content}

def review_node(state: ContentState) -> dict:
    """Score the draft on quality. Returns score and feedback."""
    prompt = (
        f"Review this article for quality:\n\n{state['draft']}\n\n"
        "Return a JSON object with 'score' (0.0-1.0) and 'feedback' (string)."
    )
    response = llm.invoke([SystemMessage(content=prompt)])
    # In production, parse structured output with PydanticOutputParser
    return {
        "review_score": 0.85,  # placeholder — use structured parsing
        "review_feedback": response.content,
    }

def human_approval_node(state: ContentState) -> Command:
    """Pause and wait for human review. Returns Command to resume."""
    decision = interrupt({
        "draft": state["draft"],
        "score": state["review_score"],
        "feedback": state["review_feedback"],
    })
    if decision.get("approved"):
        return Command(
            update={"approved": True, "final_article": state["draft"]},
            goto="finalize",
        )
    return Command(
        update={
            "approved": False,
            "revision_count": state.get("revision_count", 0) + 1,
        },
        goto="revise",
    )

def finalize_node(state: ContentState) -> dict:
    """Mark the pipeline as complete."""
    print(f"Article approved. Final length: {len(state['final_article'])} chars")
    return {}

def revise_node(state: ContentState) -> dict:
    """Revise the draft based on review feedback."""
    prompt = (
        f"Revise this article based on the feedback:\n\n"
        f"Feedback: {state['review_feedback']}\n\n"
        f"Article:\n{state['draft']}\n\n"
        "Return the revised article."
    )
    response = llm.invoke([SystemMessage(content=prompt)])
    return {"draft": response.content}

# ---------------------------------------------------------------------------
# Conditional routing functions
# ---------------------------------------------------------------------------

def route_after_review(state: ContentState) -> Literal["human_approval", "revise"]:
    """Route based on review score. High scores skip to human approval."""
    if state["review_score"] >= 0.7:
        return "human_approval"
    return "revise"

def route_after_revision(state: ContentState) -> Literal["review", "finalize", "escalate"]:
    """Route after revision. Cap retries to prevent infinite loops."""
    if state.get("revision_count", 0) >= 3:
        return "escalate"
    return "review"

# ---------------------------------------------------------------------------
# Build the graph
# ---------------------------------------------------------------------------

builder = StateGraph(ContentState)

# Add nodes
builder.add_node("research", research_node)
builder.add_node("draft", draft_node)
builder.add_node("review", review_node)
builder.add_node("human_approval", human_approval_node)
builder.add_node("revise", revise_node)
builder.add_node("finalize", finalize_node)

# Add edges
builder.add_edge(START, "research")
builder.add_edge("research", "draft")
builder.add_edge("draft", "review")
builder.add_conditional_edges("review", route_after_review)
builder.add_edge("human_approval", "finalize")
builder.add_conditional_edges("revise", route_after_revision)
builder.add_edge("finalize", END)

# ---------------------------------------------------------------------------
# Compile with Postgres checkpointer (production)
# ---------------------------------------------------------------------------

connection_string = os.getenv(
    "POSTGRES_URL",
    "postgresql://postgres:postgres@localhost:5432/langgraph",
)

with PostgresSaver.from_conn_string(connection_string) as saver:
    saver.setup()  # idempotent: creates tables if they don't exist
    graph = builder.compile(checkpointer=saver)

    # -----------------------------------------------------------------------
    # Execute
    # -----------------------------------------------------------------------

    config = {"configurable": {"thread_id": "content-pipeline-001"}}

    # First run: research, draft, review
    for event in graph.stream(
        {"topic": "Multi-agent AI systems in enterprise 2026"},
        config,
        stream_mode="updates",
    ):
        for node_name, output in event.items():
            print(f"[{node_name}] completed")

    # Human-in-the-loop: resume with approval
    graph.invoke(
        Command(resume={"approved": True, "edited": None}),
        config,
    )

    # Check final state
    final_state = graph.get_state(config)
    print(f"Approved: {final_state.values['approved']}")

Setup Instructions

# Install LangGraph with Postgres support
pip install -U langgraph langgraph-checkpoint-postgres langchain-openai

# Or with uv (faster)
uv pip install -U langgraph langgraph-checkpoint-postgres langchain-openai

# Set your API keys
export OPENAI_API_KEY="sk-..."
export POSTGRES_URL="postgresql://postgres:postgres@localhost:5432/langgraph"

# Verify installation
python -c "from langgraph.graph import StateGraph; print('OK')"

How to Use Effectively

1. Design state first, nodes second

The state schema is the contract every node must satisfy. Design it before writing any node logic. Every field should be JSON-serializable from day one — LangGraph checkpoints serialize state to the database after every node.

from typing import Annotated, TypedDict
import operator

class AgentState(TypedDict):
    messages: Annotated[list, operator.add]  # append-only
    findings: Annotated[list[str], operator.add]
    current_step: int
    approved: bool
    retry_count: int

Use Annotated[list, operator.add] for fields that accumulate across nodes (messages, research findings, tool results). Use plain types for fields that should be overwritten (current step, approval status).

2. Use PostgresSaver, not MemorySaver, from day one

MemorySaver stores checkpoints in a Python dict. It is lost on process restart. PostgresSaver persists checkpoints to PostgreSQL, enabling crash recovery, time-travel debugging, and human-in-the-loop across process restarts.

from langgraph.checkpoint.postgres import PostgresSaver

with PostgresSaver.from_conn_string(POSTGRES_URL) as saver:
    saver.setup()  # creates tables if they don't exist
    graph = builder.compile(checkpointer=saver)

The setup() call is idempotent — safe to run on every deploy. PostgresSaver uses connection pooling internally. For high-throughput deployments, tune the pool size via pool_config.

3. Use interrupt() for human-in-the-loop, not custom state flags

The interrupt() function pauses graph execution, checkpoints the state, and returns control to the caller. The caller inspects the state, makes a decision, and resumes with Command(resume=...). This is the only pattern that guarantees the graph state is consistent at the pause point.

from langgraph.types import interrupt, Command

def approval_node(state: AgentState) -> Command:
    decision = interrupt({"draft": state["draft"]})
    if decision.get("approved"):
        return Command(update={"approved": True}, goto="next")
    return Command(update={"approved": False}, goto="revise")

4. Use the Send API for map-reduce fan-out

When you need to process multiple items in parallel (research multiple sources, review multiple files), use the Send API. Each Send creates an independent branch that runs concurrently.

from langgraph.types import Send

def fanout_node(state: AgentState) -> list[Send]:
    return [
        Send("process_source", {"source": s, "topic": state["topic"]})
        for s in state["sources"]
    ]

The runtime executes all Send branches concurrently and fans back into the next node when all complete. This is the production pattern for parallel research, parallel code review, and parallel data extraction.

5. Set recursion limits and per-node timeouts

LangGraph’s default recursion limit is 25 steps. For graphs with loops (retry, revision, evaluator-optimizer), set an explicit limit to prevent runaway execution.

# Global recursion limit
graph = builder.compile(
    checkpointer=saver,
    interrupt_before=["human_approval"],
    recursion_limit=50,
)

# Per-node timeout (LangGraph 1.2+)
builder.add_node("long_task", long_task_fn, timeout=120)

Use Cases

1. Content Production Pipeline with Quality Gates

When you’d use this: You need to research, draft, review, and publish articles at scale with automatic revision loops and human approval for borderline-quality drafts.

Why LangGraph fits: The cyclic graph maps directly to the content workflow. The review node routes back to revision for low-scoring drafts, and the interrupt() API pauses for human approval on medium-scoring drafts. The Postgres checkpointer ensures no work is lost if the process crashes mid-revision. The Send API fans out research across multiple sources in parallel.

2. Customer Support Agent with Escalation

When you’d use this: Incoming support tickets need triage, knowledge base search, draft response, and optional human escalation — with the ability to pause mid-workflow while waiting for human input.

Why LangGraph fits: The interrupt() API pauses the graph when escalation is needed and resumes when the human responds. The checkpointer persists the ticket state across the pause, so the human can take hours to respond without losing context. Conditional edges route based on ticket urgency, customer tier, and issue category.

3. Code Review and PR Automation

When you’d use this: Every PR needs static analysis, security scanning, style checking, and architectural review — with the ability to auto-approve trivial changes and flag complex ones for human review.

Why LangGraph fits: The Send API fans out to parallel review nodes (static analysis, security, style, architecture) and fans back into a synthesis node. Conditional edges route based on the aggregate risk score. The checkpointer enables time-travel debugging: you can replay any PR’s review from any checkpoint to understand why a decision was made.

4. Multi-Step Data Pipeline with Validation

When you’d use this: Incoming data needs validation, transformation, enrichment, and loading into a target system — with retry logic for transient failures and dead-letter routing for permanent failures.

Why LangGraph fits: The cyclic graph expresses retry loops natively: a failed validation node routes back to a retry node with exponential backoff. Conditional edges route permanently failed records to a dead-letter queue node. The checkpointer ensures exactly-once processing: if the pipeline crashes mid-transform, it resumes from the last checkpoint without duplicating records.

5. Evaluator-Optimizer Loop for Code Generation

When you’d use this: An LLM generates code, a test runner evaluates it, and the system iterates until tests pass or a max iteration limit is reached.

Why LangGraph fits: The cyclic graph expresses the generate-evaluate loop as a first-class pattern. The evaluator node runs tests and returns pass/fail with error messages. A conditional edge routes back to the generator for failed tests or forward to the output node for passing tests. The recursion limit prevents infinite loops. The checkpointer saves each iteration’s state, enabling post-hoc analysis of the optimization trajectory.

Cheat Sheet

Aspect	Detail
License	MIT
GitHub stars	35,000+
Latest stable version	1.2.6 (June 2026)
Python version	3.10 - 3.13
TypeScript support	Yes (langgraphjs)
Core abstraction	StateGraph with typed state, nodes, and edges
State management	TypedDict with reducer annotations (Annotated[list, add])
Checkpoint backends	MemorySaver (dev), PostgresSaver (production), SQLiteSaver
Checkpoint storage reduction	DeltaChannel: 41x reduction vs full snapshots
Human-in-the-loop	interrupt() / Command(resume=...)
Parallel fan-out	Send() API for map-reduce
Streaming	Token-by-token, node-by-node, updates mode
Per-node timeouts	timeout=60 on add_node(), idle timeout with heartbeats
Error handlers	Node-level error handlers with saga/compensation pattern
Graceful shutdown	RunControl with cooperative drain on SIGTERM
LLM providers	Any LangChain-compatible: OpenAI, Anthropic, Ollama, Azure, AWS Bedrock, Google
Tool integration	ToolNode for parallel tool execution, any LangChain tool
Multi-agent patterns	Single-agent, supervisor-workers, hierarchical, map-reduce
Observability	LangSmith tracing (one env var), OpenTelemetry per-node spans
Deployment	Self-hosted (FastAPI + gunicorn), LangSmith Cloud (managed)
Production users	Klarna (85M users), LinkedIn, Uber, Replit, Elastic, J.P. Morgan
Median latency (research)	14.1s (GPT-4o, AWS m6i.4xlarge)
Token overhead vs raw API	+9%
Workflow completion rate	97% (with checkpoints)
Setup to first working graph	~2 hours

Vibe Coding Projects

Project 1: Personal Research Assistant with Parallel Search

What it does: A LangGraph agent that takes a research question, fans out to three parallel search nodes (web search, academic search, news search), synthesizes the results, and produces a structured research brief with citations. Includes a retry loop for failed searches.

What you’ll learn: Building a StateGraph with the Send API for parallel fan-out, configuring ToolNode for tool execution, implementing conditional edges for retry logic, and using PostgresSaver for checkpoint persistence.

Effort: 2-3 hours. Three parallel nodes, one synthesis node, one retry edge.

Project 2: Human-in-the-Loop Code Review Bot

What it does: Connects to a GitHub repository, reads open PRs, runs code review through three parallel agents (static analysis, security, architecture), and pauses for human approval before posting comments. The human can approve, reject, or edit the review before it is posted.

What you’ll learn: Using interrupt() for human-in-the-loop gates, implementing the Command(resume=...) pattern for resuming after human input, integrating with external APIs via custom tools, and using LangSmith tracing to debug the review pipeline.

Effort: 4-6 hours. Three parallel review nodes, one human approval gate, one post-comment node.

Project 3: Multi-Agent Code Generation with Evaluator-Optimizer Loop

What it does: An agent that generates Python code from a specification, runs it through a test suite, evaluates the results, and iterates until all tests pass or a max iteration limit is reached. Each iteration is checkpointed, so you can replay the optimization trajectory.

What you’ll learn: Building cyclic graphs with evaluator-optimizer loops, setting recursion limits to prevent runaway execution, using per-node timeouts for test execution, implementing error handlers for test failures, and analyzing checkpoint history for post-hoc debugging.

Effort: 6-8 hours. One generator node, one test executor node, one evaluator node, conditional edges for the loop, and a max-iterations guard.

Problems Solved Efficiently

Problem Type	Why LangGraph Fits	When to Look Elsewhere
Long-running stateful agents	Postgres-backed checkpoints survive process crashes; agents resume from last checkpoint	Your agent is a single LLM call with no state — use a direct API call
Human-in-the-loop workflows	Native interrupt() API pauses execution, checkpoints state, and resumes with human input	You never need human review — CrewAI’s sequential process is simpler
Cyclic workflows (retry, revision, optimization)	Graphs with cycles are a first-class concept, not a workaround	Your workflow is a fixed DAG — CrewAI or a simple script is faster to set up
Map-reduce parallel processing	Send() API fans out to N parallel branches and fans back in automatically	You have a single sequential pipeline — no parallel fan-out needed
Production-grade observability	LangSmith tracing with one env var; OpenTelemetry per-node spans	You need a fully managed platform with no self-hosting — use LangSmith Cloud
Time-travel debugging	Replay any execution from any checkpoint to debug failures	You never need to debug past executions — skip the checkpointer overhead
Multi-agent coordination	Supervisor-workers, hierarchical, and map-reduce patterns all supported	You need role-based agents with backstories — CrewAI’s metaphor is more intuitive

Architectural Tradeoffs

What We Gained

• Cyclic graphs as a first-class concept. Retry loops, revision loops, evaluator-optimizer loops — all expressed as graph edges, not while-loops bolted onto a DAG. This is the single biggest architectural advantage over every other framework.
• Durable execution with Postgres checkpoints. Every node execution is checkpointed. Crash recovery is automatic. Time-travel debugging is built-in. No other open-source framework offers this.
• Token efficiency. +9% overhead vs raw API calls, the lowest of any multi-agent framework. CrewAI is +18%, AutoGen is +31%. Over 10,000 monthly workflows, this difference is material.
• Production footprint. Klarna (85 million users), LinkedIn (AI Recruiter), Uber (test generation), Replit (coding agent), Elastic (SecOps) — all running LangGraph in production. No other framework has this breadth of verifiable production deployments.
• TypeScript support. LangGraph ships a first-class JavaScript/TypeScript package. If your stack is Node.js, you are not locked into Python.

What We Sacrificed

• Learning curve. LangGraph is a low-level framework. You define nodes, edges, state schemas, and conditional routing functions. Expect 2-3 days to go from zero to a working graph, compared to 25 minutes for CrewAI.
• Frequent breaking changes. 549 releases since inception. The API has shifted significantly across versions. Pin your LangGraph version and budget time for migration on each upgrade.
• No role-based metaphor. CrewAI’s agents have roles, goals, and backstories that shape every LLM call. LangGraph gives you raw functions and state. You build the agent persona yourself through system prompts and node logic.
• LangSmith dependency for observability. The best debugging experience requires LangSmith. The open-source tracing is functional but less polished. If you cannot use LangSmith (air-gapped environments, compliance restrictions), you lose the time-travel debugging that is LangGraph’s killer feature.
• State schema rigidity. Every field in your state schema is serialized on every checkpoint. Store large artifacts (documents, images, audio) in S3 and keep only URIs in state. The schema must be versioned — additive changes are safe, but removing or renaming fields breaks replay of existing checkpoints.

Real lesson from production: A team at a major fintech company deployed a LangGraph agent for automated invoice processing. They used MemorySaver for the first month because “we’ll switch to Postgres later.” On day 32, a Kubernetes pod restart during a 15-step workflow lost all state. The invoice was processed twice, resulting in a duplicate payment. The fix was switching to PostgresSaver — a three-line code change. The lesson: use PostgresSaver from day one, not after your first production incident.

Course-Style Deep Dive

Under the Hood: How LangGraph Executes a Graph

When you call graph.stream(inputs, config), the following happens:

1. State initialization. The runtime creates the initial state from the input dict, applying default values and reducer annotations. The state is a plain Python dict with typed fields.

2. Node execution. The runtime traverses the graph from START, executing each node in topological order. Each node receives the current state and returns a dict of state updates. The runtime applies the updates using the reducer annotations: operator.add appends to lists, operator.or_ merges dicts, and plain fields are overwritten.

3. Checkpointing. After each node completes, the runtime serializes the full state to the checkpointer (PostgresSaver, MemorySaver, etc.). With DeltaChannel (LangGraph 1.2+), only the incremental delta is stored, reducing checkpoint storage by 41x.

4. Edge traversal. After checkpointing, the runtime evaluates the outgoing edges. Normal edges always traverse. Conditional edges call the routing function with the current state and follow the returned edge name.

5. Superstep execution. When the runtime encounters a Send node, it creates N parallel branches (supersteps). Each superstep executes independently with its own state copy. The runtime waits for all supersteps to complete before fanning back into the next node.

6. Interrupt handling. When a node calls interrupt(), the runtime checkpoints the state, marks the graph as paused, and returns control to the caller. The caller inspects the state via graph.get_state(config) and resumes with graph.invoke(Command(resume=...), config).

7. Completion. When the runtime reaches END, it checkpoints the final state and returns. The caller can retrieve the final state via graph.get_state(config).

Advanced Pattern 1: Supervisor-Worker Multi-Agent Architecture

from langgraph.types import Send
from langgraph.graph import StateGraph, START, END

class SupervisorState(TypedDict):
    task: str
    sub_tasks: list[dict]
    results: Annotated[list[str], operator.add]
    final_answer: str

def planner_node(state: SupervisorState) -> dict:
    """Decompose the task into sub-tasks for workers."""
    prompt = f"Decompose this task into 3-5 sub-tasks:\n{state['task']}"
    response = llm.invoke([SystemMessage(content=prompt)])
    # Parse sub-tasks from response
    sub_tasks = [
        {"id": "1", "description": "Research competitors"},
        {"id": "2", "description": "Analyze market size"},
        {"id": "3", "description": "Identify key trends"},
    ]
    return {"sub_tasks": sub_tasks}

def assigner_node(state: SupervisorState) -> list[Send]:
    """Fan out to one worker per sub-task."""
    return [
        Send("worker", {"sub_task": st, "task": state["task"]})
        for st in state["sub_tasks"]
    ]

def worker_node(state: dict) -> dict:
    """Execute a single sub-task."""
    prompt = f"Task: {state['task']}\nSub-task: {state['sub_task']}"
    response = llm.invoke([SystemMessage(content=prompt)])
    return {"results": [response.content]}

def synthesizer_node(state: SupervisorState) -> dict:
    """Synthesize all worker results into a final answer."""
    all_results = "\n\n".join(state["results"])
    prompt = f"Synthesize these findings:\n\n{all_results}"
    response = llm.invoke([SystemMessage(content=prompt)])
    return {"final_answer": response.content}

builder = StateGraph(SupervisorState)
builder.add_node("planner", planner_node)
builder.add_node("assigner", assigner_node)
builder.add_node("worker", worker_node)
builder.add_node("synthesizer", synthesizer_node)
builder.add_edge(START, "planner")
builder.add_edge("planner", "assigner")
builder.add_conditional_edges("assigner", lambda s: [s], path_map=["worker"])
builder.add_edge("worker", "synthesizer")
builder.add_edge("synthesizer", END)

The supervisor decomposes the task, fans out to parallel workers via Send, and synthesizes the results. Each worker is a separate node that can have its own tools, LLM configuration, and error handling. This pattern is used by Klarna’s customer support system and LinkedIn’s AI Recruiter.

Advanced Pattern 2: DeltaChannel for Long-Running Agents

LangGraph 1.2 introduced DeltaChannel, which stores only incremental state deltas instead of full snapshots at every step. For agents that run hundreds of turns, this reduces checkpoint storage by 41x.

from langgraph.types import DeltaChannel

# Configure a DeltaChannel on your state field
channel = DeltaChannel(
    reducer=lambda s, xs: s + xs,
    snapshot_frequency=50,  # full snapshot every 50 steps
)

# In the graph definition, use DeltaChannel for high-churn fields
class LongRunningState(TypedDict):
    messages: Annotated[list, channel]  # DeltaChannel reducer
    accumulated_findings: Annotated[list[str], operator.add]
    step_count: int

Without DeltaChannel, a light coding-and-search agent running 500 turns produces 4 GB of checkpoint data. With DeltaChannel, the same agent produces 110 MB — a 41x reduction. For multi-file feature agents running 200 turns, the reduction is from 5.3 GB to 129 MB.

Production Considerations

• Recursion limit. Set recursion_limit on compile() to cap the maximum number of supersteps. Default is 25. For graphs with loops, set it to 50-100. For graphs without loops, leave it at 25 as a safety net.
• Checkpoint TTL. PostgresSaver checkpoints accumulate indefinitely. Set a TTL policy to prune completed threads after 7 days. Use a cron job or LangSmith’s built-in retention policy.
• Schema versioning. State schema changes must be additive. Adding a new field is safe. Removing or renaming a field breaks replay of existing checkpoints. If you must remove a field, deprecate it first (keep it in the schema but stop writing to it), then remove it after all active threads have completed.
• Bounded history. Use trim_messages() or a custom reducer to cap the message list length. Unbounded message lists grow with every turn, increasing checkpoint size and LLM context length.
• Idempotent tools. Every tool called by ToolNode should be idempotent. If a checkpoint is replayed (crash recovery, time-travel debugging), the tool may be called again. Idempotent tools prevent duplicate side effects.
• Per-node OpenTelemetry. Add OpenTelemetry spans to every node. Include thread ID, step count, and token count on every span. This is the minimum observability surface for debugging production agent failures.

The Results

Metric	Before (DAG-based orchestration)	After (LangGraph cyclic graphs)
Workflow completion rate	65%	97%
Error recovery (automatic)	0% (manual restart)	89% (checkpoint replay)
Human-in-the-loop support	Not supported	Native interrupt() API
State persistence	In-memory only	Postgres-backed checkpoints
Time-travel debugging	Not possible	Full replay from any checkpoint
Lines of orchestration code	280+	~80
Developer setup time	3-5 days	~2 hours
Token overhead vs raw API	+31% (AutoGen)	+9%
Median latency (research)	18.4s (CrewAI)	14.1s
Checkpoint storage (500 turns)	N/A (no checkpoints)	110 MB (with DeltaChannel)
Production deployments	Few verifiable	Klarna, LinkedIn, Uber, Replit, Elastic

What this means for you: If you are building an agent that must survive crashes, accept human input mid-execution, or run for more than a few turns, LangGraph is the only open-source framework that handles these requirements natively. The 32-point improvement in workflow completion rate is not theoretical — it comes from automatic checkpoint replay, which no other framework offers.

The tradeoff is a steeper learning curve and more verbose setup compared to CrewAI. But for the 20% of agent workflows that need cycles, branches, or human-in-the-loop, LangGraph is not just better — it is the only framework that can express these patterns without building a custom state machine on top of a DAG.

What to Watch Out For

Beginner Advice

1. Use PostgresSaver from day one, not MemorySaver. MemorySaver stores checkpoints in a Python dict. The first process restart loses all state. PostgresSaver is a three-line change and the only production-viable checkpointer. Do not wait for your first crash to switch.

2. Design your state schema before writing any nodes. The state schema is the contract every node must satisfy. Every field must be JSON-serializable. Use Annotated[list, operator.add] for accumulating fields and plain types for overwrite fields. Schema changes must be additive — removing a field breaks replay of existing checkpoints.

3. Set recursion_limit on every compiled graph. The default is 25. For graphs with loops (retry, revision, evaluator-optimizer), set it to 50-100. Without an explicit limit, a buggy conditional edge can create an infinite loop that burns API credits until you kill the process.

4. Test crash recovery in development, not production. Kill your agent process mid-execution and verify it resumes from the last checkpoint. Do this with PostgresSaver configured. If you used MemorySaver, the test is meaningless — there is no state to recover.

5. Use LangSmith tracing from the first run. Set LANGSMITH_TRACING=true and LANGSMITH_API_KEY before your first graph.stream() call. The trace shows every node execution, every state update, every tool call, and every edge traversal. Debugging a LangGraph agent without tracing is like debugging a distributed system without logs.

6. Keep state lean. Every field in your state schema is serialized on every checkpoint. Store large artifacts (documents, images, audio files) in S3 and keep only URIs in state. A 10 MB document in state becomes 10 MB on every checkpoint write. At 500 turns, that is 5 GB of checkpoint data.

7. Make every tool idempotent. Checkpoint replay may call a tool multiple times. If your tool sends an email or creates a database record, ensure it can be called twice without producing duplicate side effects. Use idempotency keys for API calls and upsert semantics for database writes.

8. Pin your LangGraph version. With 549 releases and frequent API changes, an unpinned version can break your graph on the next deploy. Pin to a specific version in your requirements file and test upgrades in a staging environment before rolling to production.

Lesson learned: “We deployed a LangGraph agent for automated customer response without setting a recursion limit. A conditional edge bug caused the agent to loop 47 times before we killed it. The API bill for those 47 iterations was $340. The fix was adding recursion_limit=20 to the compile call — a one-line change. Always set recursion_limit. Always.” — Engineering Lead, enterprise SaaS

Lesson learned: “Our first LangGraph deployment used MemorySaver because ‘we’ll switch to Postgres later.’ A Kubernetes pod restart during a 15-step invoice processing workflow lost all state. The invoice was processed twice, resulting in a duplicate payment of $12,000. The fix was switching to PostgresSaver — three lines of code. We should have done that before writing the first node, not after the first financial loss.” — Staff Engineer, fintech company

Lesson learned: “The supervisor-worker pattern with Send() is powerful but expensive if you don’t cap per-branch output. Our first implementation let each worker return unlimited research findings. A single 10-worker fan-out produced 80 KB of state per branch, totaling 800 KB for one superstep. After 50 turns, the checkpoint was 40 MB. We added a per-branch token limit and cut checkpoint size by 90%.” — ML Engineer, Replit

Getting Started

# Install
pip install -U langgraph langgraph-checkpoint-postgres langchain-openai

# Set up PostgreSQL (Docker for local development)
docker run -d --name langgraph-pg \
  -e POSTGRES_PASSWORD=postgres \
  -p 5432:5432 \
  postgres:16

# Set environment variables
export OPENAI_API_KEY="sk-..."
export POSTGRES_URL="postgresql://postgres:postgres@localhost:5432/langgraph"
export LANGSMITH_TRACING=true
export LANGSMITH_API_KEY="lsv2_..."

# Verify installation
python -c "from langgraph.graph import StateGraph; from langgraph.checkpoint.postgres import PostgresSaver; print('OK')"

Then work through the official LangGraph tutorials in order:

1. Quick Start — Build a simple chat agent with tools
2. State Management — Understand reducer annotations and state schemas
3. Human-in-the-Loop — Implement interrupt/resume patterns
4. Multi-Agent — Build supervisor-worker and map-reduce architectures
5. Production Deployment — Deploy with FastAPI, PostgresSaver, and LangSmith

The official documentation at langchain-ai.github.io/langgraph and the LangGraph GitHub repository at github.com/langchain-ai/langgraph are the best resources for learning. The LangGraph Discord community is active and responsive for troubleshooting.

---

Next in the Open-Source AI Tools Mastery series: AutoGen