Chapter 12: Building with LangGraph
Building with LangGraph in Building Agentic AI Systems.
Learning Objectives
By the end of this chapter, you will be able to:
- Explain the agentic AI concept behind Building with LangGraph.
- Apply Building with LangGraph to design reliable, production-grade agent systems.
- Recognize operational trade-offs in tool use, orchestration, safety, and cost.
Chapter 12: Building with LangGraph
Graph-based state machines, checkpointing, HITL, and subgraphs
Why LangGraph?
LangGraph is a graph-based agent orchestration framework from LangChain. Instead of defining agent behavior in a sequential loop, you define it as a directed graph where nodes are functions (actions) and edges are transitions between them — including conditional edges that route based on state.
Standard Agent Loop
- Linear while-loop
- Hard to add branching
- No built-in checkpointing
- Human-in-the-loop requires custom code
LangGraph
- Directed graph (nodes + edges)
- Conditional routing via edge functions
- Built-in checkpointing (durable execution)
- HITL via interrupt_before/interrupt_after
- Subgraphs for modular composition
Production adoption
LangGraph 1.0+ reached production stability in 2025. Used at Uber, LinkedIn, Klarna, and Replit. ~39M monthly PyPI downloads (early 2026).
Core Concepts
The key concept: every node receives the full state and returns a partial state update. LangGraph merges the update into the shared state automatically. This means nodes are pure functions — they don't need to know how state is passed around.
Building a ReAct Agent in LangGraph
from typing import Annotated, Literal
from typing_extensions import TypedDict
from langgraph.graph import StateGraph, END
from langgraph.graph.message import add_messages
from langgraph.prebuilt import ToolNode
from langchain_openai import ChatOpenAI
from langchain_core.tools import tool
# ── 1. Define State ──────────────────────────────────────────────────────────
class AgentState(TypedDict):
messages: Annotated[list, add_messages] # add_messages is a reducer — appends, doesn't replace
# ── 2. Define Tools ──────────────────────────────────────────────────────────
@tool
def search_web(query: str) -> str:
"""Search the web for current information."""
return f"[Mock result for '{query}']" # replace with real search API
tools = [search_web]
# ── 3. Define Nodes ──────────────────────────────────────────────────────────
llm = ChatOpenAI(model="gpt-4o").bind_tools(tools)
def agent_node(state: AgentState) -> dict:
"""Call LLM with current messages; return updated messages."""
response = llm.invoke(state["messages"])
return {"messages": [response]}
def should_continue(state: AgentState) -> Literal["tools", "__end__"]:
"""Conditional edge: route to tools if LLM made tool calls, else end."""
last_message = state["messages"][-1]
if hasattr(last_message, "tool_calls") and last_message.tool_calls:
return "tools"
return "__end__"
# ── 4. Build Graph ───────────────────────────────────────────────────────────
graph = StateGraph(AgentState)
graph.add_node("agent", agent_node)
graph.add_node("tools", ToolNode(tools)) # prebuilt node handles tool dispatch
graph.set_entry_point("agent")
graph.add_conditional_edges("agent", should_continue, {"tools": "tools", "__end__": END})
graph.add_edge("tools", "agent") # after tools, always go back to agent
app = graph.compile()
# ── 5. Run ───────────────────────────────────────────────────────────────────
result = app.invoke({"messages": [{"role": "user", "content": "What is the latest news about LLMs?"}]})
print(result["messages"][-1].content)
LLM decision
Execute calls
Production Features
Checkpointing
LangGraph checkpoints state after every node execution. If the graph crashes, it resumes from the last checkpoint — no work is lost. This is critical for long-running workflows (minutes to hours).
from langgraph.checkpoint.sqlite import SqliteSaver
checkpointer = SqliteSaver.from_conn_string("./checkpoints.db")
app = graph.compile(checkpointer=checkpointer)
# Run with a thread_id — LangGraph uses this as the checkpoint key
config = {"configurable": {"thread_id": "user-123-task-456"}}
result = app.invoke({"messages": [...]}, config=config)
# Resume the same thread later — LangGraph loads from checkpoint
result2 = app.invoke({"messages": [...]}, config=config)
Human-in-the-Loop (HITL)
LangGraph can pause execution before or after any node, waiting for human review. The state is persisted; when the human approves (or edits), execution resumes.
# Compile with interrupt_before — pause before the "tools" node
app = graph.compile(
checkpointer=checkpointer,
interrupt_before=["tools"] # pause here for human review
)
# Run until interrupt
state = app.invoke({"messages": [...]}, config=config)
# Human reviews the pending tool calls in state["messages"][-1].tool_calls
# Then resume (optionally with updated state)
state = app.invoke(None, config=config) # None = use existing state, resume
Subgraphs for modular composition
A LangGraph subgraph is a compiled graph used as a node inside a parent graph. This enables true modularity: the "research" subgraph is developed independently, tested in isolation, and plugged into the supervisor graph. State schemas must be compatible at the boundary.
Chapter 12 Quiz
1. In LangGraph, what does a "conditional edge" do?
2. What does add_messages do in the AgentState definition?
3. Why is Human-in-the-Loop (HITL) implemented via interrupt_before rather than a separate API endpoint?