LangGraph is the go-to library when standard LangChain agents aren't enough — when you need agents that loop, branch based on conditions, hand off to other agents, pause for human review, or maintain complex shared state across a long-running workflow.
This tutorial builds three progressively complex multi-agent workflows from scratch. By the end, you will understand LangGraph's core abstractions well enough to design and implement production-ready multi-agent systems.
For background on multi-agent system patterns, read Multi-Agent Systems and AI Agent Orchestration first. If you're new to LangChain agents, start with the LangChain tutorial.
What LangGraph Is and Why It Exists#
Standard LangChain agents follow a fixed loop: reason → act → observe → repeat until done. This works well for single-agent tasks with linear execution. It breaks down when you need:
- Multiple specialized agents collaborating on a single task
- Conditional routing (different agents handle different request types)
- Cycles that aren't the standard ReAct loop (e.g., a reviewer that sends work back to a writer)
- Persistent state across multiple turns of a conversation
- Human approval checkpoints mid-workflow
- Long-running workflows with fault-tolerant resume capability
LangGraph solves all of these by modeling agent workflows as directed graphs where:
- Nodes are functions (agents, tools, or any Python callable) that read and write shared state
- Edges define the transitions between nodes (fixed or conditional)
- State is a typed Python object that flows through the entire graph and persists across steps
The "stateful" part is what makes LangGraph different. Every node receives the current state and returns an updated state. LangGraph merges the updates automatically. This eliminates the need for agents to pass context through prompts — they share structured data instead.
Core Concepts Reference#
Before writing code, understand these five abstractions:
StateGraph: The graph object you build your workflow on. You define the state schema, add nodes, add edges, and compile it into a runnable.
TypedDict State: The shared state object, defined as a Python TypedDict. All nodes read from and write to this schema. LangGraph handles merging updates from parallel nodes.
Nodes: Any Python function (sync or async) that takes the current state and returns a dict with state updates. This is where your agent logic lives.
Edges: Two types — normal edges (always go from A to B) and conditional edges (a router function decides which node to go to based on current state).
Checkpointer: A persistence backend (SQLite for dev, PostgreSQL for prod) that saves the graph state after every node execution, enabling pause/resume and human-in-the-loop.
Installation#
pip install langgraph langchain-openai langchain-community python-dotenv
For checkpointing support (needed for Part 3):
pip install langgraph-checkpoint-sqlite
Create a .env file:
OPENAI_API_KEY=your-openai-api-key-here
Part 1: Linear Two-Node Graph (Research → Write)#
The simplest possible LangGraph workflow: two agents in sequence. A research agent gathers information; a writing agent produces a document from the research output.
This demonstrates the fundamental StateGraph pattern before adding complexity.
# part1_linear.py
import os
from typing import TypedDict, Annotated
from dotenv import load_dotenv
from langchain_openai import ChatOpenAI
from langchain_core.messages import HumanMessage, SystemMessage
from langgraph.graph import StateGraph, END
load_dotenv()
# ── 1. Define the shared state schema ──────────────────────────────────────
# All nodes read from and write to this TypedDict.
# Annotated[list, operator.add] means list fields are appended (not overwritten).
class ResearchWriteState(TypedDict):
topic: str # Input: the topic to research and write about
research_notes: str # Set by the research node
draft_article: str # Set by the write node
messages: list # Running log of agent actions
# ── 2. Initialize the LLM ──────────────────────────────────────────────────
llm = ChatOpenAI(model="gpt-4o", temperature=0.3)
# ── 3. Define the Research Node ────────────────────────────────────────────
# A node is any function that takes state and returns a dict of state updates.
def research_node(state: ResearchWriteState) -> dict:
"""Research agent: gathers key facts and talking points on the topic."""
topic = state["topic"]
response = llm.invoke([
SystemMessage(content=(
"You are a research analyst. Given a topic, produce structured "
"research notes: key facts, statistics, and important angles. "
"Be concise and factual. Format as bullet points."
)),
HumanMessage(content=f"Research this topic thoroughly: {topic}")
])
research_output = response.content
print(f"[Research Node] Completed research on: {topic}")
return {
"research_notes": research_output,
"messages": [f"Research completed: {len(research_output)} chars"]
}
# ── 4. Define the Write Node ───────────────────────────────────────────────
def write_node(state: ResearchWriteState) -> dict:
"""Writing agent: produces a well-structured article from research notes."""
topic = state["topic"]
research_notes = state["research_notes"]
response = llm.invoke([
SystemMessage(content=(
"You are a skilled content writer. Given a topic and research notes, "
"write a clear, engaging 400-500 word article. "
"Use proper headings and make it accessible to a general audience."
)),
HumanMessage(content=(
f"Topic: {topic}\n\n"
f"Research Notes:\n{research_notes}\n\n"
"Write the article now."
))
])
draft = response.content
print(f"[Write Node] Article drafted: {len(draft)} chars")
return {
"draft_article": draft,
"messages": [f"Article written: {len(draft)} chars"]
}
# ── 5. Build the Graph ─────────────────────────────────────────────────────
def build_linear_graph() -> StateGraph:
# Initialize graph with our state schema
graph = StateGraph(ResearchWriteState)
# Add nodes
graph.add_node("research", research_node)
graph.add_node("write", write_node)
# Set entry point
graph.set_entry_point("research")
# Add edges: research always goes to write, write always goes to END
graph.add_edge("research", "write")
graph.add_edge("write", END)
return graph.compile()
# ── 6. Run the Graph ───────────────────────────────────────────────────────
if __name__ == "__main__":
app = build_linear_graph()
initial_state = {
"topic": "The business impact of AI agents in 2026",
"research_notes": "",
"draft_article": "",
"messages": []
}
result = app.invoke(initial_state)
print("\n" + "="*60)
print("RESEARCH NOTES:")
print("="*60)
print(result["research_notes"])
print("\n" + "="*60)
print("DRAFT ARTICLE:")
print("="*60)
print(result["draft_article"])
Run it:
python part1_linear.py
You should see the research node complete, then the write node produce a structured article. The state object flowed from one node to the next, carrying the research output as context for the writer.
Part 2: Conditional Routing#
Now add a router that directs incoming requests to different specialized agents based on the request type. This is the pattern used in most production multi-agent systems.
# part2_conditional.py
import os
from typing import TypedDict, Literal
from dotenv import load_dotenv
from langchain_openai import ChatOpenAI
from langchain_core.messages import HumanMessage, SystemMessage
from langgraph.graph import StateGraph, END
load_dotenv()
# ── 1. State Schema ────────────────────────────────────────────────────────
class RouterState(TypedDict):
user_request: str # The incoming user request
request_type: str # Set by router: "technical" | "business" | "creative"
agent_response: str # Set by whichever specialist agent handles it
routing_reason: str # Why the router chose this path
# ── 2. LLM ─────────────────────────────────────────────────────────────────
llm = ChatOpenAI(model="gpt-4o", temperature=0.2)
# ── 3. Router Node ─────────────────────────────────────────────────────────
# This node classifies the request and sets request_type in state.
# The CONDITIONAL EDGE will read request_type to decide the next node.
def router_node(state: RouterState) -> dict:
"""Classifies the user request to determine which specialist to route to."""
request = state["user_request"]
response = llm.invoke([
SystemMessage(content=(
"You are a request classifier. Classify the user's request into exactly "
"one of these three categories:\n"
"- technical: coding, engineering, debugging, architecture questions\n"
"- business: strategy, marketing, sales, operations questions\n"
"- creative: writing, design, brainstorming, creative content\n\n"
"Respond with JSON: {\"type\": \"technical|business|creative\", "
"\"reason\": \"one sentence explanation\"}"
)),
HumanMessage(content=f"Classify this request: {request}")
])
# Parse the classification response
import json
try:
classification = json.loads(response.content)
request_type = classification.get("type", "business")
reason = classification.get("reason", "Default routing")
except (json.JSONDecodeError, AttributeError):
# Fallback if LLM doesn't return clean JSON
content = response.content.lower()
if "technical" in content:
request_type = "technical"
elif "creative" in content:
request_type = "creative"
else:
request_type = "business"
reason = "Fallback classification"
print(f"[Router] Classified as '{request_type}': {reason}")
return {
"request_type": request_type,
"routing_reason": reason
}
# ── 4. Specialist Agent Nodes ──────────────────────────────────────────────
def technical_agent(state: RouterState) -> dict:
"""Handles technical requests with engineering precision."""
response = llm.invoke([
SystemMessage(content=(
"You are a senior software engineer and technical architect. "
"Provide precise, technically accurate answers with code examples where relevant."
)),
HumanMessage(content=state["user_request"])
])
print("[Technical Agent] Response generated")
return {"agent_response": response.content}
def business_agent(state: RouterState) -> dict:
"""Handles business and strategy requests."""
response = llm.invoke([
SystemMessage(content=(
"You are a strategic business consultant. Provide structured, "
"actionable business advice with frameworks and clear recommendations."
)),
HumanMessage(content=state["user_request"])
])
print("[Business Agent] Response generated")
return {"agent_response": response.content}
def creative_agent(state: RouterState) -> dict:
"""Handles creative requests with imaginative flair."""
response = llm.invoke([
SystemMessage(content=(
"You are a creative director and copywriter. Produce original, "
"engaging, and imaginative content tailored to the request."
)),
HumanMessage(content=state["user_request"])
])
print("[Creative Agent] Response generated")
return {"agent_response": response.content}
# ── 5. Routing Function (the key to conditional edges) ────────────────────
# This function receives state and returns the NAME of the next node to go to.
# This is what makes conditional routing work in LangGraph.
def route_to_specialist(state: RouterState) -> Literal["technical", "business", "creative"]:
"""Reads request_type from state and returns the target node name."""
return state["request_type"]
# ── 6. Build the Conditional Graph ─────────────────────────────────────────
def build_router_graph():
graph = StateGraph(RouterState)
# Add all nodes
graph.add_node("router", router_node)
graph.add_node("technical", technical_agent)
graph.add_node("business", business_agent)
graph.add_node("creative", creative_agent)
# Entry point: always start at router
graph.set_entry_point("router")
# Conditional edge: router node → routing function decides which specialist
# The dict maps routing function return values to node names
graph.add_conditional_edges(
"router", # source node
route_to_specialist, # function that returns the next node name
{ # mapping of return values to node names
"technical": "technical",
"business": "business",
"creative": "creative"
}
)
# All specialist nodes go to END
graph.add_edge("technical", END)
graph.add_edge("business", END)
graph.add_edge("creative", END)
return graph.compile()
# ── 7. Test the Router ─────────────────────────────────────────────────────
if __name__ == "__main__":
app = build_router_graph()
test_requests = [
"How do I implement a Redis cache in Python with connection pooling?",
"What's the best go-to-market strategy for a B2B SaaS product targeting SMBs?",
"Write a compelling product launch announcement email for an AI productivity tool."
]
for request in test_requests:
print(f"\n{'='*60}")
print(f"REQUEST: {request[:70]}...")
result = app.invoke({
"user_request": request,
"request_type": "",
"agent_response": "",
"routing_reason": ""
})
print(f"ROUTED TO: {result['request_type']} ({result['routing_reason']})")
print(f"RESPONSE PREVIEW: {result['agent_response'][:200]}...")
The critical piece is add_conditional_edges. The second argument is a Python function that returns a string. LangGraph uses that string to look up the next node in the dict you provide. This pattern is the foundation of every production multi-agent routing system.
Part 3: Full Multi-Agent System with Human-in-the-Loop#
Now build a complete content creation pipeline with three agents and a human approval checkpoint. The graph researches, writes, and then pauses for human review before finalizing.
# part3_hitl.py
import os
from typing import TypedDict, Annotated
import operator
from dotenv import load_dotenv
from langchain_openai import ChatOpenAI
from langchain_core.messages import HumanMessage, SystemMessage
from langgraph.graph import StateGraph, END
from langgraph.checkpoint.sqlite import SqliteSaver
load_dotenv()
# ── 1. State Schema ────────────────────────────────────────────────────────
# Note the Annotated[list, operator.add] pattern for the review_notes list.
# This ensures review notes are appended across multiple human interactions.
class ContentPipelineState(TypedDict):
topic: str
target_audience: str
research_notes: str
draft_content: str
editor_feedback: str
final_content: str
review_notes: Annotated[list, operator.add] # Accumulated across human reviews
status: str # "researching" | "drafting" | "awaiting_review" | "revising" | "complete"
# ── 2. LLM ─────────────────────────────────────────────────────────────────
llm = ChatOpenAI(model="gpt-4o", temperature=0.4)
# ── 3. Agent Nodes ─────────────────────────────────────────────────────────
def research_agent(state: ContentPipelineState) -> dict:
"""Researches the topic and gathers key insights."""
response = llm.invoke([
SystemMessage(content=(
"You are a research specialist. Produce comprehensive, factual research notes "
"on the given topic. Focus on recent developments, statistics, and key insights. "
"Structure as clearly labeled sections."
)),
HumanMessage(content=(
f"Topic: {state['topic']}\n"
f"Target audience: {state['target_audience']}\n\n"
"Produce detailed research notes."
))
])
print("[Research Agent] Research complete")
return {
"research_notes": response.content,
"status": "drafting",
"review_notes": ["Research phase completed"]
}
def writing_agent(state: ContentPipelineState) -> dict:
"""Writes content based on research notes."""
# If editor_feedback exists, this is a revision pass
is_revision = bool(state.get("editor_feedback"))
if is_revision:
prompt = (
f"Topic: {state['topic']}\n"
f"Target audience: {state['target_audience']}\n\n"
f"Research Notes:\n{state['research_notes']}\n\n"
f"Previous Draft:\n{state['draft_content']}\n\n"
f"Editor Feedback to Address:\n{state['editor_feedback']}\n\n"
"Revise the draft to address all editor feedback while maintaining quality."
)
system = "You are a skilled writer revising content based on editor feedback. Address each point specifically."
else:
prompt = (
f"Topic: {state['topic']}\n"
f"Target audience: {state['target_audience']}\n\n"
f"Research Notes:\n{state['research_notes']}\n\n"
"Write a comprehensive, well-structured 600-800 word piece."
)
system = "You are a skilled content writer. Produce high-quality, audience-appropriate content."
response = llm.invoke([
SystemMessage(content=system),
HumanMessage(content=prompt)
])
action = "Revision" if is_revision else "Initial draft"
print(f"[Writing Agent] {action} complete: {len(response.content)} chars")
return {
"draft_content": response.content,
"status": "awaiting_review",
"review_notes": [f"{action} completed"]
}
def finalize_agent(state: ContentPipelineState) -> dict:
"""Final polish pass — runs only after human approval."""
response = llm.invoke([
SystemMessage(content=(
"You are a copy editor doing a final polish. Fix any remaining grammar issues, "
"ensure consistent tone, and improve readability without changing the substance."
)),
HumanMessage(content=(
f"Final polish this approved draft:\n\n{state['draft_content']}"
))
])
print("[Finalize Agent] Final content ready")
return {
"final_content": response.content,
"status": "complete",
"review_notes": ["Final polish completed — content approved for publication"]
}
# ── 4. Human Review Node ───────────────────────────────────────────────────
# This node runs BEFORE the interrupt point.
# It prepares the review package for the human reviewer.
def prepare_review(state: ContentPipelineState) -> dict:
"""Prepares the review summary displayed to the human reviewer."""
review_summary = (
f"CONTENT READY FOR REVIEW\n"
f"Topic: {state['topic']}\n"
f"Audience: {state['target_audience']}\n"
f"Draft length: {len(state['draft_content'])} chars\n"
f"Review history: {len(state['review_notes'])} notes\n\n"
f"DRAFT CONTENT:\n{'-'*40}\n{state['draft_content']}"
)
print("\n" + "="*60)
print(review_summary)
print("="*60)
print("\nGraph paused for human review. Use .update_state() to provide feedback.")
return {"status": "awaiting_review"}
# ── 5. Routing Functions ───────────────────────────────────────────────────
def after_review_route(state: ContentPipelineState) -> str:
"""Routes based on human feedback after the review interrupt."""
# If editor_feedback was set (non-empty) by the human, route to revision
# If editor_feedback is empty/approved, route to finalize
if state.get("editor_feedback", "").strip():
print(f"[Router] Routing to revision. Feedback: {state['editor_feedback'][:80]}...")
return "revise"
else:
print("[Router] Content approved. Routing to finalize.")
return "finalize"
# ── 6. Build Graph with Checkpointing ──────────────────────────────────────
def build_hitl_graph():
graph = StateGraph(ContentPipelineState)
# Add all nodes
graph.add_node("research", research_agent)
graph.add_node("write", writing_agent)
graph.add_node("prepare_review", prepare_review)
graph.add_node("finalize", finalize_agent)
# Set entry point
graph.set_entry_point("research")
# Linear flow: research → write → prepare_review
graph.add_edge("research", "write")
graph.add_edge("write", "prepare_review")
# After prepare_review, conditional routing based on human feedback
graph.add_conditional_edges(
"prepare_review",
after_review_route,
{
"revise": "write", # Revision loops back to write node
"finalize": "finalize"
}
)
graph.add_edge("finalize", END)
# Compile with SQLite checkpointer for state persistence
# interrupt_before="prepare_review" pauses the graph BEFORE that node runs
# The graph will stop, save state, and wait for .invoke() to be called again
memory = SqliteSaver.from_conn_string(":memory:") # Use file path for persistence
return graph.compile(
checkpointer=memory,
interrupt_before=["prepare_review"] # Pause here for human review
)
# ── 7. Run with Human-in-the-Loop ──────────────────────────────────────────
if __name__ == "__main__":
app = build_hitl_graph()
# Each run needs a unique thread_id for state isolation
# In production, this would be a user session ID or task ID
config = {"configurable": {"thread_id": "content-pipeline-001"}}
initial_state = {
"topic": "How AI agents are transforming financial services in 2026",
"target_audience": "Finance executives and CFOs",
"research_notes": "",
"draft_content": "",
"editor_feedback": "",
"final_content": "",
"review_notes": [],
"status": "researching"
}
print("="*60)
print("STARTING CONTENT PIPELINE")
print("="*60)
# First run: executes research → write, then stops before prepare_review
result = app.invoke(initial_state, config=config)
print(f"\nGraph paused at status: {result['status']}")
# ── Simulate Human Review ──────────────────────────────────────────────
# In a real application, this would be a UI interaction or API call.
# Here we simulate the human providing feedback via update_state().
print("\n[HUMAN REVIEWER] Reviewing draft...")
human_feedback = (
"Good structure but needs more specific statistics. "
"Also add a section on implementation challenges. "
"The conclusion is too brief."
)
# Update state with human feedback — this is how humans inject input
app.update_state(
config=config,
values={"editor_feedback": human_feedback}
)
print(f"[HUMAN REVIEWER] Feedback submitted: {human_feedback[:80]}...")
# Resume execution from checkpoint (prepare_review → write revision → prepare_review again)
result = app.invoke(None, config=config)
print(f"\nGraph paused again at status: {result['status']}")
# ── Second Review: Approve ─────────────────────────────────────────────
print("\n[HUMAN REVIEWER] Reviewing revision...")
# Empty editor_feedback signals approval
app.update_state(
config=config,
values={"editor_feedback": ""} # Empty = approved
)
print("[HUMAN REVIEWER] Content approved.")
# Final run: finalize agent polishes and completes
result = app.invoke(None, config=config)
print("\n" + "="*60)
print("FINAL CONTENT:")
print("="*60)
print(result.get("final_content", "Not yet finalized"))
print(f"\nReview history: {result['review_notes']}")
Debugging LangGraph Workflows#
LangGraph provides built-in debugging utilities.
Stream mode for step-by-step visibility:
# Instead of app.invoke(), use app.stream() to see each node's output
for event in app.stream(initial_state, config=config):
for node_name, node_output in event.items():
print(f"Node '{node_name}' returned: {list(node_output.keys())}")
Inspect current graph state:
# Get the current state at any point during or after execution
current_state = app.get_state(config)
print(f"Current values: {current_state.values}")
print(f"Next nodes to execute: {current_state.next}")
print(f"State history length: {len(list(app.get_state_history(config)))}")
Visualize the graph structure:
# Requires graphviz: pip install pygraphviz
from IPython.display import Image
Image(app.get_graph().draw_mermaid_png())
When to Use LangGraph vs. CrewAI vs. AutoGen#
| Scenario | Best Choice | Reason | |----------|-------------|--------| | Precise control over execution flow | LangGraph | Explicit node/edge definition | | Role-based crew with autonomous coordination | CrewAI | Agent role definitions handle coordination | | Conversational multi-agent patterns | AutoGen | Message-passing architecture suits dialogue | | Loops, cycles, complex branching | LangGraph | Native graph support | | Human-in-the-loop workflows | LangGraph | First-class interrupt/checkpoint support | | Rapid prototyping | CrewAI | Less boilerplate for common patterns | | Research and debate patterns | AutoGen | Conversational back-and-forth |
LangGraph has the steepest learning curve of the three but the highest ceiling. For a detailed comparison, see LangChain vs AutoGen and Build Multi-Agent Systems with CrewAI.
What to Build Next#
Now that you have the LangGraph fundamentals, three worthwhile next steps:
Add real tools to your nodes: Integrate web search (Tavily), database queries, or API calls into your agent nodes. Each node can call any Python function, making tool integration straightforward.
Implement persistent checkpointing: Replace SqliteSaver.from_conn_string(":memory:") with a file-path SQLite database or a PostgreSQL backend for production persistence.
Explore parallel node execution: LangGraph supports parallel node execution using Send objects. This enables fan-out patterns where multiple agents work simultaneously and their results are merged back into shared state.
For the RAG pattern that powers the research agent in production deployments, see Introduction to RAG for AI Agents. For complete multi-agent system architecture patterns, see Multi-Agent System Examples and AI Agent Orchestration.