The Big Picture: Why LangGraph?
The journey from simple LLM calls to production agents follows a clear pattern. You start with linear chains, then realize you need loops, branching, and persistent memory. LangGraph is the framework that makes this possible.
The Evolution of LLM Systems
Simple Chains: LLM chains execute once: input → LLM → output. No loops, no conditionals, no state. Perfect for summarization or classification, but hopeless for reasoning.
Agents (ReAct): An agent loops: think → act → observe → repeat. It needs a way to call tools, receive results, and adjust its reasoning. But managing state manually is chaotic.
Stateful Graphs: LangGraph introduces state—a shared memory object—and nodes—computation units—connected by edges. Now loops are natural, state flows through the graph, and you can checkpoint at any step.
Problems with Simple Chains
LangChain LCEL (Expression Language) chains are powerful for linear workflows, but they break down when you need:
No Loops
You can't retry failed steps or loop until a goal is reached without hacks.
No Branching
Conditional routing (if this, do A; else do B) requires custom code.
No State Persistence
Each call is stateless. Multi-turn conversations lose context.
No Checkpoints
Can't pause, inspect, or resume workflows mid-execution.
How LangGraph Solves It
Cycles: Graph edges can point backward, enabling retry loops and iterative reasoning.
Branching: Conditional edges route to different nodes based on state. A router node can split traffic to specialized agents.
Persistent State: A shared state dict flows through the graph. Each node reads, modifies, and passes it forward. The entire conversation history, intermediate results, and decision trails are preserved.
Checkpointing: Save state at any step. Resume from any checkpoint. Debug by replaying. Pause for human approval. This is mission-critical for production systems.
Core Building Blocks
StateGraph
Container for the entire workflow. Defines the state schema (TypedDict) and holds all nodes and edges.
Nodes
Functions that process state. Each node reads state, does work, and returns a partial state update.
Edges
Connections between nodes. Normal edges always go from A to B. Conditional edges choose dynamically.
Checkpointing
Save state snapshots. Resume interrupted workflows. Enables time-travel debugging and human approval gates.
Relationship to LangChain
LangGraph extends LangChain. LangChain provides the primitives (LLMs, tools, memory), while LangGraph adds the orchestration layer (graphs, state, cycles). You still use LangChain components (ChatOpenAI, WikipediaRetriever) inside LangGraph nodes.
Architecture Comparison: Linear Chain vs LangGraph
Core Concepts: State
State is the memory of the graph. It's a shared dictionary that flows through every node, accumulating information, decisions, and results. Every node reads it, modifies it, and passes it forward.
State Schema: Define Your Memory
State is a TypedDict—a Python dict with typed keys. It defines what data flows through your graph:
from typing import TypedDict, Annotated
from typing_extensions import NotRequired
from operator import add
class AgentState(TypedDict):
messages: Annotated[list, add] # Chat history (appended)
step_count: int # Steps taken
tool_results: list # Tool outputsEach key has a type. Nodes return partial dicts with only changed keys. LangGraph merges updates automatically using reducers.
State Flow Through the Graph
State flows like a relay race: each node receives it, modifies it, and passes it forward. By execution's end, state holds all accumulated work.
messages: Annotated[list, add] # add reducer: append step_count: int # default: replace (last write wins)
Reducers: How to Merge Updates
add reducer: Concatenate lists. Perfect for message histories that grow.
Default (replace): Last write wins. Overwrites previous value.
Custom: Define your own merge logic.
Core Concepts: Nodes
Nodes are the computation units of a graph. Each node is a function that receives state, does work, and returns a partial state update. Nodes are where the actual logic happens.
Node Function Signature
Every node follows this pattern:
def my_node(state: AgentState) -> dict:
"""Process state and return partial update."""
# Read from state
messages = state.get("messages", [])
step_count = state["step_count"]
# Do work
result = do_something(messages, step_count)
# Return partial update
return {
"messages": [...],
"step_count": step_count + 1
}Input: The full state dict. Read whatever you need.
Output: A partial dict. Only include keys you're updating. LangGraph merges this into shared state.
The LLM Node Pattern
def agent_node(state: AgentState) -> dict:
"""Call LLM and add response to messages."""
llm = ChatOpenAI(model="gpt-4", temperature=0)
messages = state["messages"]
response = llm.invoke(messages)
return {
"messages": [response],
"step_count": state["step_count"] + 1
}The Tool Node Pattern
def tool_node(state: AgentState) -> dict:
"""Execute tools called by LLM."""
last_msg = state["messages"][-1]
results = []
for tool_call in last_msg.tool_calls:
result = execute_tool(tool_call["name"], tool_call["args"])
results.append({"type": "tool_result", "result": result})
return {"messages": results, "tool_results": results}Core Concepts: Edges
Edges connect nodes. Normal edges always route from source to destination. Conditional edges choose based on state. Together, they define the flow through your graph.
Normal Edges: Static Routing
A normal edge always goes from source to destination:
graph.add_edge("agent", "tools") # Always: agent → toolsConditional Edges: Dynamic Routing
Choose the next node based on state:
def should_continue(state: AgentState) -> str:
last_msg = state["messages"][-1]
if last_msg.tool_calls:
return "tools"
return "end"
graph.add_conditional_edges(
"agent",
should_continue,
{"tools": "tools", "end": END}
)Compiling the Graph
After defining nodes and edges, compile into an executable:
app = graph.compile()
result = app.invoke({"messages": [HumanMessage(content="...")]})Building a Simple ReAct Agent
ReAct = Reasoning + Acting + Observing. The agent reasons about what to do, acts by calling a tool, and observes the result. This loop repeats until done. LangGraph makes ReAct agents natural.
The ReAct Loop
1. Reason: LLM receives messages. It thinks and decides to call a tool or finish.
2. Act: If tool decided, execute it and capture the result.
3. Observe: Add result back to messages. Loop back to step 1.
4. End: When LLM doesn't call tools, return the final response.
ReAct as a LangGraph
Checkpointing & State Persistence
Checkpointing saves the graph's state at each step. This enables resuming interrupted workflows, human-in-the-loop pauses, time-travel debugging, and conversation persistence. It's essential for production systems.
Why Checkpointing Matters
Resume Interrupted Workflows: If execution crashes, resume from the last checkpoint instead of restarting from scratch.
Human-in-the-Loop: Pause before a sensitive action, let a human approve, then resume.
Debugging: Replay from any checkpoint to inspect intermediate state.
Multi-Turn Conversations: Each user message starts a new execution. Checkpoints maintain conversation history.
Checkpoint Savers
MemorySaver
In-memory checkpointing. Good for development. Lost on restart.
SQLiteSaver
Persistent file-based storage. Perfect for local/small apps.
RedisSaver
Fast, distributed checkpointing. For production systems.
PostgresSaver
Enterprise-grade SQL backend. Full query capabilities.
Using MemorySaver (Development)
from langgraph.checkpoint.memory import MemorySaver
checkpointer = MemorySaver()
app = graph.compile(checkpointer=checkpointer)
# Invoke with thread_id for persistence
config = {"configurable": {"thread_id": "user_123"}}
result = app.invoke({"messages": [HumanMessage(content="Hi")]}, config)
# Next call with same thread_id resumes state
result = app.invoke({"messages": [HumanMessage(content="Tell me more")]}, config)Using SQLiteSaver (Persistent)
from langgraph.checkpoint.sqlite import SqliteSaver
import sqlite3
checkpointer = SqliteSaver(conn=sqlite3.connect("checkpoints.db"))
app = graph.compile(checkpointer=checkpointer)
config = {"configurable": {"thread_id": "conversation_456"}}
# First turn
response_1 = app.invoke({"messages": [HumanMessage(content="...")]}, config)
# Second turn (state is resumed from last checkpoint)
response_2 = app.invoke({"messages": [HumanMessage(content="...")]}, config)Thread IDs: Isolating Conversations
Each conversation has a unique thread_id. State is isolated per thread. Multiple conversations don't interfere:
app = graph.compile(checkpointer=checkpointer)
# User A's conversation
config_a = {"configurable": {"thread_id": "user_alice_123"}}
app.invoke({"messages": [...]}, config_a)
# User B's conversation (separate state)
config_b = {"configurable": {"thread_id": "user_bob_456"}}
app.invoke({"messages": [...]}, config_b)
# Both threads are independentInterrupt Before/After Nodes
Pause execution before or after specific nodes for human approval:
app = graph.compile(
checkpointer=checkpointer,
interrupt_before=["approve_node"],
interrupt_after=["execute_action"]
)
config = {"configurable": {"thread_id": "user_123"}}
result = app.invoke({"messages": [...]}, config)
# Execution pauses at "approve_node"
# Inspect state
state = app.get_state(config)
print(f"Paused at: {state.next}")
# Resume execution
app.invoke(None, config)Conditional Workflows & Branching
Real-world workflows rarely run linearly. Queries branch to specialized handlers. Chains fork and merge. Workflows loop conditionally. LangGraph makes this natural.
Router Node Pattern
A router classifies the input and dispatches to specialized agents:
def router_node(state: AgentState) -> dict:
"""Classify query intent and route accordingly."""
query = state["messages"][-1].content
if "refund" in query.lower():
intent = "billing"
elif "error" in query.lower():
intent = "technical"
else:
intent = "general"
return {"intent": intent}
def route_by_intent(state: AgentState) -> str:
return state["intent"]
graph.add_conditional_edges(
"router",
route_by_intent,
{
"billing": "billing_agent",
"technical": "technical_agent",
"general": "general_agent"
}
)Multi-Agent Branching
Run specialized agents and combine results:
def fork_agents(state: AgentState) -> dict:
"""Run all specialized agents."""
query = state["messages"][-1].content
technical = technical_agent({"query": query})
business = business_agent({"query": query})
sentiment = sentiment_agent({"query": query})
return {
"analyses": [technical, business, sentiment],
"messages": state["messages"]
}
def merge_results(state: AgentState) -> dict:
"""Combine results from parallel branches."""
combined = combine_analyses(state["analyses"])
return {"final_response": combined}Sub-Graphs: Nesting Workflows
For modular, reusable workflows, nest graphs within graphs:
def create_billing_subgraph() -> StateGraph:
"""Sub-graph for billing queries."""
subgraph = StateGraph(AgentState)
subgraph.add_node("classify", classify_billing_issue)
subgraph.add_node("retrieve", retrieve_billing_info)
subgraph.add_node("generate", generate_response)
# ... edges ...
return subgraph.compile()
# Main graph uses the sub-graph
main_graph = StateGraph(AgentState)
main_graph.add_node("router", router_node)
billing_subgraph = create_billing_subgraph()
main_graph.add_node("billing", billing_subgraph)Cross-Reference: For multi-agent coordination patterns, see multi_agent_systems_guide.html.
Human-in-the-Loop Workflows
Some decisions are too important to automate. Financial transactions, medical recommendations, sensitive business logic—these need human judgment. LangGraph's interrupt mechanism enables seamless approval gates.
When to Pause for Humans
High-stakes actions: Large transfers, policy changes, sensitive data deletion.
Uncertainty: Agent is unsure (low confidence). Ask human for clarification.
Policy enforcement: Action violates policy. Human escalation required.
Correction: Agent made an error. Human corrects and resumes.
Using interrupt_before
Pause right before a node executes:
checkpointer = MemorySaver()
app = graph.compile(
checkpointer=checkpointer,
interrupt_before=["approve_transfer", "delete_records"]
)
config = {"configurable": {"thread_id": "transaction_123"}}
# Run until it reaches "approve_transfer", then pauses
result = app.invoke({
"messages": [HumanMessage(content="Transfer $10,000")],
"action": "transfer",
"amount": 10000
}, config)
# Check what's about to happen
state = app.get_state(config)
print(f"Paused before: {state.next}")
print(f"Current state: {state.values}")
# Human approves or rejects
if user_approved:
app.invoke(None, config) # Resume executionUsing interrupt_after
Pause right after a node completes. Useful for final approval of results:
app = graph.compile(
checkpointer=checkpointer,
interrupt_after=["generate_report", "format_response"]
)
config = {"configurable": {"thread_id": "report_456"}}
result = app.invoke({"query": "Create Q3 report"}, config)
# Human reviews generated report
state = app.get_state(config)
print(f"Generated report: {state.values['report']}")
# If satisfied, resume
app.invoke(None, config)Approval Workflow Pattern
Streaming & Observability
Real-time visibility is critical for production systems. Streaming events let you see execution as it happens. LangSmith integration provides detailed traces. Together, they enable debugging and optimization.
Streaming Events
Stream events instead of waiting for the full result:
app = graph.compile(checkpointer=checkpointer)
config = {"configurable": {"thread_id": "stream_123"}}
for event in app.stream({"messages": [HumanMessage(content="Hello")]}, config):
print(f"Event: {event}")
# Output events:
# on_node_start, on_node_end, on_tool_start, on_tool_end, etc.Event Types
on_node_start: A node begins. Includes input state.
on_node_end: A node completes. Includes output/updates.
on_tool_start: A tool executes. Shows tool name and input.
on_tool_end: Tool completes. Shows output.
LangSmith Integration
Automatic tracing of all LLM calls and graph execution:
import os os.environ["LANGCHAIN_API_KEY"] = "your_key" os.environ["LANGCHAIN_TRACING_V2"] = "true" os.environ["LANGCHAIN_PROJECT"] = "my_agent_project" # All LLM calls and graph steps are automatically traced result = app.invoke(input, config) # View traces at https://smith.langchain.com
Debugging with Print Statements
def agent_node(state: AgentState) -> dict:
print(f"=== Agent Node ===")
print(f"Input messages: {len(state['messages'])}")
print(f"Last message: {state['messages'][-1].content[:100]}...")
response = llm.invoke(state["messages"])
print(f"Response: {response.content[:100]}...")
print(f"Tool calls: {response.tool_calls}")
return {"messages": [response]}Cross-Reference: For detailed tracing patterns, see production_agent_workflows_guide.html.
LangGraph vs Alternatives
Many frameworks exist for building agent systems. How does LangGraph compare? When should you use it?
Feature Comparison Table
| Feature | LangGraph | Plain Python | AutoGen | CrewAI | Dify |
|---|---|---|---|---|---|
| State Persistence | Built-in | Manual | Limited | Limited | Built-in |
| Cycles & Loops | Native | Easy | Native | Native | Native |
| Branching & Routing | Conditional edges | Easy | Manual | Manual | Visual |
| Human-in-Loop | Interrupts | Manual | Limited | Limited | Built-in |
| Multi-Agent | Flexible | Manual | Designed for | Designed for | Limited |
| Observability | LangSmith | DIY | Limited | Limited | Built-in |
| No-Code UI | Code-first | Code | Code | Code | Visual editor |
| Production-Ready | Yes | Depends | Emerging | Emerging | Yes |
When to Choose LangGraph
Complex workflows with many branches – Conditional edges handle routing naturally.
Custom agent patterns – Full control over nodes, edges, state, and logic.
Production systems – Checkpointing, interrupts, and observability are built-in.
Multi-turn conversations – State persistence across turns is seamless.
LangChain integration – Uses LangChain tools, retrievers, and chains directly.
When to Choose Alternatives
Plain Python: Simple scripts, proof-of-concept, learning. No dependencies, maximum control.
AutoGen: Multi-agent research, conversation-based coordination. Excellent for complex multi-agent scenarios.
CrewAI: Team-based agents with roles and responsibilities. Great for role-playing scenarios.
Dify: No-code workflows, rapid prototyping, business user workflows. Visual editor, no coding required.
Cross-References:
- multi_agent_systems_guide.html – Deep dive into multi-agent coordination.
- rag_patterns_guide.html – Retrieval-augmented generation patterns with LangGraph.
- agent_tools_mcp_guide.html – Tool integration and MCP servers.
- production_agent_workflows_guide.html – Production deployment and scaling.
Python Implementation: Complete Working Example
Theory is great, but let's build something that works. This section contains a complete, runnable Python implementation of a ReAct agent—no external LLM API calls, no langgraph imports needed. We use mock LLM responses and implement the graph execution manually.
Core Components
1. GraphState (TypedDict): Defines the state schema.
2. MockLLM: Simulates an LLM with tool calls and text responses.
3. SimpleGraph: A manual graph implementation with nodes, edges, and execution.
4. ReAct Agent: Reason-Act-Observe loop using our graph.
Full Implementation
from typing import TypedDict, Annotated, Any
from operator import add
import json
# STATE SCHEMA
class GraphState(TypedDict):
"""Shared state flowing through the graph."""
messages: Annotated[list, add]
step_count: int
tool_results: list
is_done: bool
# MOCK LLM
class MockLLM:
"""Simulates an LLM with tool calls and text responses."""
def __init__(self):
self.step = 0
def invoke(self, messages):
"""Given messages, return response with optional tool calls."""
self.step += 1
last_msg = messages[-1]["content"] if messages else ""
if self.step == 1 and "5 + 3" in last_msg:
return {
"type": "assistant",
"content": "I'll calculate 5 + 3 for you.",
"tool_calls": [
{"name": "add_numbers", "args": {"a": 5, "b": 3}, "id": "call_1"}
]
}
if self.step == 2:
return {
"type": "assistant",
"content": "The result of 5 + 3 is 8. Done!",
"tool_calls": []
}
return {"type": "assistant", "content": "I'm done.", "tool_calls": []}
# TOOLS
TOOLS = {
"add_numbers": lambda a, b: {"result": a + b},
"multiply": lambda a, b: {"result": a * b},
}
# GRAPH NODES
def agent_node(state: GraphState) -> dict:
"""LLM node: think and decide action."""
print(f"\n[AGENT] Step {state['step_count'] + 1}")
llm = MockLLM()
messages = state["messages"]
response = llm.invoke(messages)
print(f" Response: {response['content']}")
return {
"messages": [response],
"step_count": 1,
}
def tool_node(state: GraphState) -> dict:
"""Execute tools called by LLM."""
print(f"\n[TOOLS]")
last_msg = state["messages"][-1]
tool_calls = last_msg.get("tool_calls", [])
results = []
for call in tool_calls:
tool_name = call["name"]
args = call["args"]
print(f" Executing {tool_name}({args})")
fn = TOOLS.get(tool_name)
if fn:
result = fn(**args)
print(f" Result: {result}")
results.append({
"type": "tool_result",
"tool_name": tool_name,
"result": result,
"call_id": call["id"]
})
return {
"messages": results,
"tool_results": results
}
def should_continue(state: GraphState) -> str:
"""Decide: continue loop or end?"""
last_msg = state["messages"][-1]
if last_msg.get("tool_calls"):
print(f" → Continue to tools")
return "tools"
print(f" → Done")
return "end"
# GRAPH IMPLEMENTATION
class SimpleGraph:
"""Manual graph (mirrors LangGraph logic)."""
def __init__(self, state_schema):
self.nodes = {}
self.edges = []
self.conditional_edges = []
self.entry_point = None
self.state_schema = state_schema
def add_node(self, name, fn):
self.nodes[name] = fn
def set_entry_point(self, name):
self.entry_point = name
def add_edge(self, source, dest):
self.edges.append((source, dest))
def add_conditional_edges(self, source, condition_fn, mapping):
self.conditional_edges.append((source, condition_fn, mapping))
def compile(self):
return CompiledGraph(self)
class CompiledGraph:
"""Executable graph."""
def __init__(self, graph):
self.graph = graph
def invoke(self, initial_input):
"""Execute the graph."""
state = {
"messages": initial_input.get("messages", []),
"step_count": 0,
"tool_results": [],
"is_done": False
}
current_node = self.graph.entry_point
print(f"Starting at node: {current_node}")
iteration = 0
while current_node and iteration < 10:
iteration += 1
print(f"\n=== Iteration {iteration} ===")
node_fn = self.graph.nodes[current_node]
update = node_fn(state)
state = self._merge_state(state, update)
next_node = None
for source, condition_fn, mapping in self.graph.conditional_edges:
if source == current_node:
key = condition_fn(state)
next_node = mapping.get(key, "end")
break
if next_node == "end":
break
current_node = next_node
print(f"\n=== Final State ===")
print(f"Messages: {len(state['messages'])} total")
print(f"Steps: {state['step_count']}")
print(f"Final: {state['messages'][-1].get('content', 'N/A')}")
return state
@staticmethod
def _merge_state(old_state, update):
"""Merge partial update into state."""
new_state = old_state.copy()
for key, value in update.items():
if key == "messages":
new_state["messages"] = old_state.get("messages", []) + value
else:
new_state[key] = value
return new_state
# BUILD AND RUN
def main():
print("=" * 60)
print("LangGraph ReAct Agent (Mock Implementation)")
print("=" * 60)
graph = SimpleGraph(GraphState)
graph.add_node("agent", agent_node)
graph.add_node("tools", tool_node)
graph.set_entry_point("agent")
graph.add_conditional_edges(
"agent",
should_continue,
{"tools": "tools", "end": "end"}
)
graph.edges.append(("tools", "agent"))
app = graph.compile()
input_data = {
"messages": [{"type": "user", "content": "What is 5 + 3?"}]
}
print(f"\nUser Query: {input_data['messages'][0]['content']}")
result = app.invoke(input_data)
print("\n" + "=" * 60)
print("EXECUTION COMPLETE")
print("=" * 60)
print(f"Total messages: {len(result['messages'])}")
print(f"Total steps: {result['step_count']}")
if __name__ == "__main__":
main()How It Works
1. GraphState: TypedDict with messages (append reducer), step_count, tool_results.
2. Nodes: agent_node() calls MockLLM, tool_node() executes tools. Each returns a partial state update.
3. Edges: conditional_edges determine routing. should_continue() checks if LLM called tools (yes → tools, no → end).
4. Graph Execution: Start at entry_point. Execute node, merge update into state. Check conditional edges for next node. Repeat until END.
Key Insights
State Merging
Partial updates merge into shared state. The "add" reducer concatenates messages.
Node Execution
Nodes are pure functions: state in, partial update out. No side effects.
Routing Logic
Conditional edges use a decision function that returns a destination.
Loop Control
Loops are explicit via edges. Current node checks conditional edges for next node.
Running the Example
Copy the code into a Python file and run:
python agent.py
Output:
============================================================
LangGraph ReAct Agent (Mock Implementation)
============================================================
User Query: What is 5 + 3?
Starting at node: agent
=== Iteration 1 ===
[AGENT] Step 1
Response: I'll calculate 5 + 3 for you.
→ Continue to tools
[TOOLS]
Executing add_numbers({'a': 5, 'b': 3})
Result: {'result': 8}
=== Iteration 2 ===
[AGENT] Step 2
Response: The result of 5 + 3 is 8. Done!
→ Done
============================================================
EXECUTION COMPLETE
============================================================
Total messages: 3
Total steps: 2Extensions & Next Steps
Add persistence: Save state to a file or database between executions.
Add streaming: Implement the stream() method to yield events in real-time.
Add interrupts: Pause before sensitive nodes for human approval.
Add multi-agent: Multiple agent nodes with different specializations.
Use real LangGraph: Replace SimpleGraph with actual LangGraph and the real ChatOpenAI LLM.