← Back to Blog
Tutorial

LangGraph Complete Guide: Build Production AI Agents with Stateful Workflows (2026)

Master LangGraph from first graph to production deployment. Complete guide covering state schemas, nodes, edges, checkpoints, human-in-the-loop, and multi-agent patterns.

· Huzaifa Tahir
LangGraphLangChainAI AgentsState MachinesPythonWorkflow Orchestration

LangGraph Complete Guide: Build Production AI Agents with Stateful Workflows (2026)

Most AI agent tutorials stop at a single LLM call with a few tools. Production systems need something different: durable state, conditional routing, retries, human approval gates, and the ability to resume a workflow hours later without losing context. LangGraph, built on top of LangChain, gives you a graph-based orchestration layer designed exactly for that.

In 2026, LangGraph has become the default choice for teams moving from prototype agents to production pipelines. This guide walks you through the mental model, core APIs, and patterns you will use on real projects, from a minimal ReAct loop to multi-agent research pipelines with checkpointing and observability.

💡

What You'll Learn

  • How LangGraph differs from plain LangChain chains and when to use each
  • Defining typed state, nodes, and conditional edges
  • Building a ReAct agent with tool routing and error recovery
  • Checkpointing, persistence, and human-in-the-loop interrupts
  • Multi-agent supervisor patterns and production deployment tips

What Is LangGraph?

LangGraph models agent workflows as directed graphs. Each node is a function (or callable) that reads and updates shared state. Edges define execution order, either fixed transitions or conditional branches based on state values.

Unlike a linear chain where step A always leads to step B, a graph lets you:

  • Loop until a condition is met (e.g., keep calling tools until the model produces a final answer)
  • Branch based on LLM output, tool results, or external signals
  • Pause execution for human review and resume from the exact checkpoint
  • Run parallel branches and merge results

The key insight is state as the single source of truth. Every node receives the current state and returns a partial update. LangGraph merges updates according to your schema, typically using reducers for list fields like message history.

LangGraph vs. LangChain Chains

AspectLangChain ChainLangGraph
Control flowLinearCyclic, conditional, parallel
StatePassed implicitlyExplicit typed schema
PersistenceManualBuilt-in checkpointing
Human approvalCustom codeNative interrupt/resume
Best forSimple pipelinesAgents, workflows, bots

If your agent needs to call tools in a loop, recover from failures, or involve humans mid-flight, LangGraph is the right layer.

Core Concepts: State, Nodes, and Edges

Defining State with TypedDict

State is typically a TypedDict or Pydantic model. Message history uses LangChain’s add_messages reducer so new messages append rather than overwrite:

from typing import Annotated, TypedDict
from langgraph.graph.message import add_messages
from langchain_core.messages import BaseMessage

class AgentState(TypedDict):
    messages: Annotated[list[BaseMessage], add_messages]
    retry_count: int

The Annotated[..., add_messages] pattern is essential for chat-style agents. Without a reducer, each node would replace the entire message list.

Nodes and the StateGraph API

A node is any callable (state) -> partial_state_update. Build the graph with StateGraph:

from langgraph.graph import StateGraph, START, END

def call_model(state: AgentState) -> dict:
    response = llm.invoke(state["messages"])
    return {"messages": [response]}

def should_continue(state: AgentState) -> str:
    last = state["messages"][-1]
    if hasattr(last, "tool_calls") and last.tool_calls:
        return "tools"
    return END

builder = StateGraph(AgentState)
builder.add_node("agent", call_model)
builder.add_node("tools", tool_node)
builder.add_edge(START, "agent")
builder.add_conditional_edges("agent", should_continue, {"tools": "tools", END: END})
builder.add_edge("tools", "agent")
graph = builder.compile()

This is the canonical ReAct loop: model → tools → model until no tool calls remain.

Building a ReAct Agent with Tools

Tool Node Setup

LangGraph provides ToolNode from langgraph.prebuilt for standard tool execution:

from langgraph.prebuilt import ToolNode
from langchain_core.tools import tool

@tool
def search_docs(query: str) -> str:
    """Search internal documentation."""
    return vector_store.similarity_search(query, k=3)

tools = [search_docs]
tool_node = ToolNode(tools)

llm_with_tools = llm.bind_tools(tools)

Wire the model node to use llm_with_tools.invoke(state["messages"]) and route tool calls through ToolNode. The conditional edge inspects the last AI message for tool_calls.

Error Handling in Tool Nodes

Production agents must survive bad tool inputs and API failures. Wrap tool execution:

def safe_tool_node(state: AgentState) -> dict:
    try:
        return tool_node.invoke(state)
    except Exception as e:
        from langchain_core.messages import ToolMessage
        error_msg = ToolMessage(
            content=f"Tool failed: {e}",
            tool_call_id=state["messages"][-1].tool_calls[0]["id"],
        )
        return {"messages": [error_msg], "retry_count": state.get("retry_count", 0) + 1}

Add a conditional edge that routes to END or a fallback node when retry_count exceeds a threshold.

Checkpointing and Persistence

Checkpointing is LangGraph’s killer feature for production. A checkpointer saves graph state after each node, enabling:

  • Conversation resume across server restarts
  • Time-travel debugging (replay from any checkpoint)
  • Human-in-the-loop approval before sensitive actions
from langgraph.checkpoint.memory import MemorySaver
# Production: use PostgresSaver or SqliteSaver

memory = MemorySaver()
graph = builder.compile(checkpointer=memory)

config = {"configurable": {"thread_id": "user-session-42"}}
result = graph.invoke({"messages": [("user", "Summarize Q3 revenue")]}, config)

Every invoke with the same thread_id continues from the last checkpoint. Swap MemorySaver for PostgresSaver in production for durability across instances.

Human-in-the-Loop with Interrupts

Mark nodes that require approval with interrupt_before or interrupt_after:

graph = builder.compile(
    checkpointer=memory,
    interrupt_before=["execute_payment"],
)

# First run stops before execute_payment
graph.invoke(input_state, config)

# Human approves in UI, then:
graph.invoke(None, config)  # resumes from interrupt

This pattern powers compliance workflows, expense approvals, and any action that must not run autonomously.

Multi-Agent Patterns

Supervisor Pattern

A supervisor node reads the task and delegates to specialized sub-agents (researcher, coder, reviewer). Each sub-agent is its own compiled subgraph or node:

def supervisor(state: AgentState) -> dict:
    decision = supervisor_llm.invoke(state["messages"])
    return {"next_agent": parse_routing(decision)}

builder.add_conditional_edges(
    "supervisor",
    lambda s: s["next_agent"],
    {"researcher": "researcher", "coder": "coder", END: END},
)

Keep sub-agents focused with narrow tool sets. The supervisor handles routing; specialists handle execution.

Map-Reduce for Parallel Research

For tasks like “compare five vendors,” fan out to parallel research nodes, then reduce:

from langgraph.graph import Send

def fan_out(state: AgentState):
    vendors = state["vendors"]
    return [Send("research_vendor", {"vendor": v}) for v in vendors]

def merge_results(state: AgentState) -> dict:
    combined = synthesize_llm.invoke(state["vendor_reports"])
    return {"messages": [combined]}

LangGraph’s Send API enables dynamic parallel branches without hardcoding node count.

Observability and Debugging

Integrate LangSmith or OpenTelemetry for trace visibility. Every node execution appears as a span with inputs, outputs, and latency. When agents misbehave in production, traces show exactly which tool returned bad data or which routing decision went wrong.

Enable verbose logging during development:

import langchain
langchain.debug = True  # logs full prompts, disable in production

Use graph.get_state(config) to inspect current state without advancing the graph, useful for admin dashboards showing “waiting for approval.”

Production Deployment Checklist

Scaling Considerations

LangGraph graphs run in-process by default. For high throughput:

  • Deploy as a FastAPI service with async invoke
  • Use a task queue (Celery, Temporal) for long-running graphs
  • Separate read-heavy checkpoint queries from write-heavy invoke paths
  • Consider LangGraph Platform for managed hosting if ops overhead is high

Common Pitfalls

Unbounded loops. Always set a max iteration count in conditional edges. A model that keeps calling tools without terminating will burn tokens and money.

State schema drift. Adding fields to state without migration breaks resumed threads. Version state schemas and test checkpoint restore.

Oversized message history. Trim or summarize messages before they exceed context windows. Use a dedicated “summarize history” node every N turns.

Missing tool_call_id on ToolMessage. LangChain requires matching IDs between AI tool calls and tool responses. Mismatches cause silent model confusion.

When to Choose LangGraph

Choose LangGraph when you need cyclic agent behavior, persistence, or human gates. Stick with simpler LangChain LCEL chains for one-shot RAG Q&A or fixed ETL-style LLM pipelines with no loops.

For teams already on LangChain, LangGraph is a natural upgrade path, same message types, tools, and model integrations, with production-grade orchestration on top.

Next Steps

Start with the prebuilt create_react_agent helper in langgraph.prebuilt, then refactor into explicit nodes once you need custom routing or interrupts. Pair LangGraph with MCP servers for external tool access, and RAG pipelines for grounded retrieval, covered in our companion guides on MCP and RAG architecture.