LangGraph: Beginner to Advanced
LangGraph is used to build advanced AI workflows. It is especially useful when your AI system needs multiple steps, branching, retries, or state-based decision making.
What Is LangGraph?
LangGraph helps build workflows where each step is connected like a graph.
A node is one step. An edge is the connection from one step to another.
It is useful when one simple chain is not enough.
Start
↓
Understand Question
↓
┌─────┴─────┐
↓ ↓
Search Tool Database
↓ ↓
└─────┬─────┘
↓
Reasoning
↓
Final AnswerWhy LangGraph Is Important
Some AI tasks need retries, loops, branching, or multi-step logic.
LangGraph makes these flows easier to manage than a simple chain.
Simple LangGraph Example
This example shows a very small workflow with two connected steps.
from langgraph.graph import StateGraph
graph = StateGraph()
def step1(state):
return "Step 1 processed"
def step2(state):
return "Step 2 processed"
graph.add_node("step1", step1)
graph.add_node("step2", step2)
graph.set_entry_point("step1")
graph.add_edge("step1", "step2")
workflow = graph.compile()Intermediate: State Management in LangGraph
Every node in a LangGraph workflow reads from and writes to a shared state object. The state is the single source of truth for the entire workflow.
You define state using TypedDict. Each node receives the current state and returns an updated version of it.
Good state design is the foundation of any reliable LangGraph workflow.
from typing import TypedDict, Optional
from langgraph.graph import StateGraph, END
# Define the shared state structure
class ResearchState(TypedDict):
question: str
research_result: Optional[str]
final_answer: Optional[str]
error: Optional[str]
# Node 1: Simulates research
def do_research(state: ResearchState) -> ResearchState:
question = state["question"]
# Simulate research (in real app, call a search API here)
result = f"Research findings on '{question}': This topic has significant recent developments."
return {**state, "research_result": result}
# Node 2: Write a final answer from research
def write_answer(state: ResearchState) -> ResearchState:
research = state["research_result"]
answer = f"Based on research: {research}. Therefore, this is a well-documented area."
return {**state, "final_answer": answer}
# Build the graph
graph = StateGraph(ResearchState)
graph.add_node("research", do_research)
graph.add_node("answer", write_answer)
graph.set_entry_point("research")
graph.add_edge("research", "answer")
graph.add_edge("answer", END)
workflow = graph.compile()
# Run the workflow
output = workflow.invoke({"question": "What is prompt engineering?"})
print("Final Answer:", output["final_answer"])Intermediate: Conditional Branching in Workflows
Conditional edges let your workflow make decisions. Based on the current state, the graph can take different paths.
For example: if research found results, go to the writer; if it failed, go to a fallback node.
This is how real AI workflows handle different outcomes without crashing.
from typing import TypedDict, Optional, Literal
from langgraph.graph import StateGraph, END
class TaskState(TypedDict):
query: str
search_result: Optional[str]
answer: Optional[str]
status: str # "success" or "failed"
# Simulate a search that sometimes fails
def search_node(state: TaskState) -> TaskState:
import random
if random.random() > 0.3: # 70% success rate
return {**state, "search_result": f"Found data for: {state['query']}", "status": "success"}
else:
return {**state, "search_result": None, "status": "failed"}
# Write answer when search succeeds
def answer_node(state: TaskState) -> TaskState:
answer = f"Answer based on search: {state['search_result']}"
return {**state, "answer": answer}
# Fallback when search fails
def fallback_node(state: TaskState) -> TaskState:
answer = "I could not find specific data. Here is a general answer based on my knowledge."
return {**state, "answer": answer}
# Routing function — decides next node based on state
def route_after_search(state: TaskState) -> Literal["answer", "fallback"]:
if state["status"] == "success":
return "answer"
return "fallback"
# Build the graph with conditional routing
graph = StateGraph(TaskState)
graph.add_node("search", search_node)
graph.add_node("answer", answer_node)
graph.add_node("fallback", fallback_node)
graph.set_entry_point("search")
graph.add_conditional_edges("search", route_after_search, {
"answer": "answer",
"fallback": "fallback"
})
graph.add_edge("answer", END)
graph.add_edge("fallback", END)
workflow = graph.compile()
result = workflow.invoke({"query": "Latest AI research", "status": "pending"})
print("Answer:", result["answer"])Intermediate: Adding Retry Logic to Nodes
When a node calls an external API or tool, it can fail. Retry logic wraps the node function to try again if it gets a transient error.
A good retry pattern: try up to 3 times, wait briefly between tries, and after max retries set an error in state for the routing logic to handle.
This prevents flaky external services from breaking your entire workflow.
import time
from typing import TypedDict, Optional
from langgraph.graph import StateGraph, END
class ApiState(TypedDict):
query: str
api_result: Optional[str]
retries: int
error: Optional[str]
def call_external_api(query: str) -> str:
"""Simulates an external API call that might fail."""
import random
if random.random() < 0.5:
raise ConnectionError("API temporarily unavailable")
return f"API response for: {query}"
def api_node_with_retry(state: ApiState) -> ApiState:
max_retries = 3
retries = state.get("retries", 0)
for attempt in range(max_retries):
try:
result = call_external_api(state["query"])
return {**state, "api_result": result, "error": None, "retries": attempt + 1}
except ConnectionError as e:
if attempt < max_retries - 1:
time.sleep(0.5) # brief wait before retry
continue
# All retries exhausted
return {**state, "api_result": None, "error": str(e), "retries": max_retries}
def success_node(state: ApiState) -> ApiState:
print(f"Success after {state['retries']} attempt(s):", state["api_result"])
return state
def error_node(state: ApiState) -> ApiState:
print(f"Failed after {state['retries']} retries. Error:", state["error"])
return state
def route(state: ApiState):
return "success" if state["api_result"] else "error"
graph = StateGraph(ApiState)
graph.add_node("api_call", api_node_with_retry)
graph.add_node("success", success_node)
graph.add_node("error", error_node)
graph.set_entry_point("api_call")
graph.add_conditional_edges("api_call", route, {"success": "success", "error": "error"})
graph.add_edge("success", END)
graph.add_edge("error", END)
workflow = graph.compile()
workflow.invoke({"query": "student grades", "retries": 0})Advanced: Multi-Agent Graph Orchestration
In advanced LangGraph applications, different agents handle specialized tasks. The graph connects them so output from one agent flows as input to the next.
You can define subgraphs — a complete graph that itself acts as one node in a larger graph. This allows nested, modular architectures.
Multi-agent graphs are used in production for complex tasks like automated research → writing → review pipelines.
from typing import TypedDict, Optional
from langgraph.graph import StateGraph, END
from langchain.chat_models import ChatOpenAI
from langchain.schema import HumanMessage, SystemMessage
llm = ChatOpenAI(model="gpt-3.5-turbo", temperature=0.3)
class PipelineState(TypedDict):
topic: str
research: Optional[str]
draft: Optional[str]
reviewed: Optional[str]
# Agent 1: Research Agent
def research_agent(state: PipelineState) -> PipelineState:
messages = [
SystemMessage(content="You are a research assistant. Find 3 key facts."),
HumanMessage(content=f"Research this topic: {state['topic']}")
]
result = llm(messages).content
return {**state, "research": result}
# Agent 2: Writer Agent
def writer_agent(state: PipelineState) -> PipelineState:
messages = [
SystemMessage(content="You are a writer. Write a clear 2-paragraph summary."),
HumanMessage(content=f"Using these facts, write a summary:\n{state['research']}")
]
result = llm(messages).content
return {**state, "draft": result}
# Agent 3: Reviewer Agent
def reviewer_agent(state: PipelineState) -> PipelineState:
messages = [
SystemMessage(content="You are an editor. Improve clarity and fix any issues. Return the final version."),
HumanMessage(content=f"Review this draft:\n{state['draft']}")
]
result = llm(messages).content
return {**state, "reviewed": result}
# Build the multi-agent graph
graph = StateGraph(PipelineState)
graph.add_node("research", research_agent)
graph.add_node("write", writer_agent)
graph.add_node("review", reviewer_agent)
graph.set_entry_point("research")
graph.add_edge("research", "write")
graph.add_edge("write", "review")
graph.add_edge("review", END)
pipeline = graph.compile()
output = pipeline.invoke({"topic": "Benefits of AI in education"})
print("Final Reviewed Output:\n", output["reviewed"])Advanced: Human-in-the-Loop Workflows
Some workflows need a human to approve or modify a result before the process continues. LangGraph supports interrupt points for this pattern.
You set a breakpoint at a specific node. The graph pauses, sends its current state to the user, waits for a human response, then continues.
This is critical for applications where AI-generated content must be reviewed before being published or acted upon.
from typing import TypedDict, Optional
from langgraph.graph import StateGraph, END
class ReviewState(TypedDict):
draft: str
human_feedback: Optional[str]
final_output: Optional[str]
# Node 1: Generate a draft
def generate_draft(state: ReviewState) -> ReviewState:
# In real app, this calls an LLM
draft = "AI is transforming how students learn by personalizing content delivery."
return {**state, "draft": draft}
# Node 2: Human review step (interrupt point)
def human_review(state: ReviewState) -> ReviewState:
# In a real app with LangGraph interrupt, this would pause and send state to UI.
# For demonstration, we simulate human input.
print("\n--- HUMAN REVIEW REQUIRED ---")
print("Draft to review:", state["draft"])
# Simulate human feedback (in production, this comes from a UI)
feedback = input("Enter feedback (or press Enter to approve): ").strip()
if not feedback:
feedback = "APPROVED"
return {**state, "human_feedback": feedback}
# Node 3: Finalize based on feedback
def finalize(state: ReviewState) -> ReviewState:
if state["human_feedback"] == "APPROVED":
return {**state, "final_output": state["draft"]}
else:
# In real app, re-run LLM with feedback here
revised = f"Revised based on feedback '{state['human_feedback']}': {state['draft']} [revised]"
return {**state, "final_output": revised}
graph = StateGraph(ReviewState)
graph.add_node("generate", generate_draft)
graph.add_node("review", human_review)
graph.add_node("finalize", finalize)
graph.set_entry_point("generate")
graph.add_edge("generate", "review")
graph.add_edge("review", "finalize")
graph.add_edge("finalize", END)
workflow = graph.compile()
result = workflow.invoke({"draft": ""})
print("\nFinal Output:", result["final_output"])Project Milestones by Level
Beginner Project: Build a two-node graph — one node processes input, another formats the output. Test with 5 different inputs.
Intermediate Project: Build a research workflow with conditional routing: search node → if search succeeds go to writer, if it fails go to fallback → both paths merge at a final formatter node.
Advanced Project: Build a full multi-agent pipeline (researcher → writer → reviewer) with human-in-the-loop review at the writer stage and retry logic on the research node.
Frequently Asked Questions
What is a node in LangGraph?
A node is one step or function inside the workflow.
When should I learn LangGraph?
After you understand Python, prompts, and basic LangChain concepts.
Why do advanced teams prefer graph workflows?
Graph workflows are easier to scale for complex logic because branching, retries, and state transitions are explicit and easier to debug.