A Complete Guide to LangGraph [2026 Edition]

Neelam Koshiya — Sat, 01 Aug 2026 02:44:29 +0000

A Complete Guide to LangGraph [2026 Edition]

Build production-grade agentic AI systems with stateful, controllable workflows

July 2026 · 15 min read*

If you've outgrown LangChain's linear pipelines and need real control over how your AI agents think, branch, loop, and recover — LangGraph is the framework you've been waiting for. This guide covers everything: from core concepts to production patterns, with working code throughout.

What Is LangGraph and Why Does It Exist?
Core Concepts: Nodes, Edges, State
Your First LangGraph App
Conditional Edges & Branching
Cycles & Loops — The Real Power
State Management Deep Dive
Human-in-the-Loop Patterns
Multi-Agent Architectures
Persistence & Checkpointing
Streaming & Real-Time Output
LangGraph in Production
LangGraph vs Alternatives

Part 1 — What Is LangGraph and Why Does It Exist? {#part-1}

LangGraph is a framework built on top of LangChain for creating stateful, multi-step AI workflows modeled as directed graphs. It emerged from a deceptively simple problem: real-world AI tasks don't run in straight lines.

LangChain's chains are linear — they execute A → B → C and terminate. That model works for demos. It breaks down in production, where agents need to retry failed tool calls, branch on LLM decisions, pause for human review, and maintain context across sessions spanning hours or days.

The core insight: An agent is just a loop — perceive the world, decide, act, observe the result, repeat. LangGraph makes that loop explicit, observable, and controllable.

Five problems LangGraph solves

Retry on failure: When a tool call returns an error, the agent can loop back and try a different approach rather than crashing the entire chain.

Dynamic branching: An LLM classifier node can route to a billing specialist, a technical support flow, or a general handler — decided at runtime, not hard-coded at design time.

Human-in-the-loop: Execution pauses at a defined checkpoint and resumes only after a human reviews and approves — critical for regulated industries and high-stakes decisions.

Long-running sessions: State is checkpointed to a database, so an agent can pick up exactly where it left off after a crash, restart, or multi-day pause.

Multi-agent collaboration: Specialized sub-agents can be composed into a single orchestrated system with a clean state boundary between them.

Part 2 — Core Concepts: Nodes, Edges, State {#part-2}

LangGraph has three primitives. Master these and every pattern in this guide becomes readable at a glance.

The State

State is a typed Python dictionary shared across every node in the graph. Each node reads from it, performs its work, and returns a partial update. The Annotated type hint controls how updates are merged — the most important pattern being add_messages, which appends to the message list rather than replacing it.

from typing import TypedDict, Annotated
from langgraph.graph import add_messages

class AgentState(TypedDict):
    # add_messages = APPEND each update, not replace
    # Without it: each node would overwrite history
    messages: Annotated[list, add_messages]
    task: str
    result: str | None
    attempts: int
    approved: bool

Why add_messages matters: Without it, each node that returns {"messages": [new_msg]} would replace the entire history. With add_messages, the new message is appended — conversation history accumulates naturally without any extra bookkeeping.

Nodes

Nodes are plain Python functions. They receive the current state, do their work (call an LLM, execute a tool, run validation logic), and return a partial state update — only the keys that changed.

from langchain_anthropic import ChatAnthropic

llm = ChatAnthropic(model="claude-opus-4-6")

def analyst_node(state: AgentState) -> dict:
    """Analyzes the task and produces a structured plan."""
    response = llm.invoke([
        {"role": "system", "content": "You are a senior analyst. Break this task into steps."},
        {"role": "user",   "content": f"Task: {state['task']}"}
    ])
    # Return ONLY the keys that changed — LangGraph merges for you
    return {
        "messages": [response],   # appended via add_messages
        "result": response.content
    }

def validator_node(state: AgentState) -> dict:
    """Quality-checks the result and increments attempt counter."""
    response = llm.invoke([
        {"role": "system", "content": "Score this output 1–10. Reply with only a number."},
        {"role": "user",   "content": state["result"]}
    ])
    score = int(response.content.strip())
    # Only return what changed — don't touch messages or task
    return {
        "attempts": state["attempts"] + 1,
        "approved": score >= 8
    }

Edges

Edges define control flow. A normal edge always routes A → B. A conditional edge calls a routing function that returns a string key, directing execution to one of several possible next nodes.

from langgraph.graph import StateGraph, END

workflow = StateGraph(AgentState)

# Normal edge — always goes analyst → validator
workflow.add_edge("analyst", "validator")

# Conditional edge — routing function returns next node key
def route_after_validation(state: AgentState) -> str:
    if state["attempts"] >= 3:
        return END               # Hard limit: give up after 3 tries
    if state["approved"]:
        return "writer"          # Quality met, proceed
    return "analyst"             # Low quality, retry

workflow.add_conditional_edges(
    "validator",
    route_after_validation,
    # Map of return values → destination nodes
    {END: END, "writer": "writer", "analyst": "analyst"}
)

Part 3 — Your First LangGraph App {#part-3}

Theory earns its keep when you build something. Here's a complete SQL query generation agent — it translates a natural-language question into SQL, validates the query for correctness and safety, and iterates until the quality bar is met or three attempts are exhausted.

Figure: The generator-validator-revise loop — iterate until SQL quality score ≥ 8, hard limit of 3 attempts.

from typing import TypedDict, Annotated
from langgraph.graph import StateGraph, END, add_messages
from langchain_anthropic import ChatAnthropic
from langchain_core.messages import HumanMessage, AIMessage

llm       = ChatAnthropic(model="claude-opus-4-6", temperature=0.2)
judge_llm = ChatAnthropic(model="claude-haiku-4-5-20251001", temperature=0)

# ── State ──────────────────────────────────────────────────────────────
class SQLState(TypedDict):
    messages:      Annotated[list, add_messages]
    question:      str      # natural-language question from user
    schema:        str      # database schema context
    sql_query:     str      # generated SQL
    quality_score: int      # 1–10 from validator
    feedback:      str      # specific improvement hint
    iterations:    int

# ── Nodes ──────────────────────────────────────────────────────────────
def generator_node(state: SQLState) -> dict:
    """Translates a natural-language question into a SQL query."""
    prev_feedback = state.get("feedback", "")
    prompt = f"""You are an expert SQL engineer. Write a single correct SQL query.

Database schema:
{state['schema']}

Question: {state['question']}

{f"Previous attempt failed. Fix this: {prev_feedback}" if prev_feedback else ""}

Return ONLY the SQL query — no explanation, no markdown fences."""

    response = llm.invoke([HumanMessage(content=prompt)])
    iteration = state.get("iterations", 0) + 1
    return {
        "sql_query":  response.content.strip(),
        "iterations": iteration,
        "messages":   [AIMessage(content=f"Query attempt {iteration} generated")]
    }

def validator_node(state: SQLState) -> dict:
    """Checks the SQL for correctness, safety, and efficiency."""
    prompt = f"""You are a senior DBA. Review this SQL query against the schema.

Schema:
{state['schema']}

Query:
{state['sql_query']}

Score it 1–10 (10 = production-ready). Check for:
- Correctness (joins, filters match the schema)
- Safety (no DROP/DELETE without WHERE, no unbounded full-table scans)
- Efficiency (proper use of indexes, avoid SELECT *)

Format exactly:
SCORE: <number>
FEEDBACK: <one specific fix if score < 8, else "Looks good">"""

    response = judge_llm.invoke([HumanMessage(content=prompt)])
    lines    = response.content.strip().split('\n')
    score    = int(lines[0].replace('SCORE:', '').strip())
    feedback = lines[1].replace('FEEDBACK:', '').strip()
    return {"quality_score": score, "feedback": feedback}

# ── Routing function ───────────────────────────────────────────────────
def should_continue(state: SQLState) -> str:
    if state["quality_score"] >= 8:  return "approved"   # Production-ready
    if state["iterations"]    >= 3:  return "approved"   # Best effort after 3 tries
    return "revise"

# ── Build graph ────────────────────────────────────────────────────────
workflow = StateGraph(SQLState)
workflow.add_node("generator", generator_node)
workflow.add_node("validator", validator_node)

workflow.set_entry_point("generator")
workflow.add_edge("generator", "validator")
workflow.add_conditional_edges(
    "validator",
    should_continue,
    {"approved": END, "revise": "generator"}
)

app = workflow.compile()

# ── Run ────────────────────────────────────────────────────────────────
SCHEMA = """
  users(id, name, email, created_at, plan)
  orders(id, user_id, total, status, created_at)
  order_items(id, order_id, product_id, quantity, price)
"""

result = app.invoke({
    "question":      "Find the top 5 users by total spend in the last 30 days",
    "schema":        SCHEMA,
    "sql_query":     "",
    "quality_score": 0,
    "feedback":      "",
    "iterations":    0,
    "messages":      []
})
print(f"Final SQL (score {result['quality_score']}/10):\n{result['sql_query']}")

Key pattern: Use a fast, cheap model (Haiku) as the DBA validator and a capable model (Opus) as the generator. The validator's job is structured scoring — it doesn't need frontier reasoning, and this keeps costs low through multiple feedback iterations.

Part 4 — Conditional Edges & Branching {#part-4}

Conditional edges are where LangGraph implements decision-making. The routing function can be driven by LLM output, state values, business logic, or any combination. Here's a customer support router that classifies intent and dispatches to the right handler.

from typing import TypedDict, Literal
from langgraph.graph import StateGraph, END

class SupportState(TypedDict):
    user_message: str
    intent:       str
    response:     str

def classify_intent(state: SupportState) -> dict:
    result = llm.invoke(f"""Classify into ONE category:
    billing | technical | general | escalate

    Message: {state['user_message']}
    Reply with only the category word.""")
    return {"intent": result.content.strip().lower()}

# Routing function — return value must match edge map keys
def route_by_intent(state: SupportState) -> Literal["billing","technical","general","escalate"]:
    return state["intent"]

def billing_node(state):   return {"response": "Connecting to billing specialist..."}
def technical_node(state): return {"response": "Looking up technical documentation..."}
def escalate_node(state):  return {"response": "Transferring to senior agent."}
def general_node(state):
    answer = llm.invoke(f"Answer: {state['user_message']}").content
    return {"response": answer}

# Build
workflow = StateGraph(SupportState)
for name, fn in [("classifier", classify_intent), ("billing", billing_node),
                  ("technical", technical_node), ("general", general_node),
                  ("escalate", escalate_node)]:
    workflow.add_node(name, fn)

workflow.set_entry_point("classifier")
workflow.add_conditional_edges(
    "classifier", route_by_intent,
    {"billing":"billing", "technical":"technical",
     "general":"general", "escalate":"escalate"}
)
# All branches converge to END
for node in ["billing", "technical", "general", "escalate"]:
    workflow.add_edge(node, END)

support_app = workflow.compile()

Common mistake: The routing function must return a value that exactly matches one of the keys in the edge map dict. If the LLM returns "billing" but your map has "Billing", the graph raises a GraphValueError. Always .strip().lower() your LLM outputs before routing.

Part 5 — Cycles & Loops — The Real Power {#part-5}

Loops are what make LangGraph fundamentally different from every linear alternative. An agent can keep working until it achieves a goal — search until it has enough information, generate until quality is met, retry until a tool succeeds.

The pattern is always the same: add a conditional edge whose routing function can return a node that already ran. The graph's cycle detector allows this by design.

from typing import TypedDict

class ResearchState(TypedDict):
    question:       str
    search_results: list[str]
    analysis:       str
    is_sufficient:  bool
    search_count:   int

def search_node(state: ResearchState) -> dict:
    """Generate a targeted search query and execute it."""
    query = llm.invoke(f"""Generate a specific search query for: {state['question']}
Searches done so far: {state['search_count']}
Last result: {state['search_results'][-1] if state['search_results'] else 'none'}
Make this query different from previous ones.""").content

    result = execute_search(query)   # Replace with real search API
    return {
        "search_results": state["search_results"] + [result],
        "search_count":   state["search_count"] + 1
    }

def analyze_node(state: ResearchState) -> dict:
    """Determine if we have enough information to answer."""
    assessment = llm.invoke(f"""Question: {state['question']}

Research gathered ({len(state['search_results'])} searches):
{chr(10).join(state['search_results'])}

Do you have sufficient information?
Reply: SUFFICIENT or INSUFFICIENT, then your analysis.""").content

    is_sufficient = assessment.startswith("SUFFICIENT")
    analysis = assessment.split('\n', 1)[1] if '\n' in assessment else assessment
    return {"is_sufficient": is_sufficient, "analysis": analysis}

def should_search_more(state: ResearchState) -> str:
    if state["search_count"] >= 5:  return "done"   # Hard cap
    if state["is_sufficient"]:      return "done"
    return "search_more"

# Build — the loop is created by the conditional edge pointing back
workflow = StateGraph(ResearchState)
workflow.add_node("search",  search_node)
workflow.add_node("analyze", analyze_node)
workflow.set_entry_point("search")
workflow.add_edge("search", "analyze")
workflow.add_conditional_edges(
    "analyze",
    should_search_more,
    {"done": END, "search_more": "search"}  # ← this creates the loop
)

Always add a hard exit condition. Every loop must have a maximum iteration count that terminates regardless of the LLM's assessment. LLMs can get stuck in an INSUFFICIENT loop if search results are consistently irrelevant. The hard cap at 5 searches above is your circuit breaker.

Part 6 — State Management Deep Dive {#part-6}

Every state key has a reducer — a function that controls how updates from nodes are merged into the existing state. The default reducer is last-write-wins. Understanding reducers unlocks sophisticated parallel and collaborative node patterns.

from typing import TypedDict, Annotated
from langgraph.graph.message import add_messages
import operator

# ── Pattern 1: Default (last-write-wins) ──────────────────────────────
class SimpleState(TypedDict):
    status: str   # each update replaces the value
    count:  int   # each update replaces the value

# ── Pattern 2: Append-only (chat history) ─────────────────────────────
class ChatState(TypedDict):
    messages: Annotated[list, add_messages]   # each update appends

# ── Pattern 3: Custom reducer ─────────────────────────────────────────
def merge_results(existing: list, new: list) -> list:
    """Deduplicate by ID while preserving insertion order."""
    seen, merged = set(), []
    for item in existing + new:
        key = item.get("id") if isinstance(item, dict) else item
        if key not in seen:
            seen.add(key)
            merged.append(item)
    return merged

class PipelineState(TypedDict):
    results: Annotated[list, merge_results]   # smart deduplicated merge
    errors:  Annotated[list, operator.add]    # simple append
    status:  str                              # last-write-wins

# ── Pattern 4: Parallel writes converge safely ────────────────────────
# When multiple parallel nodes write to the same key simultaneously,
# LangGraph calls the reducer with all updates — no race condition.
# This is what makes map-reduce patterns safe without locks.

Reducer = thread safety for free. When you fan out to parallel nodes using the Send API and all of them write to the same key, LangGraph's reducer merges the results atomically. Use operator.add for lists, add_messages for chat history, or your own function for custom logic.

Part 7 — Human-in-the-Loop Patterns {#part-7}

This is one of LangGraph's defining capabilities. Execution halts at a defined checkpoint and resumes only after a human provides input. This is not a polling loop — it's a first-class interrupt mechanism backed by the checkpointer.

from langgraph.checkpoint.memory import MemorySaver
from langgraph.graph import StateGraph, END
from langgraph.types import interrupt, Command
from typing import TypedDict

class ApprovalState(TypedDict):
    task:           str
    plan:           str
    human_approved: bool
    final_result:   str

def planner_node(state: ApprovalState) -> dict:
    plan = llm.invoke(f"Create a detailed execution plan for: {state['task']}").content
    return {"plan": plan}

def human_review_node(state: ApprovalState) -> dict:
    # interrupt() pauses the graph here.
    # The caller sees the payload; execution resumes when Command(resume=...) is sent.
    human_input = interrupt({
        "question": "Approve this plan?",
        "plan":     state["plan"]
    })
    return {"human_approved": human_input.get("approved", False)}

def executor_node(state: ApprovalState) -> dict:
    if not state["human_approved"]:
        return {"final_result": "Task cancelled."}
    result = llm.invoke(f"Execute this plan:\n{state['plan']}").content
    return {"final_result": result}

# Checkpointer is REQUIRED — it stores state between interrupt and resume
memory = MemorySaver()

workflow = StateGraph(ApprovalState)
workflow.add_node("planner",      planner_node)
workflow.add_node("human_review", human_review_node)
workflow.add_node("executor",     executor_node)
workflow.set_entry_point("planner")
workflow.add_edge("planner",      "human_review")
workflow.add_edge("human_review", "executor")
workflow.add_edge("executor",     END)

app = workflow.compile(checkpointer=memory)

# ── Step 1: run until interrupt ────────────────────────────────────────
thread = {"configurable": {"thread_id": "deploy-001"}}
state  = app.invoke({"task": "Deploy payment service to production"}, thread)
print("Plan:", state["plan"])
print("— WAITING FOR APPROVAL —")

# ── Step 2: human reviews, then resumes ───────────────────────────────
final = app.invoke(Command(resume={"approved": True}), thread)

`interrupt_before` vs `interrupt_after`

# interrupt_before: pause BEFORE the node runs — you can edit state first
app = workflow.compile(checkpointer=memory, interrupt_before=["dangerous_action"])

# interrupt_after: pause AFTER the node runs — review output before continuing
app = workflow.compile(checkpointer=memory, interrupt_after=["code_generator"])

# Dynamic interrupt: pause only when risk logic fires inside a node
def smart_executor(state):
    result = execute_action(state)
    if result.risk_score > 0.8:
        decision = interrupt({"reason": "High-risk action", "details": result})
        if not decision["proceed"]:
            return {"status": "cancelled"}
    return {"result": result}

Checkpointer is mandatory. Human-in-the-loop requires a checkpointer — the graph's state must be persisted somewhere so it can be retrieved when the human resumes. Without a checkpointer, interrupt() raises a runtime error. Use MemorySaver in development, PostgresSaver in production.

Part 8 — Multi-Agent Architectures {#part-8}

LangGraph's composability is its production superpower. Each sub-agent is a fully compiled LangGraph app. A supervisor coordinates them by calling them as nodes — their internal complexity is invisible to the outer graph.

Figure: Supervisor pattern — a coordinator LLM decides which specialist to invoke next, each specialist reports back.

Pattern 1 — Supervisor / Orchestrator

from typing import TypedDict, Annotated
from langgraph.graph import StateGraph, END, add_messages
from langchain_core.messages import AIMessage

class SupervisorState(TypedDict):
    messages:   Annotated[list, add_messages]
    task:       str
    next_agent: str

def supervisor_node(state: SupervisorState) -> dict:
    decision = llm.invoke(f"""You are a project manager.
Task: {state['task']}
History: {state['messages']}

Which specialist acts next?
Options: researcher | writer | reviewer | FINISH
Reply with only one word.""").content.strip()
    return {"next_agent": decision}

def route_to_agent(state: SupervisorState) -> str:
    return END if state["next_agent"] == "FINISH" else state["next_agent"]

# Each specialist can itself be a compiled LangGraph app
def researcher_node(state: SupervisorState) -> dict:
    result = research_graph.invoke(state)   # Sub-graph!
    return {"messages": [AIMessage(content=f"Research: {result['analysis']}")]}

def writer_node(state: SupervisorState) -> dict:
    draft = llm.invoke(f"Write based on: {state['messages'][-1].content}").content
    return {"messages": [AIMessage(content=f"Draft: {draft}")]}

def reviewer_node(state: SupervisorState) -> dict:
    review = llm.invoke(f"Review: {state['messages'][-1].content}").content
    return {"messages": [AIMessage(content=f"Review: {review}")]}

# Build supervisor graph
g = StateGraph(SupervisorState)
for name, fn in [("supervisor",supervisor_node),("researcher",researcher_node),
                  ("writer",writer_node),("reviewer",reviewer_node)]:
    g.add_node(name, fn)

g.set_entry_point("supervisor")
g.add_conditional_edges("supervisor", route_to_agent)
for agent in ["researcher", "writer", "reviewer"]:
    g.add_edge(agent, "supervisor")    # Always report back

multi_agent_app = g.compile()

Pattern 2 — Parallel Agents (Map-Reduce)

from langgraph.types import Send
import operator

class MapReduceState(TypedDict):
    documents:    list[str]
    summaries:    Annotated[list, operator.add]  # parallel writes merged here
    final_report: str

async def summarize_doc(state: dict) -> dict:
    summary = await llm.ainvoke(f"Summarize: {state['document']}")
    return {"summaries": [summary.content]}

def reduce_node(state: MapReduceState) -> dict:
    combined = "\n\n".join(state["summaries"])
    report   = llm.invoke(f"Unified report:\n{combined}").content
    return {"final_report": report}

# fan_out returns a list of Send objects — one parallel branch per document
def fan_out(state: MapReduceState) -> list[Send]:
    return [Send("summarize", {"document": doc}) for doc in state["documents"]]

workflow = StateGraph(MapReduceState)
workflow.add_node("summarize", summarize_doc)
workflow.add_node("reduce",    reduce_node)
workflow.set_entry_point("fan_out_node")
workflow.add_conditional_edges("fan_out_node", fan_out)
workflow.add_edge("summarize", "reduce")

Part 9 — Persistence & Checkpointing {#part-9}

Checkpointing transforms stateless function calls into resumable, multi-session agents. Every graph execution is saved as a series of checkpoints — one per node execution. You can replay history, roll back to any prior state, or fork from a checkpoint.

# ── Dev: in-memory (lost on restart) ──────────────────────────────────
from langgraph.checkpoint.memory import MemorySaver
memory = MemorySaver()

# ── Prod: PostgreSQL (persistent, queryable) ───────────────────────────
from langgraph.checkpoint.postgres import PostgresSaver
from psycopg import Connection

conn   = Connection.connect("postgresql://user:pass@localhost/langgraph_db")
pg     = PostgresSaver(conn)
pg.setup()   # Creates tables on first run

# ── High-throughput: Redis ─────────────────────────────────────────────
from langgraph.checkpoint.redis import RedisSaver
redis = RedisSaver.from_conn_string("redis://localhost:6379")

# ── Compile with checkpointer ─────────────────────────────────────────
app = workflow.compile(checkpointer=pg)

# Thread IDs = isolated conversation sessions
cfg_alice = {"configurable": {"thread_id": "alice-session-1"}}
cfg_bob   = {"configurable": {"thread_id": "bob-session-1"}}

# Independent state per thread
app.invoke({"task": "Help with Python"}, cfg_alice)
app.invoke({"task": "Write a cover letter"}, cfg_bob)

# Continue alice's session later — full history preserved
app.invoke({"task": "Now explain decorators"}, cfg_alice)

# ── Inspect and rollback ───────────────────────────────────────────────
history = list(app.get_state_history(cfg_alice))
for checkpoint in history:
    print(f"Step {checkpoint.metadata['step']}: {checkpoint.values.get('task')}")

# Rollback to 3 steps ago
app.update_state(cfg_alice, history[2].values)

Thread IDs are your conversation sessions. Each unique thread_id is a completely independent state history. Use one per user session, ticket, or workflow instance. The checkpointer handles concurrent threads safely — no locking required in your application code.

Part 10 — Streaming & Real-Time Output {#part-10}

Users expect token-by-token responses. LangGraph's astream_events API gives you fine-grained control: stream individual tokens, get notified when each node starts, observe tool calls as they happen.

import asyncio, json
from fastapi import FastAPI
from fastapi.responses import StreamingResponse

async def stream_agent(user_input: str):
    async for event in app.astream_events({"task": user_input}, version="v2"):
        kind = event["event"]

        if kind == "on_chat_model_stream":
            chunk = event["data"]["chunk"]
            if chunk.content:
                print(chunk.content, end="", flush=True)   # token-by-token

        elif kind == "on_chain_start":
            print(f"\n[▶ {event['name']}]")    # node started

        elif kind == "on_tool_start":
            print(f"\n[⚙ {event['name']}({event['data']['input']})]")  # tool call

# FastAPI SSE endpoint
api = FastAPI()

@api.post("/chat/stream")
async def chat_stream(request: dict):
    async def generate():
        async for event in app.astream_events(
            {"task": request["message"]},
            config={"configurable": {"thread_id": request["session_id"]}},
            version="v2"
        ):
            if event["event"] == "on_chat_model_stream":
                token = event["data"]["chunk"].content
                if token:
                    yield f"data: {json.dumps({'token': token})}\n\n"
        yield "data: [DONE]\n\n"

    return StreamingResponse(generate(), media_type="text/event-stream")

Part 11 — LangGraph in Production {#part-11}

Production agents fail in ways that clean demos never do. Here are the two patterns that matter most: retry with backoff for transient errors, and structured observability so you know exactly what happened when something goes wrong.

Error Handling & Retries

from tenacity import retry, stop_after_attempt, wait_exponential
import logging

logger = logging.getLogger(__name__)

def resilient_llm_node(state: AgentState) -> dict:
    @retry(
        stop=stop_after_attempt(3),
        wait=wait_exponential(multiplier=1, min=2, max=10)
    )
    def call_with_retry():
        return llm.invoke(state["messages"])

    try:
        response = call_with_retry()
        return {"messages": [response], "error": None}
    except Exception as e:
        logger.error(f"Node failed after 3 retries: {e}")
        return {"messages": [], "error": str(e), "status": "failed"}

def error_router(state: AgentState) -> str:
    return "error_handler" if state.get("error") else "next_step"

Observability with LangSmith

import os
os.environ["LANGCHAIN_TRACING_V2"] = "true"
os.environ["LANGCHAIN_API_KEY"]    = "your-key"
os.environ["LANGCHAIN_PROJECT"]    = "production-agent"

# Every execution now appears in LangSmith:
# - per-node execution time, token usage, cost
# - full state at each checkpoint
# - error traces with full context

result = app.invoke(
    {"task": "user question"},
    config={
        "configurable": {"thread_id": "session-123"},
        "metadata": {
            "user_id":  "user-456",
            "product":  "support-bot",
            "version":  "2.1.0"
        }
    }
)

Deployment with LangGraph Platform

{
    "dependencies": ["./my_agent"],
    "graphs": {
        "support_agent":  "./my_agent/graph.py:app",
        "research_agent": "./my_agent/research.py:app"
    },
    "env": ".env"
}

Deploy to LangGraph Cloud with langgraph deploy, or self-host with Docker using the langchain/langgraph-api base image.

Part 12 — LangGraph vs Alternatives {#part-12}

No framework wins every scenario. Here's an honest comparison:

Feature	LangGraph	CrewAI	AutoGen	LangChain
Graph-based control flow	✅ First-class	❌	❌	⚠️ Partial
Human-in-the-loop	✅ Built-in	⚠️ Limited	✅ Manual	❌
Persistence / checkpointing	✅ Built-in	❌	❌	⚠️ Manual
Streaming	✅ First-class	⚠️ Limited	⚠️ Limited	⚠️ Basic
Multi-agent composability	✅ Sub-graphs	✅ Role-based	✅ Conversation	⚠️ Limited
Debugging / observability	✅ LangSmith	⚠️ Limited	⚠️ Limited	✅ LangSmith
Production platform	✅ LangGraph Cloud	DIY	DIY	DIY
Learning curve	Medium	Low	Low	Low

Choose LangGraph when you need fine-grained control, human oversight is a hard requirement, your workflow has non-trivial branching or loops, you need persistent multi-session state, or you're deploying at scale and need proper observability.

Choose CrewAI when you want fast prototyping with a role/task mental model and your pipeline is primarily sequential handoffs.

Quick Reference — LangGraph Cheatsheet

from typing import TypedDict, Annotated
from langgraph.graph import StateGraph, END, add_messages
from langgraph.checkpoint.memory import MemorySaver
from langchain_core.messages import AIMessage

# 1. Define state
class State(TypedDict):
    messages: Annotated[list, add_messages]

# 2. Define nodes
def my_node(state: State) -> dict:
    return {"messages": [AIMessage(content="Hello!")]}

# 3. Build graph
graph = StateGraph(State)
graph.add_node("my_node", my_node)
graph.set_entry_point("my_node")
graph.add_edge("my_node", END)

# 4. Compile (with optional checkpointer for persistence)
app = graph.compile(checkpointer=MemorySaver())

# 5. Invoke (stateless — no thread_id)
result = app.invoke({"messages": []})

# 6. Invoke with thread (persistent memory)
config = {"configurable": {"thread_id": "my-session"}}
result = app.invoke({"messages": []}, config)

# 7. Stream token-by-token
async for event in app.astream_events({"messages": []}, version="v2"):
    if event["event"] == "on_chat_model_stream":
        print(event["data"]["chunk"].content, end="")

# 8. Conditional edge pattern
def router(state: State) -> str:
    return "node_a" if condition else "node_b"

graph.add_conditional_edges("source", router, {"node_a": "node_a", "node_b": "node_b"})

Resources

Official Docs: langchain-ai.github.io/langgraph
LangGraph Academy (free course): academy.langchain.com
LangSmith (observability): smith.langchain.com
GitHub: github.com/langchain-ai/langgraph
Templates: github.com/langchain-ai/langgraph-example

DEV Community: Neelam Koshiya

A Complete Guide to LangGraph [2026 Edition]

A Complete Guide to LangGraph [2026 Edition]

Table of Contents

Part 1 — What Is LangGraph and Why Does It Exist? {#part-1}

Five problems LangGraph solves

Part 2 — Core Concepts: Nodes, Edges, State {#part-2}

The State

Nodes

Edges

Part 3 — Your First LangGraph App {#part-3}

Part 4 — Conditional Edges & Branching {#part-4}

Part 5 — Cycles & Loops — The Real Power {#part-5}

Part 6 — State Management Deep Dive {#part-6}

Part 7 — Human-in-the-Loop Patterns {#part-7}

`interrupt_before` vs `interrupt_after`

Part 8 — Multi-Agent Architectures {#part-8}

Pattern 1 — Supervisor / Orchestrator

Pattern 2 — Parallel Agents (Map-Reduce)

Part 9 — Persistence & Checkpointing {#part-9}

Part 10 — Streaming & Real-Time Output {#part-10}

Part 11 — LangGraph in Production {#part-11}

Error Handling & Retries

Observability with LangSmith

Deployment with LangGraph Platform

Part 12 — LangGraph vs Alternatives {#part-12}

Quick Reference — LangGraph Cheatsheet

Resources

DEV Community: Neelam Koshiya

A Complete Guide to LangGraph [2026 Edition]

A Complete Guide to LangGraph [2026 Edition]

Table of Contents

Part 1 — What Is LangGraph and Why Does It Exist? {#part-1}

Five problems LangGraph solves

Part 2 — Core Concepts: Nodes, Edges, State {#part-2}

The State

Nodes

Edges

Part 3 — Your First LangGraph App {#part-3}

Part 4 — Conditional Edges & Branching {#part-4}

Part 5 — Cycles & Loops — The Real Power {#part-5}

Part 6 — State Management Deep Dive {#part-6}

Part 7 — Human-in-the-Loop Patterns {#part-7}

interrupt_before vs interrupt_after

Part 8 — Multi-Agent Architectures {#part-8}

Pattern 1 — Supervisor / Orchestrator

Pattern 2 — Parallel Agents (Map-Reduce)

Part 9 — Persistence & Checkpointing {#part-9}

Part 10 — Streaming & Real-Time Output {#part-10}

Part 11 — LangGraph in Production {#part-11}

Error Handling & Retries

Observability with LangSmith

Deployment with LangGraph Platform

Part 12 — LangGraph vs Alternatives {#part-12}

Quick Reference — LangGraph Cheatsheet

Resources

`interrupt_before` vs `interrupt_after`