LangGraph StateGraph: Beginner's Guide to AI Agent Workflows 2026

Why LangGraph Exists: The Limits of Linear Chains

A LangChain LCEL chain is a DAG — a Directed Acyclic Graph. Data flows from left to right; each component executes exactly once; there are no cycles. For simple pipelines (user query → retrieve context → generate answer), this is perfectly adequate.

The limitation emerges when your agent needs to make decisions mid-execution:

• Call a search tool. If the results are insufficient, call a different tool.
• Generate a code solution. If it fails validation, revise it.
• Attempt to answer a question. If confidence is low, retrieve more context.
• Process a request. If it requires human approval, pause and wait.

All of these require loops — executing the same node multiple times with updated state based on previous results. DAGs cannot express this. LangGraph's StateGraph can, because it supports directed graphs with cycles.

The mental model: a LangChain chain is a conveyor belt (linear, one pass). A LangGraph StateGraph is a flowchart (branches, loops, conditional paths). Both are appropriate for different problems. The complexity of your agent's reasoning determines which you need.

The 3 StateGraph Primitives

Primitive 1: State

State is a TypedDict (or Pydantic model) that defines the data structure shared between all nodes. Every node receives the current state as input and returns a dict of updates. LangGraph merges these updates using reducers.

The default reducer is replacement: if node A returns {"answer": "42"}, the state's answer field becomes "42", replacing any previous value. For fields that should accumulate across multiple node calls — like a list of messages or tool calls — use Annotated[list, operator.add] as the type, which tells LangGraph to concatenate rather than replace.

Primitive 2: Nodes

Nodes are Python functions (or LCEL chains) that take the current state as input and return a dict of state updates. A node that calls an LLM reads the current messages from state, calls the LLM, and returns the LLM's response as an update to the messages field. A node that calls a tool reads the tool name and arguments from state, executes the tool, and returns the result.

Nodes should be pure in intent: each node has one clear responsibility. An LLM decision node decides what to do next. A tool execution node executes one specific tool. A validation node checks results against a schema. Mixing concerns makes debugging and testing significantly harder.

Primitive 3: Edges

Edges define how control flows between nodes. There are two types:

Direct edges: graph.add_edge("node_a", "node_b") — control always flows from node_a to node_b. Use for deterministic sequential steps.

Conditional edges: graph.add_conditional_edges("node_a", routing_function) — control flows to the node returned by routing_function(state). The routing function examines the current state and returns a node name (or END). This is the agent's decision mechanism.

Conditional Edges: The Agent's Decision Engine

Conditional edges are what makes LangGraph agents capable of genuine reasoning loops. The routing function is a plain Python function that inspects the current state and returns the name of the next node. The LLM's last message determines the routing decision.

The canonical pattern for a tool-using agent:

1. LLM node: given the current state (messages), call the LLM. It either returns a final answer or specifies a tool call.
2. Router function: if the LLM's last message contains a tool call, return "tools". Otherwise, return END.
3. Tool node: execute the specified tool, add the result to state as a tool message.
4. Edge from tool node back to LLM node — creating the loop.

This four-element pattern (LLM node → conditional router → tool node → back to LLM) is the foundation of every ReAct-style agent in LangGraph. The agent loops until the LLM decides it has enough information to produce a final answer without calling another tool.

Checkpointing: Persistent State Across Turns

LangGraph's checkpointing system saves the complete graph state at every step. This enables two critical production capabilities:

Human-in-the-loop: Compile the graph with interrupt_before=["approval_node"]. The graph executes until it reaches the approval node, then pauses and saves its state. A human reviews the pending action, approves or rejects it, and the graph resumes from exactly the checkpoint — no state is lost.

Multi-turn conversations: Each user message continues the same graph execution by passing the same thread_id in the config. The checkpointer loads the saved state for that thread, appends the new user message, and continues execution. This is how agents maintain conversation context across multiple HTTP requests in production.

For development: use MemorySaver (in-memory, ephemeral). For production: use PostgresSaver from langgraph-checkpoint-postgres for durable state that survives pod restarts.

Complete Code: Build Your First LangGraph Agent

# pip install langgraph langchain-openai

import operator
from typing import Annotated, TypedDict, Literal
from langchain_openai import ChatOpenAI
from langchain_core.messages import HumanMessage, BaseMessage, ToolMessage
from langchain_core.tools import tool
from langgraph.graph import StateGraph, END
from langgraph.checkpoint.memory import MemorySaver

# ===================================================================
# Step 1: Define State
# WHY TypedDict + Annotated: explicit schema prevents state mutation bugs.
# operator.add on messages means each node APPENDS, not replaces the list.
# ===================================================================

class AgentState(TypedDict):
    messages: Annotated[list[BaseMessage], operator.add]
    # Add more fields as your agent grows:
    # retrieved_context: str
    # tool_results: Annotated[list[dict], operator.add]
    # approval_status: str

# ===================================================================
# Step 2: Define Tools
# WHY @tool decorator: LangChain wraps the function with metadata
# (name, description, input schema) that the LLM uses to decide
# when and how to call it. Docstring becomes the tool description.
# ===================================================================

@tool
def search_knowledge_base(query: str) -> str:
    """Search the DevOps knowledge base for answers to technical questions."""
    # In production: replace with real vector store retrieval
    # See embeddings-explained-ai-engineers-2026.html for the full implementation
    mock_results = {
        "kubernetes": "Kubernetes is a container orchestration platform that automates deployment, scaling, and management of containerised applications.",
        "helm": "Helm is the package manager for Kubernetes. Charts bundle K8s YAML manifests into reusable, versioned packages.",
        "gitops": "GitOps is a practice where Git is the single source of truth for infrastructure and application config. ArgoCD and Flux implement GitOps for Kubernetes.",
    }
    for keyword, result in mock_results.items():
        if keyword.lower() in query.lower():
            return result
    return "No specific result found. Recommend consulting official documentation."

@tool
def calculate_resource_cost(cpu_cores: float, memory_gb: float, hours: int) -> str:
    """Calculate the estimated monthly cost for a Kubernetes workload."""
    # Simplified cost model (USD, AWS us-east-1 approximate pricing)
    cpu_cost = cpu_cores * 0.048 * hours
    memory_cost = memory_gb * 0.006 * hours
    total = cpu_cost + memory_cost
    return f"Estimated cost: ${total:.2f}/month for {cpu_cores} vCPU, {memory_gb}GB RAM, {hours}h/month"

tools = [search_knowledge_base, calculate_resource_cost]

# ===================================================================
# Step 3: Define Nodes
# Each node is a function: AgentState -> dict (state updates only)
# ===================================================================

llm = ChatOpenAI(model="gpt-4o-mini", temperature=0.0)
llm_with_tools = llm.bind_tools(tools)   # LLM knows about available tools

def llm_node(state: AgentState) -> dict:
    """
    The reasoning node. The LLM sees all messages and decides:
    (a) call a tool, or (b) return a final answer.
    Returns a dict — LangGraph merges it into state using the reducer.
    """
    response = llm_with_tools.invoke(state["messages"])
    return {"messages": [response]}  # operator.add appends this to the list

def tool_node(state: AgentState) -> dict:
    """
    Execute the tool calls specified in the last LLM message.
    Returns tool results as ToolMessages appended to state.
    """
    last_message = state["messages"][-1]
    tool_messages = []

    for tool_call in last_message.tool_calls:
        # Find the matching tool function
        tool_fn = next((t for t in tools if t.name == tool_call["name"]), None)
        if tool_fn is None:
            result = f"Error: tool '{tool_call['name']}' not found"
        else:
            try:
                result = tool_fn.invoke(tool_call["args"])
            except Exception as e:
                result = f"Error executing {tool_call['name']}: {str(e)}"

        tool_messages.append(ToolMessage(
            content=str(result),
            tool_call_id=tool_call["id"],
        ))

    return {"messages": tool_messages}

# ===================================================================
# Step 4: Define the Router (Conditional Edge function)
# This function examines state and returns the next node name.
# WHY: the LLM's tool_calls attribute signals its decision.
# ===================================================================

def should_continue(state: AgentState) -> Literal["tools", "__end__"]:
    """
    Router function for the conditional edge.
    If the last LLM message has tool_calls → route to tool node.
    Otherwise → END (LLM has produced its final answer).
    """
    last_message = state["messages"][-1]
    if hasattr(last_message, "tool_calls") and last_message.tool_calls:
        return "tools"
    return END

# ===================================================================
# Step 5: Build and Compile the Graph
# WHY compile(): validates graph structure (no orphan nodes, etc.)
# and returns the executable CompiledGraph.
# ===================================================================

# MemorySaver: development checkpointer (in-memory, not persistent)
# Replace with PostgresSaver for production
checkpointer = MemorySaver()

builder = StateGraph(AgentState)

# Add nodes
builder.add_node("llm", llm_node)
builder.add_node("tools", tool_node)

# Set entry point
builder.set_entry_point("llm")

# Direct edge: after tools executes, always go back to LLM
builder.add_edge("tools", "llm")

# Conditional edge: after LLM executes, route based on should_continue()
builder.add_conditional_edges(
    "llm",            # source node
    should_continue,  # routing function
    # Optional explicit mapping (improves graph visualisation):
    {"tools": "tools", END: END}
)

# Compile the graph with checkpointer
agent = builder.compile(checkpointer=checkpointer)

print("Graph compiled successfully")
print(agent.get_graph().draw_ascii())  # ASCII visualisation in terminal

# ===================================================================
# Step 6: Invoke the Agent
# thread_id in config = session identifier for checkpointing.
# Same thread_id on the next call = continues the same conversation.
# ===================================================================

config = {"configurable": {"thread_id": "workshop-session-01"}}

# Turn 1
result = agent.invoke(
    {"messages": [HumanMessage(content="What is Helm and how does it relate to Kubernetes?")]},
    config=config
)
print("\n=== Turn 1 ===")
print("Answer:", result["messages"][-1].content)
print("Total messages in state:", len(result["messages"]))

# Turn 2 — same thread_id, state is automatically loaded from checkpoint
result = agent.invoke(
    {"messages": [HumanMessage(content="Now calculate the cost for 2 vCPU, 4GB RAM, 720 hours/month")]},
    config=config
)
print("\n=== Turn 2 (same session) ===")
print("Answer:", result["messages"][-1].content)
print("Total messages in state:", len(result["messages"]))  # Growing list

# ===================================================================
# Step 7: Async streaming (for FastAPI integration)
# ===================================================================

import asyncio

async def stream_agent():
    async for chunk in agent.astream(
        {"messages": [HumanMessage(content="What is GitOps?")]},
        config={"configurable": {"thread_id": "stream-session-01"}}
    ):
        if "llm" in chunk:
            # Streaming tokens from the LLM node
            last_msg = chunk["llm"]["messages"][-1]
            if hasattr(last_msg, "content") and last_msg.content:
                print(last_msg.content, end="", flush=True)
    print()

# asyncio.run(stream_agent())

# ===================================================================
# Understanding the execution trace for "What is Helm?"
# ===================================================================
# 1. Entry: START → llm_node
# 2. llm_node: LLM receives [HumanMessage("What is Helm...")]
#              LLM decides to call search_knowledge_base("helm")
#              Returns: AIMessage with tool_calls=[{name:"search_knowledge_base",...}]
#              State: messages = [HumanMessage, AIMessage(tool_call)]
#
# 3. should_continue: last message has tool_calls → return "tools"
#
# 4. tool_node: executes search_knowledge_base("helm")
#              Returns: ToolMessage("Helm is the package manager...")
#              State: messages = [HumanMessage, AIMessage(tool_call), ToolMessage]
#
# 5. Direct edge: tools → llm
#
# 6. llm_node: LLM receives all 3 messages including tool result
#              Generates final answer (no more tool_calls needed)
#              Returns: AIMessage("Helm is the package manager for Kubernetes...")
#              State: messages = [HumanMessage, AIMessage(tc), ToolMessage, AIMessage(answer)]
#
# 7. should_continue: last message has no tool_calls → return END
#
# 8. Graph complete. Final answer in state["messages"][-1].content

Conclusion: StateGraph Is the Foundation

Every advanced LangGraph pattern in the workshop — the supervisor multi-agent system, the human-in-the-loop approval gate, the self-correcting code generator — is built on the three primitives covered in this post. State, nodes, and conditional edges. Once you understand how a single tool-using agent loops to completion using a conditional edge, the step to a multi-agent supervisor is straightforward: the supervisor's routing decision becomes a conditional edge that routes to one of several worker agents instead of one of several tools.

Day 1 of the Agentic AI Workshop is entirely focused on building StateGraph agents from scratch — starting with exactly the pattern in this post and progressively adding complexity: multiple tools, error handling, checkpointing, human approval gates. By end of Day 1, every participant has shipped a working multi-turn agent to a FastAPI endpoint. The fundamentals covered here are what make that possible in a single day.