StateGraph: Your First Graph

Capsule overview

StateGraph is LangGraph's foundational class. It's the container where you define your workflow's state, add nodes (functions), connect them with edges (connections), compile, and run. Every graph in LangGraph starts with StateGraph.

In the previous capsule you saw the big picture: why LangGraph exists, which create_agent limitations it solves, and how a graph is simply a flowchart with nodes and edges. Now you're going to build it. You'll create your first working graph, understand how typed state works with TypedDict and Annotated, discover why reducers are the most important piece (and the #1 source of bugs), and visualize your graph with draw_mermaid_png().

The pieces of the puzzle

Before writing any code, these are the 5 pieces you need to build any graph:

Defining state: TypedDict + Annotated

State is the heart of your graph. It's a typed dictionary that all nodes share: each node receives it, reads it, and returns updates.

The problem without reducers

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

class State(TypedDict):
    messages: list[str]

def step_a(state: State) -> dict:
    return {"messages": ["Message from A"]}

def step_b(state: State) -> dict:
    return {"messages": ["Message from B"]}

graph_builder = StateGraph(State)
graph_builder.add_node("step_a", step_a)
graph_builder.add_node("step_b", step_b)
graph_builder.add_edge(START, "step_a")
graph_builder.add_edge("step_a", "step_b")
graph_builder.add_edge("step_b", END)

graph = graph_builder.compile()
result = graph.invoke({"messages": ["Initial message"]})
print(result)
# Output: {"messages": ["Message from B"]}

A's message vanished. And so did the initial message. Each node replaced the whole list with its own value. step_b wrote ["Message from B"] and wiped out everything before it.

This is the #1 beginner mistake. Without a reducer, every node overwrites the entire field.

The fix: Annotated + operator.add

from typing import TypedDict, Annotated
import operator
from langgraph.graph import StateGraph, START, END

class State(TypedDict):
    messages: Annotated[list[str], operator.add]

def step_a(state: State) -> dict:
    return {"messages": ["Message from A"]}

def step_b(state: State) -> dict:
    return {"messages": ["Message from B"]}

graph_builder = StateGraph(State)
graph_builder.add_node("step_a", step_a)
graph_builder.add_node("step_b", step_b)
graph_builder.add_edge(START, "step_a")
graph_builder.add_edge("step_a", "step_b")
graph_builder.add_edge("step_b", END)

graph = graph_builder.compile()
result = graph.invoke({"messages": ["Initial message"]})
print(result)
# Output: {"messages": ["Initial message", "Message from A", "Message from B"]}

Now all three messages are there. operator.add tells LangGraph: "when a node returns a value for messages, don't replace — concatenate it with what's already there."

How it works under the hood

Without a reducer (replace):
  Current state: {"messages": ["Initial", "A"]}
  Node returns:  {"messages": ["B"]}
  Result:        {"messages": ["B"]}              ← everything was lost

With operator.add (accumulate):
  Current state: {"messages": ["Initial", "A"]}
  Node returns:  {"messages": ["B"]}
  Result:        {"messages": ["Initial", "A", "B"]}  ← it accumulated

The rule is simple:

• Without Annotated: the returned value replaces the existing value
• With Annotated[type, operator.add]: the returned value is concatenated to the existing one

Fields with and without a reducer in the same state

You can mix fields that accumulate with fields that replace:

from typing import TypedDict, Annotated
import operator

class State(TypedDict):
    messages: Annotated[list[str], operator.add]
    step_count: int
    current_phase: str

• messages uses operator.add → accumulates
• step_count with no reducer → gets replaced (last value wins)
• current_phase with no reducer → gets replaced

Design your state by asking: "does this field need to accumulate history, or do I only need the latest value?"

Creating the graph and connecting nodes

StateGraph takes the state class and gives you a builder to add nodes and edges:

from langgraph.graph import StateGraph, START, END

graph_builder = StateGraph(State)

The builder has three main operations:

START and END are virtual nodes marking entry and exit. You always connect START to your first real node, and your last node to END. If you forget either one, the error shows up at compile time.

Your first complete graph

Now let's put all the pieces together:

from typing import TypedDict, Annotated
import operator
from langgraph.graph import StateGraph, START, END

class State(TypedDict):
    messages: Annotated[list[str], operator.add]
    step_count: int

def greet(state: State) -> dict:
    return {
        "messages": ["Hi! I'm your first graph."],
        "step_count": state["step_count"] + 1
    }

graph_builder = StateGraph(State)
graph_builder.add_node("greet", greet)
graph_builder.add_edge(START, "greet")
graph_builder.add_edge("greet", END)

graph = graph_builder.compile()
result = graph.invoke({"messages": [], "step_count": 0})
print(result)
# {"messages": ["Hi! I'm your first graph."], "step_count": 1}

Step by step:

1. You define the state — State with messages (accumulates) and step_count (replaces)
2. You define the function — greet receives the state, returns the updates
3. You create the builder — StateGraph(State) with your state class
4. You add the node — add_node("greet", greet) registers the function
5. You connect the edges — START → greet → END
6. You compile — graph_builder.compile() produces a runnable graph
7. You run it — graph.invoke(...) passes the initial state through the graph

A node doesn't need to return every field of the state — only the ones it wants to update. Fields left out keep their value.

A graph with multiple nodes

A single node isn't very useful. Let's add more:

from typing import TypedDict, Annotated
import operator
from langgraph.graph import StateGraph, START, END

class State(TypedDict):
    messages: Annotated[list[str], operator.add]
    step_count: int

def collect_info(state: State) -> dict:
    return {
        "messages": ["Step 1: Collecting information..."],
        "step_count": state["step_count"] + 1
    }

def analyze(state: State) -> dict:
    return {
        "messages": [f"Step 2: Analyzing ({state['step_count']} previous steps)..."],
        "step_count": state["step_count"] + 1
    }

def respond(state: State) -> dict:
    return {
        "messages": [f"Step 3: Answer generated after {state['step_count']} steps."],
        "step_count": state["step_count"] + 1
    }

graph_builder = StateGraph(State)
graph_builder.add_node("collect", collect_info)
graph_builder.add_node("analyze", analyze)
graph_builder.add_node("respond", respond)

graph_builder.add_edge(START, "collect")
graph_builder.add_edge("collect", "analyze")
graph_builder.add_edge("analyze", "respond")
graph_builder.add_edge("respond", END)

graph = graph_builder.compile()
result = graph.invoke({"messages": [], "step_count": 0})

for msg in result["messages"]:
    print(msg)
# Step 1: Collecting information...
# Step 2: Analyzing (1 previous steps)...
# Step 3: Answer generated after 2 steps.

print(f"Total steps: {result['step_count']}")
# Total steps: 3

The flow is: START → collect → analyze → respond → END. Each node reads the current state, adds its message, bumps the counter, and passes it on.

Notice how messages accumulates (thanks to operator.add) but step_count gets replaced every time (the final value is 3, not the sum of all the increments).

Execution: invoke vs stream

The compiled graph has two ways to run:

invoke — run it all

result = graph.invoke({"messages": [], "step_count": 0})
print(result)
# Returns the complete final state

stream — step by step

for step in graph.stream({"messages": [], "step_count": 0}):
    print(step)
    print("---")

# Output:
# {"collect": {"messages": ["Step 1: Collecting information..."], "step_count": 1}}
# ---
# {"analyze": {"messages": ["Step 2: Analyzing (1 previous steps)..."], "step_count": 2}}
# ---
# {"respond": {"messages": ["Step 3: Answer generated after 2 steps."], "step_count": 3}}
# ---

stream shows you what each node produced individually. Each chunk is a dict where the key is the node name and the value is what that node returned. Useful for debugging and for giving progressive feedback to the user.

Visualization: draw_mermaid_png()

draw_mermaid_png() isn't optional — it's your main debugging tool. Before running a graph, visualize it to confirm the flow is right:

from IPython.display import Image, display

display(Image(graph.get_graph().draw_mermaid_png()))

If you're outside a notebook, you can save the image:

png_data = graph.get_graph().draw_mermaid_png()
with open("my_graph.png", "wb") as f:
    f.write(png_data)

Or use the Mermaid format as text (no external dependencies):

print(graph.get_graph().draw_mermaid())
# %%{init: {'flowchart': {'curve': 'linear'}}}%%
# graph TD;
#     __start__([__start__]):::first
#     collect(collect)
#     analyze(analyze)
#     respond(respond)
#     __end__([__end__]):::last
#     __start__ --> collect;
#     collect --> analyze;
#     analyze --> respond;
#     respond --> __end__;

You can copy that output and paste it into mermaid.live to view it in your browser.

Get used to this flow: draw first, code second. If your visualized graph doesn't have the flow you expect, the code won't either.

The bridge from create_agent

Remember: in create_agent, the framework controlled the flow. Now you define every step:

create_agent is still the right tool for 80% of cases. But when you need a flow the ReAct loop can't express, StateGraph is ready.

Troubleshooting

Problem 1: "Messages disappear between nodes"

Symptom: Your message list only has the last message, not the full history. Cause: You're not using Annotated with operator.add on the messages field. Fix:

# Wrong — each node replaces the whole list
class State(TypedDict):
    messages: list[str]

# Right — each node appends to the existing list
class State(TypedDict):
    messages: Annotated[list[str], operator.add]

Problem 2: "ValueError at compile time — unreachable node"

Symptom: A compile error saying a node isn't reachable from START. Cause: There's a node with no path from START. You probably forgot an edge. Fix: Check that all your nodes are connected. Use draw_mermaid() before compiling to spot orphan nodes.

Problem 3: "The graph compiles but produces no output"

Symptom: graph.invoke() returns the initial state unchanged. Cause: You forgot to connect START to your first node. Fix:

graph_builder.add_edge(START, "greet")  # Don't forget this line
graph_builder.add_edge("greet", END)

Problem 4: "draw_mermaid_png() throws an error"

Symptom: An error when trying to visualize with draw_mermaid_png(). Cause: It needs an internet connection or the pyppeteer dependency. Fix: Use draw_mermaid() as an alternative (it produces Mermaid text with no dependencies). Copy the output and paste it into mermaid.live to view it.

Exercises

Exercise 1: Single-node graph (Easy)

Create a graph with a single node hello that adds the message "Hello, LangGraph!" to the state. The state must have a messages field with a reducer. Run the graph with an initial message and check that both messages are in the result.

from typing import TypedDict, Annotated
import operator
from langgraph.graph import StateGraph, START, END

class State(TypedDict):
    messages: Annotated[list[str], operator.add]

def hello(state: State) -> dict:
    return {"messages": ["Hello, LangGraph!"]}

graph_builder = StateGraph(State)
graph_builder.add_node("hello", hello)
graph_builder.add_edge(START, "hello")
graph_builder.add_edge("hello", END)

graph = graph_builder.compile()
result = graph.invoke({"messages": ["Initial message"]})
print(result)
# {"messages": ["Initial message", "Hello, LangGraph!"]}

Exercise 2: Demonstrate the difference with and without a reducer (Easy)

Create two identical graphs with two nodes in sequence. In one, the messages field uses operator.add. In the other, it doesn't. Run both with an initial message and compare the results.

from typing import TypedDict, Annotated
import operator
from langgraph.graph import StateGraph, START, END

class StateNoReducer(TypedDict):
    messages: list[str]

class StateWithReducer(TypedDict):
    messages: Annotated[list[str], operator.add]

def step_a(state) -> dict:
    return {"messages": ["A"]}

def step_b(state) -> dict:
    return {"messages": ["B"]}

def build_graph(state_class):
    builder = StateGraph(state_class)
    builder.add_node("a", step_a)
    builder.add_node("b", step_b)
    builder.add_edge(START, "a")
    builder.add_edge("a", "b")
    builder.add_edge("b", END)
    return builder.compile()

result_no = build_graph(StateNoReducer).invoke({"messages": ["Initial"]})
print(f"Without reducer: {result_no['messages']}")
# Without reducer: ['B']

result_yes = build_graph(StateWithReducer).invoke({"messages": ["Initial"]})
print(f"With reducer: {result_yes['messages']}")
# With reducer: ['Initial', 'A', 'B']

Exercise 3: 3-step pipeline with stream (Medium)

Create a graph with three nodes: extract, transform, load (the ETL pattern). Each node adds a message describing its action and updates a phase field. Use graph.stream() to see the output of each node.

from typing import TypedDict, Annotated
import operator
from langgraph.graph import StateGraph, START, END

class State(TypedDict):
    messages: Annotated[list[str], operator.add]
    phase: str
    record_count: int

def extract(state: State) -> dict:
    return {
        "messages": ["Extracting data from the source..."],
        "phase": "extract",
        "record_count": 150
    }

def transform(state: State) -> dict:
    cleaned = int(state["record_count"] * 0.9)
    return {
        "messages": [f"Transforming {state['record_count']} records → {cleaned} valid"],
        "phase": "transform",
        "record_count": cleaned
    }

def load(state: State) -> dict:
    return {
        "messages": [f"Loading {state['record_count']} records into the destination. Done!"],
        "phase": "load"
    }

graph_builder = StateGraph(State)
graph_builder.add_node("extract", extract)
graph_builder.add_node("transform", transform)
graph_builder.add_node("load", load)
graph_builder.add_edge(START, "extract")
graph_builder.add_edge("extract", "transform")
graph_builder.add_edge("transform", "load")
graph_builder.add_edge("load", END)

graph = graph_builder.compile()

for step in graph.stream({"messages": [], "phase": "pending", "record_count": 0}):
    for node_name, output in step.items():
        print(f"[{node_name}] phase={output.get('phase', '?')} | {output.get('messages', [])}")

# Output:
# [extract] phase=extract | ['Extracting data from the source...']
# [transform] phase=transform | ['Transforming 150 records → 135 valid']
# [load] phase=load | ['Loading 135 records into the destination. Done!']

Exercise 4: Graph with mixed state and visualization (Medium)

Create a graph that simulates form validation. The state has: messages (with a reducer), errors (with a reducer), and is_valid (a boolean without a reducer). One node validates, another reports. Visualize the graph with draw_mermaid().

from typing import TypedDict, Annotated
import operator
from langgraph.graph import StateGraph, START, END

class FormState(TypedDict):
    messages: Annotated[list[str], operator.add]
    errors: Annotated[list[str], operator.add]
    is_valid: bool

def validate(state: FormState) -> dict:
    found_errors = ["Field 'email' is empty", "Field 'name' is too short"]
    return {
        "messages": ["Validation ran"],
        "errors": found_errors,
        "is_valid": len(found_errors) == 0
    }

def report(state: FormState) -> dict:
    if state["is_valid"]:
        return {"messages": ["Form is valid. Submitting..."]}
    return {"messages": [f"Form is invalid. {len(state['errors'])} errors found."]}

graph_builder = StateGraph(FormState)
graph_builder.add_node("validate", validate)
graph_builder.add_node("report", report)
graph_builder.add_edge(START, "validate")
graph_builder.add_edge("validate", "report")
graph_builder.add_edge("report", END)

graph = graph_builder.compile()

print(graph.get_graph().draw_mermaid())
# graph TD;
#     __start__ --> validate;
#     validate --> report;
#     report --> __end__;

result = graph.invoke({"messages": [], "errors": [], "is_valid": False})
print(f"Messages: {result['messages']}")
# Messages: ['Validation ran', 'Form is invalid. 2 errors found.']
print(f"Errors: {result['errors']}")
# Errors: ["Field 'email' is empty", "Field 'name' is too short"]

Exercise 5: State design — choosing reducers (Advanced)

Design the state for an order-processing system. Requirements: accumulate an action log, track the order's current status, accumulate items, keep the total (latest calculation), accumulate warnings. Create the TypedDict, 3 nodes, and run the graph. Justify your reducer decisions.

from typing import TypedDict, Annotated
import operator
from langgraph.graph import StateGraph, START, END

class OrderState(TypedDict):
    action_log: Annotated[list[str], operator.add]
    status: str
    items: Annotated[list[dict], operator.add]
    total: float
    warnings: Annotated[list[str], operator.add]

def receive_order(state: OrderState) -> dict:
    items = [
        {"name": "Laptop", "price": 999.99, "qty": 1},
        {"name": "Mouse", "price": 29.99, "qty": 2},
    ]
    return {
        "action_log": ["Order received with 2 products"],
        "status": "received",
        "items": items,
    }

def validate_order(state: OrderState) -> dict:
    warnings = []
    total = sum(item["price"] * item["qty"] for item in state["items"])
    if total > 500:
        warnings.append(f"High-value order: ${total:.2f} — needs approval")
    return {
        "action_log": [f"Validation complete. Total: ${total:.2f}"],
        "status": "validated",
        "total": total,
        "warnings": warnings,
    }

def confirm_order(state: OrderState) -> dict:
    return {
        "action_log": [f"Order confirmed. Final total: ${state['total']:.2f}"],
        "status": "confirmed",
    }

graph_builder = StateGraph(OrderState)
graph_builder.add_node("receive", receive_order)
graph_builder.add_node("validate", validate_order)
graph_builder.add_node("confirm", confirm_order)
graph_builder.add_edge(START, "receive")
graph_builder.add_edge("receive", "validate")
graph_builder.add_edge("validate", "confirm")
graph_builder.add_edge("confirm", END)

graph = graph_builder.compile()
result = graph.invoke({
    "action_log": [], "status": "pending", "items": [],
    "total": 0.0, "warnings": []
})

print(f"Status: {result['status']}")
# Status: confirmed
print(f"Total: ${result['total']:.2f}")
# Total: $1059.97
print(f"Warnings: {result['warnings']}")
# Warnings: ['High-value order: $1059.97 — needs approval']
for entry in result['action_log']:
    print(f"  → {entry}")
# → Order received with 2 products
# → Validation complete. Total: $1059.97
# → Order confirmed. Final total: $1059.97

Exercise 6: invoke vs stream with 4 nodes (Challenge)

Create a graph with 4 math nodes in sequence: multiply (*2), add_ten (+10), square (^2), report. Run it with invoke (print the final result) and with stream (print step by step with a counter of executed nodes).

from typing import TypedDict, Annotated
import operator
from langgraph.graph import StateGraph, START, END

class State(TypedDict):
    messages: Annotated[list[str], operator.add]
    value: int

def multiply(state: State) -> dict:
    new_val = state["value"] * 2
    return {"messages": [f"multiply: {state['value']} → {new_val}"], "value": new_val}

def add_ten(state: State) -> dict:
    new_val = state["value"] + 10
    return {"messages": [f"add_ten: {state['value']} → {new_val}"], "value": new_val}

def square(state: State) -> dict:
    new_val = state["value"] ** 2
    return {"messages": [f"square: {state['value']} → {new_val}"], "value": new_val}

def report(state: State) -> dict:
    return {"messages": [f"Final result: {state['value']}"]}

graph_builder = StateGraph(State)
graph_builder.add_node("multiply", multiply)
graph_builder.add_node("add_ten", add_ten)
graph_builder.add_node("square", square)
graph_builder.add_node("report", report)
graph_builder.add_edge(START, "multiply")
graph_builder.add_edge("multiply", "add_ten")
graph_builder.add_edge("add_ten", "square")
graph_builder.add_edge("square", "report")
graph_builder.add_edge("report", END)

graph = graph_builder.compile()

print("=== INVOKE ===")
result = graph.invoke({"messages": [], "value": 5})
print(f"Value: {result['value']}")
print(f"Messages: {result['messages']}")
# Value: 400
# Messages: ['multiply: 5 → 10', 'add_ten: 10 → 20', 'square: 20 → 400', 'Final result: 400']

print("\n=== STREAM ===")
node_count = 0
for step in graph.stream({"messages": [], "value": 5}):
    for node_name, output in step.items():
        node_count += 1
        print(f"  Node #{node_count} [{node_name}]: value={output.get('value', '—')}")
print(f"Total nodes executed: {node_count}")
# Node #1 [multiply]: value=10
# Node #2 [add_ten]: value=20
# Node #3 [square]: value=400
# Node #4 [report]: value=—
# Total nodes executed: 4

Summary

In this capsule you learned:

• StateGraph is LangGraph's foundational class — the container where you define state, nodes, edges, compile, and run
• State is defined with TypedDict. Each node receives the full state and returns only the fields it wants to update
• Without Annotated + reducer: each node replaces the field's value. With Annotated[type, operator.add]: each node appends to the existing value
• Reducers are the #1 source of bugs for beginners — if your messages "disappear", check that you have operator.add
• START and END are special nodes marking the graph's entry and exit
• graph_builder.compile() turns the definition into a runnable graph
• invoke runs everything and returns the final state. stream shows each node's output individually
• draw_mermaid_png() is your main debugging tool — draw first, code second
• In create_agent, the framework controlled the flow. Now you define every step, every connection — more power, more responsibility

Next capsule: Nodes: Functions That Transform State — you'll learn to create nodes that call the model, nodes that run tools, and patterns for designing effective node functions.

Additional resources

1. StateGraph Reference — Full reference for the StateGraph class
2. LangGraph Quickstart — Official step-by-step tutorial
3. How to define graph state — Guide to designing state with TypedDict, reducers, and Pydantic
4. How to visualize your graph — Visualization options: Mermaid, PNG, and ASCII
5. State reducers — Conceptual documentation of reducers and Annotated
6. TypedDict — Python docs — TypedDict reference for typed state
7. operator module — Python docs — Reference for operator.add and other operators
8. Annotated — Python docs — Reference for Annotated and how LangGraph uses type metadata

Module 5 — LangChain & LangGraph: From Chains to Agents