LangGraph state management
Every LangGraph application revolves around state. It is the shared data structure that flows between nodes and gets updated at each step. How state looks determines how execution routes through the graph.
How state works
When you create a StateGraph, you pass it a state schema. This schema defines what data the graph operates on. Every node receives the full state as input and returns a dictionary of partial updates.
By default, updates overwrite existing values. If a node returns {"current_step": "classify"}, the current_step field gets set to "classify". But for list fields like message histories, you usually want to append rather than replace. That is what reducers are for.
Reducers are functions that define how updates to a field are combined. The built-in add_messages reducer appends new messages to the existing list and deduplicates by message ID. You can write custom reducers for any merging logic you need.
Building blocks
The five components that make up every LangGraph workflow.
StateGraph
The main class. Takes a state schema as its argument. You add nodes and edges to it, then call compile() to get a runnable graph.
Nodes
Python functions that do the work. Each node takes the full state as input and returns a dictionary of partial updates. Only the keys you return get updated.
Edges
Connections between nodes. Normal edges always go to the same target. Conditional edges call a routing function to decide where to go based on the current state.
Reducers
Functions that define how state updates are combined. Without a reducer, new values replace old ones. add_messages appends to a list. operator.add sums numbers.
START / END
Special constants. START is the graph entry point. END terminates execution. Connect your first node to START and your final nodes to END.
Defining state
State is defined as a Python TypedDict. Fields without a reducer use last-write-wins semantics. Fields with Annotated use the specified reducer function.
from typing import Annotated
from typing_extensions import TypedDict
from langgraph.graph.message import add_messages
class State(TypedDict):
# With reducer: new messages are appended
messages: Annotated[list, add_messages]
# Without reducer: last write wins
current_step: str
# Custom reducer: values are added together
total_cost: Annotated[float, lambda old, new: old + new]Using MessagesState
For agent-style graphs that primarily work with message lists, LangGraph provides MessagesState as a shortcut. It already includes a messages field with the add_messages reducer. Extend it to add your own fields.
from langgraph.graph import MessagesState
# MessagesState already has messages: Annotated[list, add_messages]
# Just extend it with any extra fields you need
class State(MessagesState):
current_agent: str
iteration_count: intWriting nodes
Nodes are plain Python functions. They receive the full state and return a dictionary with only the keys that changed. LangGraph merges the returned dictionary into the existing state using the configured reducers.
from langgraph.graph import StateGraph, START, END
def classify(state: State):
"""Classify the user's intent."""
# Read from state, return partial updates
last_msg = state["messages"][-1]
intent = model.invoke([
("system", "Classify as 'question' or 'task'."),
last_msg,
])
return {"current_step": intent.content}
def answer(state: State):
"""Answer a question."""
response = model.invoke(state["messages"])
return {"messages": [response]}
def execute(state: State):
"""Execute a task."""
response = model.invoke([
("system", "Execute the requested task."),
*state["messages"],
])
return {"messages": [response]}Connecting nodes with edges
Edges define how execution flows through the graph. Use add_edge for fixed transitions and add_conditional_edges for dynamic routing. The routing function receives the current state and returns the name of the next node.
graph = StateGraph(State)
# Add nodes
graph.add_node("classify", classify)
graph.add_node("answer", answer)
graph.add_node("execute", execute)
# Entry point
graph.add_edge(START, "classify")
# Conditional edge: route based on classification
def route_by_intent(state: State):
if state["current_step"] == "question":
return "answer"
return "execute"
graph.add_conditional_edges("classify", route_by_intent)
# Both endpoints go to END
graph.add_edge("answer", END)
graph.add_edge("execute", END)
app = graph.compile()Parallel execution
LangGraph supports fan-out/fan-in patterns using the Send API. A conditional edge function can return a list of Send objects, each targeting the same node with different inputs. The node executes in parallel for each input.
from langgraph.types import Send
def fan_out(state: State):
"""Send each item to a worker node in parallel."""
return [
Send("process_item", {"item": item})
for item in state["items"]
]
graph = StateGraph(State)
graph.add_node("process_item", process_item)
graph.add_conditional_edges(START, fan_out)Best practices
Keep state flat
Avoid deeply nested state objects. Flat state schemas are easier to update, debug, and serialize. If you need complex data, store it as a JSON string or use a separate store.
Use reducers for lists
If multiple nodes write to the same list field, use a reducer like add_messages to append values. Without a reducer, each write overwrites the previous value.
Return only what changed
Nodes should return a dictionary with only the keys that changed. Returning the full state works but is wasteful and can cause issues with reducers.
Name your routing functions clearly
Conditional edge routing functions should have descriptive names like route_by_intent or should_continue. This makes the graph structure easier to understand when debugging.
Use input and output schemas
StateGraph accepts optional input_schema and output_schema parameters. Use them to restrict which state keys are exposed as inputs and outputs, keeping your API clean.
FAQ
State management questions
Deploy your graph to production
Built your graph? Deploy it with a single command.