State, Graphs, and StateGraph
Learn graph concepts in LangGraph: StateGraph, State schema, nodes, edges, compilation, and execution flow.
State, Graphs, and StateGraph
LangGraph is built around the concept of a stateful directed graph. Understanding how StateGraph works is essential before writing any agent code.
What is a StateGraph?
StateGraph is the primary class for building LangGraph applications. It manages:
- A type schema that defines the shape of state
- A collection of nodes that process and update state
- A set of edges that define the execution topology
- Compilation that validates and freezes the graph
from langgraph.graph import StateGraph, START, END
from typing_extensions import TypedDict
class SearchState(TypedDict):
query: str
results: list[str]
status: str
builder = StateGraph(SearchState)Always use StateGraph over the basic Graph class. StateGraph provides typed state, checkpointing, and all production features. The basic Graph class is deprecated for most use cases.
State Schema
The State is a dictionary that flows through every node. It is defined using TypedDict, dataclass, or pydantic.BaseModel.
TypedDict (Recommended for Beginners)
from typing_extensions import TypedDict
from typing import List, Optional
class AgentState(TypedDict):
messages: List[str]
turn_count: int
is_complete: bool
final_answer: Optional[str]Every node in the graph receives a dict matching this schema and returns a partial dict with only the keys it wants to update.
def first_node(state: AgentState) -> dict:
# Read from state
current_turn = state["turn_count"]
# Return only the updates
return {
"messages": state["messages"] + ["Hello from node 1"],
"turn_count": current_turn + 1
# is_complete and final_answer are unchanged
}Dataclass State
from dataclasses import dataclass, field
from typing import List
@dataclass
class AgentState:
messages: List[str] = field(default_factory=list)
turn_count: int = 0
is_complete: bool = FalseDataclasses give you default values and mutable state. Use field(default_factory=...) for mutable defaults like lists.
Pydantic BaseModel State
from pydantic import BaseModel, Field
from typing import List
class AgentState(BaseModel):
messages: List[str] = Field(default_factory=list)
turn_count: int = 0
is_complete: bool = FalseUse TypedDict for prototyping (minimal boilerplate). Use BaseModel for production (validation, serialization, JSON schema). All three approaches work identically at the graph level.
How State Flows Through the Graph
Initial State β Node A β (updates state) β Node B β (updates state) β Final State
β |
ββββββββββββ (loop back to A) βββββββββββββββββββ
Each node:
- Receives the complete current state as a dict
- Processes the data (calls LLM, runs tools, etc.)
- Returns a partial dict of updates
- LangGraph merges the updates into the shared state
State Merging Rules
| Return Value | Behavior |
|---|---|
{"key": "value"} | Updates state["key"] to "value" |
{"key": state["key"] + ["new"]} | Replaces state["key"] with new list |
return None | No changes to state |
return {} | No changes to state |
State updates use a shallow merge. If state has nested dicts, returning {"nested": {"inner": 1}} replaces the entire nested key β it does not deep-merge. For deep merges, you need custom reducers (covered in the Intermediate course).
Nodes
Nodes are the processing units of a graph. A node is simply a Python function that receives state and returns updates.
Function Signature
def node_function(state: StateType) -> dict:
# Process state
# Return updates
return {"key": new_value}Node Registration
builder = StateGraph(AgentState)
builder.add_node("process", node_function)
# ^name ^function referenceNode names must be unique. Use descriptive names like "analyze_query", "search_database", "generate_response" rather than "node1", "node2".
Nodes with Config
def node_with_config(state: StateType, config: dict) -> dict:
# Access configurable parameters
user_id = config.get("configurable", {}).get("user_id")
return {"processed": True}
builder.add_node("configurable", node_with_config)Nodes with Extra Keyword Arguments
from langgraph.graph import StateGraph
def node_with_kwargs(state: StateType, **kwargs) -> dict:
# kwargs contains additional runtime parameters
return {"received": kwargs.get("extra_param", "default")}Edges
Edges connect nodes and define the execution path.
Basic Edge
# After node A finishes, run node B
builder.add_edge("A", "B")Entry Point
# Mark the starting node
builder.set_entry_point("A")
# Or using the START constant
from langgraph.graph import START
builder.add_edge(START, "A")Finish Point
# Mark the ending node
builder.set_finish_point("C")
# Or using the END constant
from langgraph.graph import END
builder.add_edge("C", END)Using START and END constants is the modern approach. They are available from langgraph.graph and make the graph definition more readable.
Compilation
Compilation validates the graph structure and produces a runnable object.
# Compile the graph
app = builder.compile()
# The graph is now frozen β no more nodes or edges can be addedWhat Compilation Does
- Validates that all referenced nodes exist
- Checks for unreachable nodes (no incoming edge)
- Verifies the graph is connected (every node reachable from START)
- Freezes the topology so it can be invoked efficiently
- Prepares checkpointers if configured
Invocation
# Invoke with initial state
result = app.invoke({
"messages": [],
"turn_count": 0,
"is_complete": False,
"final_answer": None
})
# Access the final state
print(result["messages"])
print(result["turn_count"])Streaming
# Stream to see intermediate states
for event in app.stream({
"messages": [],
"turn_count": 0,
"is_complete": False,
"final_answer": None
}):
for node_name, state_update in event.items():
if node_name != "__end__":
print(f"[{node_name}]: {state_update}")Complete Minimal Example
from langgraph.graph import StateGraph, START, END
from typing_extensions import TypedDict
class MyState(TypedDict):
value: str
step_count: int
def step_one(state: MyState) -> dict:
print("Step 1")
return {
"value": f"Processed: {state['value']}",
"step_count": state["step_count"] + 1
}
def step_two(state: MyState) -> dict:
print("Step 2")
return {
"value": f"Final: {state['value']}",
"step_count": state["step_count"] + 1
}
# Build
builder = StateGraph(MyState)
builder.add_node("step1", step_one)
builder.add_node("step2", step_two)
builder.add_edge(START, "step1")
builder.add_edge("step1", "step2")
builder.add_edge("step2", END)
# Compile
app = builder.compile()
# Run
result = app.invoke({"value": "hello", "step_count": 0})
print(result["value"]) # Final: Processed: hello
print(result["step_count"]) # 2This pattern β define state, add nodes, add edges, compile, invoke β is the foundation of every LangGraph application. Every agent you build will follow these five steps.
Visualizing the Graph
LangGraph supports generating Mermaid diagrams from compiled graphs:
# Get the Mermaid diagram as a string
mermaid_code = app.get_graph().draw_mermaid()
print(mermaid_code)
# Save to file
with open("graph.md", "w") as f:
f.write(app.get_graph().draw_mermaid())Output Mermaid:
Use app.get_graph().draw_mermaid() during development to verify your graph topology matches your design.
Error Handling at Graph Level
from langgraph.errors import GraphRecursionError
try:
result = app.invoke(initial_state, {"recursion_limit": 10})
except GraphRecursionError:
print("Graph hit recursion limit β possible infinite loop")
except Exception as e:
print(f"Graph execution failed: {e}")Always set a recursion_limit for graphs with loops. The default is usually 25 steps. Without it, a buggy conditional edge can cause an infinite loop.
Practice Questions
What does a node function in LangGraph receive and return?
What does the compile() method do?
Which of the following is a valid way to define state in LangGraph?
What does set_entry_point() do?
What happens if a node function returns None?
What is the purpose of the START constant?
How does LangGraph merge state updates from a node?
Can you add nodes to a compiled graph?
What does recursion_limit control?
What tool does LangGraph provide for visualizing graph structure?
Key Takeaways
- StateGraph is the core class; it manages typed state, nodes, edges, and compilation
- State is a dict defined with TypedDict, dataclass, or BaseModel
- Nodes are functions that receive the full state and return partial updates
- Edges (add_edge, START, END) define the execution topology
- compile() validates and freezes the graph into a runnable object
- invoke() runs the graph; stream() shows intermediate states
- State updates use shallow merge β top-level keys are replaced
- Always set recursion_limit for graphs with loops
- Use get_graph().draw_mermaid() to visualize your graph