LangGraph is LangChain's framework for complex, stateful agent workflows. While LCEL suffices for linear chains, LangGraph enables cycles, conditional branching, and persistent state — essential for production agents.
Simple chains are linear: Input → Processing → Output. But real agent workflows need:
A LangGraph graph consists of three elements:
The shared state that flows through the graph:
from typing import TypedDict, Annotated
from langgraph.graph.message import add_messages
class AgentState(TypedDict):
messages: Annotated[list, add_messages]
current_step: str
results: dict
Functions that process and modify state:
def research_node(state: AgentState) -> dict:
# Perform research
results = search_tool.invoke(state["messages"][-1].content)
return {"results": {"research": results}, "current_step": "analyze"}
Connections between nodes — static or conditional:
from langgraph.graph import StateGraph, END
graph = StateGraph(AgentState)
graph.add_node("research", research_node)
graph.add_node("analyze", analyze_node)
graph.add_node("respond", respond_node)
graph.add_edge("research", "analyze")
graph.add_conditional_edges("analyze", route_function, {
"needs_more": "research", # Cycle back
"ready": "respond" # Continue to respond
})
graph.add_edge("respond", END)
Conditional edges allow dynamic decisions:
def route_function(state: AgentState) -> str:
if state["results"].get("confidence", 0) < 0.8:
return "needs_more"
return "ready"
LangGraph saves state after every node. On errors, you can roll back to the last checkpoint:
from langgraph.checkpoint.memory import MemorySaver
checkpointer = MemorySaver()
app = graph.compile(checkpointer=checkpointer)
Define interrupt points where human approval is required:
app = graph.compile(
checkpointer=checkpointer,
interrupt_before=["send_email", "delete_record"]
)
Practical tip: Draw your workflow as a flowchart before writing code. Every box becomes a node, every arrow an edge. LangGraph makes your architecture explicit — use that.