LangGraph State Graphs: Step-by-Step Tutorial (2026)
LangGraph state graphs are the foundation for deterministic, auditable agent workflows. A state graph is a directed acyclic graph where each node is a Python function that transforms a shared state dictionary, and edges are conditional routes that determine the next node. This article walks you through building a working research agent step-by-step: defining state, adding nodes, connecting edges, and handling conditional branching.
Defining Your State
State is the "single source of truth" for your agent. It holds the user's query, tool results, reasoning history, and the final output. Use Python's TypedDict to define the structure—this gives you type hints and makes state explicit.
from typing import TypedDict, Annotated
from typing_extensions import NotRequired
class ResearchAgentState(TypedDict):
"""State for a research agent."""
query: str # User's research question
search_results: NotRequired[str] # Fetched from web search
analysis: NotRequired[str] # LLM's analysis of results
final_answer: NotRequired[str] # Final synthesized response
iteration_count: int # Track loop depth (prevent infinite loops)
messages: list[dict] # Chat history for context
# Initialize
initial_state = {
"query": "What are the latest developments in AI safety?",
"iteration_count": 0,
"messages": []
}
Each node will receive this state dict as input and return a dict with updates. LangGraph merges returned dicts into the state automatically.
Creating Nodes
Nodes are functions that perform work: reasoning, tool calls, data transforms. Each node takes state as input and returns a state update (a dict with only the fields you changed).
from langchain_anthropic import ChatAnthropic
def research_node(state: ResearchAgentState) -> dict:
"""Query the web for information about the topic."""
# In production, use a real search API (e.g., Tavily, SerpAPI)
query = state["query"]
# Simulated search result for demonstration
search_results = f"""
1. "AI Safety Frameworks 2026" - Latest approaches to alignment research
2. "Constitutional AI and Beyond" - Techniques for training safer models
3. "Mechanistic Interpretability Progress Report" - Understanding model internals
"""
return {
"search_results": search_results,
"iteration_count": state["iteration_count"] + 1
}
def analysis_node(state: ResearchAgentState) -> dict:
"""Analyze search results with Claude."""
model = ChatAnthropic(model="claude-3-5-sonnet-20241022")
prompt = f"""
Based on the following search results, provide a concise analysis of {state['query']}:
{state.get('search_results', '')}
Keep the analysis to 250 words and highlight key insights.
"""
response = model.invoke(prompt)
return {
"analysis": response.content,
"iteration_count": state["iteration_count"] + 1
}
def synthesize_node(state: ResearchAgentState) -> dict:
"""Synthesize the analysis into a final answer."""
model = ChatAnthropic(model="claude-3-5-sonnet-20241022")
final_prompt = f"""
Query: {state['query']}
Analysis:
{state.get('analysis', '')}
Now synthesize this into a clear, actionable final answer. Use plain language suitable for a non-technical audience.
"""
response = model.invoke(final_prompt)
return {
"final_answer": response.content,
"iteration_count": state["iteration_count"] + 1
}
Each node is pure Python—no special decorators required. They're testable and debuggable. In production, nodes often call external services (APIs, databases) and handle errors gracefully.
Building the Graph
Now assemble nodes into a graph. Add nodes, define edges, set entry/finish points, and compile.
from langgraph.graph import StateGraph
# Create the graph
graph = StateGraph(ResearchAgentState)
# Add nodes
graph.add_node("research", research_node)
graph.add_node("analysis", analysis_node)
graph.add_node("synthesize", synthesize_node)
# Add edges (defines the flow)
graph.add_edge("research", "analysis")
graph.add_edge("analysis", "synthesize")
# Set entry and finish points
graph.set_entry_point("research")
graph.set_finish_point("synthesize")
# Compile the graph into a runnable
compiled_graph = graph.compile()
This creates a linear workflow: research → analysis → synthesize. Next, we'll add conditional branching.
Conditional Routing
Most real workflows aren't purely linear. A node might decide whether to search again, skip a step, or loop back. Use add_conditional_edges to route based on state.
def should_refine(state: ResearchAgentState) -> str:
"""Decide if we need to refine the analysis."""
# If the analysis is short, refine it with another search
if len(state.get("analysis", "")) < 150:
return "research" # Route back to research
else:
return "synthesize" # Proceed to final synthesis
# Replace the direct edge with conditional routing
graph = StateGraph(ResearchAgentState)
graph.add_node("research", research_node)
graph.add_node("analysis", analysis_node)
graph.add_node("synthesize", synthesize_node)
graph.add_edge("research", "analysis")
# Conditional edge: after analysis, decide what to do next
graph.add_conditional_edges(
"analysis",
should_refine,
{
"research": "research",
"synthesize": "synthesize"
}
)
graph.set_entry_point("research")
graph.set_finish_point("synthesize")
compiled_graph = graph.compile()
The should_refine function inspects state and returns a string that maps to the next node. This gives you full control over branching without complex if-else statements scattered in nodes.
Running and Checkpointing
Execute the graph and optionally persist state to enable resumable runs.
from langgraph.checkpoint.sqlite import SqliteSaver
# Set up checkpointing
saver = SqliteSaver()
compiled_graph = graph.compile(checkpointer=saver)
# Run the agent
config = {"thread_id": "research-001"}
result = compiled_graph.invoke(
initial_state,
config=config
)
print("Final Answer:")
print(result.get("final_answer"))
# If the process crashes, resume from the last checkpoint
# (no need to re-run research and analysis)
result = compiled_graph.invoke(
initial_state,
config=config # Same thread_id
)
Each node execution is checkpointed automatically. If your process crashes mid-run, restarting with the same thread_id resumes from the last completed node.
Error Handling in Nodes
Real agents fail. Model APIs time out, tools return errors, parsing breaks. Handle errors explicitly in nodes.
def research_node_robust(state: ResearchAgentState) -> dict:
"""Query the web, with error handling."""
query = state["query"]
try:
# Real search call would go here
search_results = perform_web_search(query)
except Exception as e:
# Log the error and return a safe fallback
print(f"Search failed: {e}")
search_results = f"Search unavailable for '{query}'. Proceeding with general knowledge."
return {
"search_results": search_results,
"iteration_count": state["iteration_count"] + 1
}
LangGraph doesn't automatically catch node exceptions—you handle them in the node logic. This is intentional: you decide whether an error is fatal or recoverable.
Key Takeaways
- State is the shared immutable data structure; nodes read it and return dicts of updates.
- Nodes are pure Python functions; they're testable, debuggable, and composable.
- Edges define the flow; conditional edges let nodes route based on state introspection.
- Checkpointing is automatic; same thread_id resumes from the last node, avoiding recomputation.
- Error handling is explicit in node code; there's no implicit fallback.
Frequently Asked Questions
How do I debug a LangGraph workflow?
Inspect state at each node by printing or logging it. You can also use LangGraph's built-in visualization (graph.get_graph().draw_mermaid()) to see the structure. For checkpointed runs, use the storage backend to fetch past states.
Can a node call multiple tools?
Yes, a single node can execute multiple API calls, tool invocations, and transformations before returning. Keep nodes focused on one logical task, but don't over-fragment into too many tiny nodes.
What's the max iteration count I should allow?
Set a limit (e.g., 10) to prevent infinite loops. If conditional edges keep routing back to research, the graph will loop 10 times and then stop. This is a safety mechanism.
How do I pass parameters to a node?
Nodes only receive state. To parameterize, store values in state before invoking the graph. For example, set state["temperature"] = 0.7 before running, and nodes read it.
Can I use LangGraph with synchronous-only tools?
Yes, LangGraph handles both sync and async nodes. Mix freely. The framework handles scheduling.