LangChain chains flow in one direction. Input enters, output exits, done.
Real agent workflows are not linear. A plan might need revision. A search might return empty results and require a different approach. A code review might fail and send the work back for fixes. A task might need to branch differently depending on what kind of input it receives.
LangGraph models agent workflows as directed graphs. Nodes are actions. Edges are conditional transitions. The agent's state flows through the graph, taking different paths based on what it finds at each step.
The result: agent workflows that are inspectable, debuggable, resumable from any node, and capable of complex conditional logic without becoming spaghetti code.
What LangGraph Adds to LangChain
print("LangGraph: What It Adds")
print()
print("LangChain gives you: components (LLMs, retrievers, tools)")
print("LangGraph gives you: orchestration as a directed graph")
print()
differences = {
"Control flow": ("Linear chain", "Conditional graph with loops and branches"),
"State": ("Passed between steps", "Typed state shared across all nodes"),
"Debugging": ("Trace the chain", "Visualize the graph, step through nodes"),
"Resume": ("Start over", "Checkpoint any node, resume from there"),
"Parallelism": ("Sequential only", "Parallel branches in the graph"),
"Human-in-loop": ("Not supported", "Pause graph, wait for human, resume"),
}
print(f"{'Feature':<18} {'LangChain':<30} {'LangGraph'}")
print("=" * 80)
for feature, (lc, lg) in differences.items():
print(f"{feature:<18} {lc:<30} {lg}")
print()
print("Install:")
print(" pip install langgraph langchain-core langchain-anthropic")
Core Concepts
import os
import json
from typing import TypedDict, Annotated, List, Optional, Literal
from langgraph.graph import StateGraph, END, START
from langgraph.checkpoint.memory import MemorySaver
from langchain_anthropic import ChatAnthropic
from langchain_core.messages import HumanMessage, AIMessage, BaseMessage
from langchain_core.prompts import ChatPromptTemplate
from langchain_core.output_parsers import StrOutputParser
import anthropic
import warnings
warnings.filterwarnings("ignore")
llm = ChatAnthropic(model="claude-3-5-haiku-20241022",
api_key=os.environ.get("ANTHROPIC_API_KEY"),
max_tokens=500)
parser = StrOutputParser()
class ConversationState(TypedDict):
messages: Annotated[List[BaseMessage], lambda x, y: x + y]
user_input: str
intent: str
response: str
needs_retrieval: bool
print("LangGraph Core Concepts:")
print()
print("1. STATE: A TypedDict shared across all nodes")
print(" - Nodes read from state and write back to state")
print(" - Annotated fields define how values merge (append vs replace)")
print()
print("2. NODES: Functions that take state and return updated state")
print(" - def my_node(state: MyState) -> dict:")
print(" # do work, return updates")
print(" return {'field': new_value}")
print()
print("3. EDGES: Connections between nodes")
print(" - Unconditional: always go to next node")
print(" - Conditional: function decides which node comes next")
print()
print("4. COMPILE: StateGraph().compile() turns the graph into a runnable")
Minimal Example: Intent Router
class RouterState(TypedDict):
user_input: str
intent: str
response: str
def detect_intent(state: RouterState) -> dict:
"""Node 1: Classify the user's intent."""
prompt = f"""Classify this input into exactly one category:
- math: involves calculation or numbers
- code: involves programming
- general: anything else
Input: {state['user_input']}
Respond with only the category name."""
intent = (llm | parser).invoke(prompt).strip().lower()
if intent not in ["math", "code", "general"]:
intent = "general"
print(f" [detect_intent] → '{intent}'")
return {"intent": intent}
def handle_math(state: RouterState) -> dict:
"""Node 2a: Handle math queries."""
response = (llm | parser).invoke(
f"Solve this math problem clearly: {state['user_input']}")
print(f" [handle_math] → answered")
return {"response": response}
def handle_code(state: RouterState) -> dict:
"""Node 2b: Handle code queries."""
response = (llm | parser).invoke(
f"Answer this coding question with example code: {state['user_input']}")
print(f" [handle_code] → answered")
return {"response": response}
def handle_general(state: RouterState) -> dict:
"""Node 2c: Handle general queries."""
response = (llm | parser).invoke(state["user_input"])
print(f" [handle_general] → answered")
return {"response": response}
def route_by_intent(state: RouterState) -> Literal["handle_math", "handle_code", "handle_general"]:
"""Conditional edge: decide which handler to call."""
return f"handle_{state['intent']}" if state["intent"] in ["math", "code"] else "handle_general"
graph = StateGraph(RouterState)
graph.add_node("detect_intent", detect_intent)
graph.add_node("handle_math", handle_math)
graph.add_node("handle_code", handle_code)
graph.add_node("handle_general", handle_general)
graph.add_edge(START, "detect_intent")
graph.add_conditional_edges("detect_intent", route_by_intent)
graph.add_edge("handle_math", END)
graph.add_edge("handle_code", END)
graph.add_edge("handle_general", END)
router_app = graph.compile()
print("Intent Router Graph:")
test_inputs = [
"What is 15% of 840?",
"How do I reverse a list in Python?",
"What is the capital of Japan?",
]
for inp in test_inputs:
print(f"\nInput: '{inp}'")
result = router_app.invoke({"user_input": inp, "intent": "", "response": ""})
print(f"Response: {result['response'][:120]}...")
Multi-Step Agent with Loops
class ResearchState(TypedDict):
topic: str
search_queries: List[str]
findings: List[str]
draft: str
review_score: int
final_output: str
iteration: int
def plan_research(state: ResearchState) -> dict:
"""Generate search queries for the topic."""
response = (llm | parser).invoke(
f"Generate 3 specific search queries to research: {state['topic']}\n"
f"Output as a numbered list, one query per line.")
queries = [line.strip().lstrip("123. ").strip()
for line in response.split("\n")
if line.strip() and line.strip()[0].isdigit()][:3]
print(f" [plan] Generated {len(queries)} queries")
return {"search_queries": queries}
def execute_search(state: ResearchState) -> dict:
"""Simulate searching and gathering findings."""
findings = []
for i, query in enumerate(state["search_queries"], 1):
finding = (llm | parser).invoke(
f"Provide 2-3 key facts about: {query}\nBe specific and concise.")
findings.append(f"Query {i}: {query}\n{finding}")
print(f" [search {i}] gathered findings")
return {"findings": findings}
def write_draft(state: ResearchState) -> dict:
"""Write a draft based on findings."""
findings_text = "\n\n".join(state["findings"])
draft = (llm | parser).invoke(
f"Write a 3-paragraph summary about '{state['topic']}' "
f"based on these findings:\n\n{findings_text}")
print(f" [write] draft written ({len(draft.split())} words)")
return {"draft": draft, "iteration": state.get("iteration", 0) + 1}
def review_draft(state: ResearchState) -> dict:
"""Score the draft quality."""
response = (llm | parser).invoke(
f"Rate this text quality 1-10 (just the number):\n\n{state['draft']}")
import re
score_match = re.search(r'\b([0-9]|10)\b', response)
score = int(score_match.group()) if score_match else 7
print(f" [review] score: {score}/10 (iteration {state.get('iteration', 1)})")
return {"review_score": score}
def revise_draft(state: ResearchState) -> dict:
"""Improve the draft based on review."""
revised = (llm | parser).invoke(
f"Improve this text. Make it clearer and more informative:\n\n{state['draft']}")
print(f" [revise] draft improved")
return {"draft": revised}
def finalize(state: ResearchState) -> dict:
"""Package the final output."""
print(f" [finalize] approved after {state.get('iteration', 1)} iteration(s)")
return {"final_output": state["draft"]}
def should_revise(state: ResearchState) -> Literal["revise_draft", "finalize"]:
"""Decide: revise or finalize based on score and iteration count."""
iteration = state.get("iteration", 1)
score = state.get("review_score", 7)
if score < 8 and iteration < 3:
print(f" [routing] score {score} < 8, iteration {iteration} < 3 → revise")
return "revise_draft"
print(f" [routing] score {score} or max iterations → finalize")
return "finalize"
research_graph = StateGraph(ResearchState)
research_graph.add_node("plan_research", plan_research)
research_graph.add_node("execute_search", execute_search)
research_graph.add_node("write_draft", write_draft)
research_graph.add_node("review_draft", review_draft)
research_graph.add_node("revise_draft", revise_draft)
research_graph.add_node("finalize", finalize)
research_graph.add_edge(START, "plan_research")
research_graph.add_edge("plan_research", "execute_search")
research_graph.add_edge("execute_search", "write_draft")
research_graph.add_edge("write_draft", "review_draft")
research_graph.add_conditional_edges("review_draft", should_revise)
research_graph.add_edge("revise_draft", "review_draft")
research_graph.add_edge("finalize", END)
research_app = research_graph.compile()
print("\nResearch Agent with Review Loop:")
print("=" * 55)
result = research_app.invoke({
"topic": "transformer attention mechanisms",
"search_queries": [],
"findings": [],
"draft": "",
"review_score": 0,
"final_output": "",
"iteration": 0,
})
print(f"\nFinal Output Preview:\n{result['final_output'][:300]}...")
print(f"\nCompleted in {result['iteration']} iteration(s)")
Checkpointing: Pause and Resume
print("\nLangGraph Checkpointing: Pause and Resume Any State")
print()
memory = MemorySaver()
checkpointed_app = research_graph.compile(checkpointer=memory)
thread_config = {"configurable": {"thread_id": "research_session_001"}}
print("Running with checkpointing enabled...")
result = checkpointed_app.invoke(
{
"topic": "neural network activation functions",
"search_queries": [],
"findings": [],
"draft": "",
"review_score": 0,
"final_output": "",
"iteration": 0,
},
config=thread_config
)
state_snapshot = checkpointed_app.get_state(thread_config)
print(f"\nCheckpoint saved. State snapshot:")
print(f" Next nodes available: {state_snapshot.next}")
print(f" Topic: {state_snapshot.values.get('topic', 'N/A')}")
print(f" Iteration: {state_snapshot.values.get('iteration', 0)}")
print()
print("To resume from this checkpoint later:")
print(" result = app.invoke(None, config=thread_config)")
print(" (pass None to resume from saved state)")
Human-in-the-Loop
print("\nHuman-in-the-Loop: Pause for Human Approval")
print()
from langgraph.types import interrupt
class ApprovalState(TypedDict):
content: str
approved: bool
feedback: str
def generate_content(state: ApprovalState) -> dict:
content = (llm | parser).invoke(
"Write a 2-sentence marketing tagline for a machine learning platform.")
print(f" [generate] content ready for review")
return {"content": content, "approved": False}
def human_review(state: ApprovalState) -> dict:
"""This node PAUSES the graph and waits for human input."""
print(f"\n [PAUSED] Human review required.")
print(f" Content: {state['content']}")
print(f" (In production: send notification, wait for response via API)")
decision = interrupt({
"message": "Please review and approve/reject this content",
"content": state["content"],
"actions": ["approve", "reject"],
})
return {"approved": decision == "approve", "feedback": str(decision)}
def handle_approved(state: ApprovalState) -> dict:
print(f" [approved] Content published")
return {}
def handle_rejected(state: ApprovalState) -> dict:
print(f" [rejected] Content sent back for revision")
return {}
def route_approval(state: ApprovalState) -> Literal["handle_approved", "handle_rejected"]:
return "handle_approved" if state["approved"] else "handle_rejected"
hitl_graph = StateGraph(ApprovalState)
hitl_graph.add_node("generate_content", generate_content)
hitl_graph.add_node("human_review", human_review)
hitl_graph.add_node("handle_approved", handle_approved)
hitl_graph.add_node("handle_rejected", handle_rejected)
hitl_graph.add_edge(START, "generate_content")
hitl_graph.add_edge("generate_content", "human_review")
hitl_graph.add_conditional_edges("human_review", route_approval)
hitl_graph.add_edge("handle_approved", END)
hitl_graph.add_edge("handle_rejected", END)
print("Human-in-loop graph structure:")
print(" generate_content → human_review → [approved: publish] OR [rejected: revise]")
print()
print("In production:")
print(" 1. Graph runs to 'human_review' node and pauses")
print(" 2. State saved to checkpointer (Redis, Postgres)")
print(" 3. Human receives notification (email, Slack, UI)")
print(" 4. Human responds via API endpoint")
print(" 5. Graph resumes from checkpoint with human's decision")
Parallel Branches
print("\nParallel Execution in LangGraph")
print()
class ParallelState(TypedDict):
topic: str
perspective1: str
perspective2: str
perspective3: str
synthesis: str
def get_technical(state: ParallelState) -> dict:
resp = (llm | parser).invoke(
f"Give a 2-sentence technical perspective on: {state['topic']}")
return {"perspective1": resp}
def get_business(state: ParallelState) -> dict:
resp = (llm | parser).invoke(
f"Give a 2-sentence business perspective on: {state['topic']}")
return {"perspective2": resp}
def get_ethical(state: ParallelState) -> dict:
resp = (llm | parser).invoke(
f"Give a 2-sentence ethical perspective on: {state['topic']}")
return {"perspective3": resp}
def synthesize(state: ParallelState) -> dict:
synthesis = (llm | parser).invoke(
f"Synthesize these three perspectives into one balanced paragraph:\n\n"
f"Technical: {state['perspective1']}\n"
f"Business: {state['perspective2']}\n"
f"Ethical: {state['perspective3']}")
return {"synthesis": synthesis}
parallel_graph = StateGraph(ParallelState)
parallel_graph.add_node("get_technical", get_technical)
parallel_graph.add_node("get_business", get_business)
parallel_graph.add_node("get_ethical", get_ethical)
parallel_graph.add_node("synthesize", synthesize)
parallel_graph.add_edge(START, "get_technical")
parallel_graph.add_edge(START, "get_business")
parallel_graph.add_edge(START, "get_ethical")
parallel_graph.add_edge("get_technical", "synthesize")
parallel_graph.add_edge("get_business", "synthesize")
parallel_graph.add_edge("get_ethical", "synthesize")
parallel_graph.add_edge("synthesize", END)
parallel_app = parallel_graph.compile()
print("Running three parallel perspective nodes:")
import time
start = time.time()
result = parallel_app.invoke({
"topic": "using AI to make hiring decisions",
"perspective1": "",
"perspective2": "",
"perspective3": "",
"synthesis": ""
})
elapsed = time.time() - start
print(f"Completed in {elapsed:.1f}s")
print(f"\nSynthesis:\n{result['synthesis'][:300]}...")
Reference Links
print("\nLangGraph Reference Links:")
print()
refs = {
"Official Documentation": [
("LangGraph docs", "langchain-ai.github.io/langgraph"),
("LangGraph tutorials", "langchain-ai.github.io/langgraph/tutorials/introduction"),
("How-To guides", "langchain-ai.github.io/langgraph/how-tos"),
("Conceptual guides", "langchain-ai.github.io/langgraph/concepts"),
("LangGraph Cloud (deployment)", "langchain-ai.github.io/langgraph/cloud"),
],
"Key Concepts Deep Dives": [
("State management", "langchain-ai.github.io/langgraph/concepts/low_level/#state"),
("Checkpointing / persistence", "langchain-ai.github.io/langgraph/concepts/persistence"),
("Human-in-the-loop", "langchain-ai.github.io/langgraph/concepts/human_in_the_loop"),
("Multi-agent with LangGraph", "langchain-ai.github.io/langgraph/tutorials/multi_agent"),
("Streaming", "langchain-ai.github.io/langgraph/how-tos/streaming-tokens"),
],
"Cheat Sheets": [
("LangGraph GitHub", "github.com/langchain-ai/langgraph"),
("LangGraph Python API ref", "langchain-ai.github.io/langgraph/reference/graphs"),
("StateGraph methods", "langchain-ai.github.io/langgraph/reference/graphs/#langgraph.graph.StateGraph"),
("LangSmith tracing with LangGraph","docs.smith.langchain.com/how_to_guides/tracing/langgraph"),
],
"Examples": [
("Agent supervisor pattern", "langchain-ai.github.io/langgraph/tutorials/multi_agent/agent_supervisor"),
("Plan-and-execute agent", "langchain-ai.github.io/langgraph/tutorials/plan-and-execute/plan-and-execute"),
("ReAct agent from scratch", "langchain-ai.github.io/langgraph/tutorials/introduction"),
("RAG with LangGraph", "langchain-ai.github.io/langgraph/tutorials/rag/langgraph_agentic_rag"),
],
}
for category, links in refs.items():
print(f" {category}:")
for name, url in links:
print(f" • {name:<45} {url}")
print()
Try This
Create langgraph_practice.py.
Part 1: build the intent router from this post. Extend it with two more intents: "translation" and "summarization." Add appropriate handler nodes. Test with 10 different inputs and verify routing is correct.
Part 2: build a workflow with a revision loop. Create a writing agent that: generates content → evaluates readability → revises if score below threshold → repeats up to 3 times. Track iteration count in state. Does quality improve across iterations?
Part 3: add checkpointing. Compile your graph with MemorySaver(). Run a workflow partway, inspect the checkpoint state with app.get_state(). Modify one field in the state and resume with app.invoke(None, config=thread_config). Verify the modified state is used.
Part 4: parallel branches. Build a graph that simultaneously generates three different versions of a piece of writing (formal, casual, technical), then synthesizes them into one final version. Time the parallel vs sequential execution.
What's Next
You can now build complex stateful agent workflows. The next post is the Phase 10 capstone: build a complete autonomous research assistant that plans, searches, writes, reviews, and publishes reports autonomously using everything from posts 101 to 106.
Top comments (0)