What You’ll Learn: LangGraph extends LangChain with stateful, graph-based orchestration for complex agent workflows. While basic agents use simple loops, LangGraph lets you build workflows with conditional routing, parallel execution, human-in-the-loop checkpoints, and persistent state. Think of it like upgrading from a single agent to an entire workflow engine — each node is a step, edges define the flow, and the graph handles all the orchestration complexity.
1. StateGraph Basics
SDK Track Note: This is the LangChain SDK Track — a hands-on companion to Foundation Track Part 8 (Stateful Agent Workflows). Read that article first for graph-based workflow concepts.
1.1 State Definition
from typing import Annotated, TypedDict
from langgraph.graph import StateGraph, START, END
from langgraph.graph.message import add_messages
# Define state schema with reducers
class AgentState(TypedDict):
messages: Annotated[list, add_messages] # Append-only message list
current_step: str
iteration_count: int
final_answer: str
# Create graph
graph = StateGraph(AgentState)
1.2 Nodes & Edges
from typing import Annotated, TypedDict
from langchain_openai import ChatOpenAI
from langchain_core.messages import HumanMessage, SystemMessage
from langgraph.graph import StateGraph, START, END
from langgraph.graph.message import add_messages
class AgentState(TypedDict):
messages: Annotated[list, add_messages]
current_step: str
model = ChatOpenAI(model="gpt-4o-mini")
# Define node functions (receive state, return partial state update)
def classify_node(state: AgentState) -> dict:
"""Classify the user's intent."""
response = model.invoke([
SystemMessage(content="Classify intent as: question, task, or chat"),
*state["messages"]
])
return {"current_step": response.content.strip().lower()}
def respond_node(state: AgentState) -> dict:
"""Generate a response."""
response = model.invoke(state["messages"])
return {"messages": [response]}
# Build graph
graph = StateGraph(AgentState)
graph.add_node("classify", classify_node)
graph.add_node("respond", respond_node)
graph.add_edge(START, "classify")
graph.add_edge("classify", "respond")
graph.add_edge("respond", END)
# Compile and run
app = graph.compile()
result = app.invoke({"messages": [HumanMessage(content="What is LangGraph?")]})
print(result["messages"][-1].content)
1.3 Conditional Edges
from typing import Annotated, Literal, TypedDict
from langchain_openai import ChatOpenAI
from langchain_core.messages import HumanMessage, SystemMessage
from langgraph.graph import StateGraph, START, END
from langgraph.graph.message import add_messages
class AgentState(TypedDict):
messages: Annotated[list, add_messages]
intent: str
model = ChatOpenAI(model="gpt-4o-mini", temperature=0)
def classify(state: AgentState) -> dict:
response = model.invoke([
SystemMessage(content="Classify as: question, task, or chitchat. Reply with ONE word."),
*state["messages"]
])
return {"intent": response.content.strip().lower()}
def handle_question(state: AgentState) -> dict:
response = model.invoke([SystemMessage(content="Answer the question thoroughly."), *state["messages"]])
return {"messages": [response]}
def handle_task(state: AgentState) -> dict:
response = model.invoke([SystemMessage(content="Break this task into steps."), *state["messages"]])
return {"messages": [response]}
def handle_chat(state: AgentState) -> dict:
response = model.invoke([SystemMessage(content="Respond casually."), *state["messages"]])
return {"messages": [response]}
# Routing function
def route_intent(state: AgentState) -> Literal["question", "task", "chat"]:
intent = state.get("intent", "chat")
if "question" in intent:
return "question"
elif "task" in intent:
return "task"
return "chat"
# Build graph with conditional edges
graph = StateGraph(AgentState)
graph.add_node("classify", classify)
graph.add_node("question", handle_question)
graph.add_node("task", handle_task)
graph.add_node("chat", handle_chat)
graph.add_edge(START, "classify")
graph.add_conditional_edges("classify", route_intent)
graph.add_edge("question", END)
graph.add_edge("task", END)
graph.add_edge("chat", END)
app = graph.compile()
result = app.invoke({"messages": [HumanMessage(content="How does TCP work?")]})
print(result["messages"][-1].content)
Insurance Claim Processing
An insurance company built their claims pipeline as a LangGraph: intake → document verification → damage assessment → policy lookup → payout calculation → approval routing. Conditional edges route high-value claims to human reviewers. Checkpointing saves state so partially-processed claims survive system restarts. Result: 80% of claims processed end-to-end in under 5 minutes.