LangGraph development for stateful multi-agent applications, cyclic workflows, conditional routing, human-in-the-loop patterns, and persistent state management...
LangGraph is LangChain's framework for building stateful, multi-actor applications with LLMs. It enables cyclic graphs, conditional branching, persistent state management, and human-in-the-loop workflows.
Key capabilities: StateGraph for workflow orchestration, checkpointers for persistence, prebuilt ReAct agents, streaming at multiple granularities, and LangGraph Platform for managed deployment.
This skill is loaded by backend-developer when:
langgraph in requirements.txt, pyproject.toml, or setup.py@langchain/langgraph in package.json dependenciesfrom langgraph.graph import StateGraphLANGSMITH_API_KEY, LANGGRAPH_API_URL, or LANGCHAIN_API_KEY present| Package | Description |
|---|---|
langgraph |
Core StateGraph, MessageGraph, prebuilt components |
langgraph-checkpoint-postgres |
PostgreSQL checkpointer for production |
langgraph-checkpoint-sqlite |
SQLite checkpointer with async support |
langgraph-sdk |
Python client for LangGraph Platform API |
@langchain/langgraph |
TypeScript/JavaScript implementation |
Compatibility: LangGraph 0.2.x requires langchain-core>=0.3.0.
"""LangGraph Quick Start - Python"""
from langgraph.graph import StateGraph, START, END
from typing import TypedDict
# 1. Define state schema
class State(TypedDict):
messages: list[str]
count: int
# 2. Define node functions
def process(state: State) -> dict:
"""Process node - increments counter."""
return {"count": state["count"] + 1}
def respond(state: State) -> dict:
"""Response node - adds message to state."""
return {"messages": state["messages"] + [f"Processed {state['count']} times"]}
# 3. Build graph
graph = StateGraph(State)
graph.add_node("process", process)
graph.add_node("respond", respond)
graph.add_edge(START, "process")
graph.add_edge("process", "respond")
graph.add_edge("respond", END)
# 4. Compile and run
app = graph.compile()
result = app.invoke({"messages": [], "count": 0})
# {'messages': ['Processed 1 times'], 'count': 1}
"""LangGraph with ChatOpenAI and tools."""
from langgraph.graph import StateGraph, START, MessagesState
from langgraph.prebuilt import ToolNode, tools_condition
from langchain_openai import ChatOpenAI
from langchain_core.tools import tool
@tool
def get_weather(city: str) -> str:
"""Get weather for a city."""
return f"Weather in {city}: 72F, sunny"
llm = ChatOpenAI(model="gpt-4o")
tools = [get_weather]
llm_with_tools = llm.bind_tools(tools)
def agent(state: MessagesState) -> dict:
response = llm_with_tools.invoke(state["messages"])
return {"messages": [response]}
graph = StateGraph(MessagesState)
graph.add_node("agent", agent)
graph.add_node("tools", ToolNode(tools))
graph.add_edge(START, "agent")
graph.add_conditional_edges("agent", tools_condition)
graph.add_edge("tools", "agent")
app = graph.compile()
result = app.invoke({"messages": [("user", "What's the weather in NYC?")]})
import { StateGraph, START, END, Annotation } from "@langchain/langgraph";
const State = Annotation.Root({
messages: Annotation<string[]>({
default: () => [],
reducer: (curr, update) => [...curr, ...update],
}),
count: Annotation<number>({ default: () => 0 }),
});
const process = (state: typeof State.State) => ({
count: state.count + 1,
});
const respond = (state: typeof State.State) => ({
messages: [`Processed ${state.count} times`],
});
const graph = new StateGraph(State)
.addNode("process", process)
.addNode("respond", respond)
.addEdge(START, "process")
.addEdge("process", "respond")
.addEdge("respond", END);
const app = graph.compile();
const result = await app.invoke({ messages: [], count: 0 });
The core abstraction - a directed graph where nodes read and write to shared state.
from langgraph.graph import StateGraph
graph = StateGraph(State) # State is a TypedDict
Defines the shape of data flowing through the graph.
from typing import TypedDict, Annotated
from langgraph.graph import add_messages
class State(TypedDict):
messages: Annotated[list, add_messages] # With reducer
context: str # Simple overwrite
Functions that receive state and return partial updates.
def my_node(state: State) -> dict:
return {"context": f"{state['context']} - processed"}
Connections between nodes - unconditional or conditional.
# Unconditional
graph.add_edge("node_a", "node_b")
# Conditional
graph.add_conditional_edges("router", routing_function)
| Node | Purpose |
|---|---|
START |
Entry point of the graph |
END |
Terminal node, execution stops |
| Component | Import | Purpose |
|---|---|---|
create_react_agent |
langgraph.prebuilt |
Full ReAct agent with tool loop |
ToolNode |
langgraph.prebuilt |
Execute tools from AI messages |
tools_condition |
langgraph.prebuilt |
Route: tools called vs end |
MessagesState |
langgraph.graph |
Prebuilt chat state schema |
from langgraph.prebuilt import create_react_agent
from langchain_openai import ChatOpenAI
from langchain_core.tools import tool
@tool
def search(query: str) -> str:
"""Search the web."""
return f"Results for: {query}"
llm = ChatOpenAI(model="gpt-4o")
agent = create_react_agent(llm, [search])
result = agent.invoke({"messages": [("user", "Search for LangGraph")]})
graph.add_edge(START, "extract")
graph.add_edge("extract", "transform")
graph.add_edge("transform", "load")
graph.add_edge("load", END)
def router(state: State) -> str:
if state["intent"] == "search":
return "search_node"
return "default_node"
graph.add_conditional_edges("classifier", router)
def check_done(state):
return END if state["done"] else "process"
graph.add_edge(START, "process")
graph.add_edge("process", "check")
graph.add_conditional_edges("check", check_done)
from langgraph.checkpoint.memory import MemorySaver
# Development - in-memory (lost on restart)
memory = MemorySaver()
app = graph.compile(checkpointer=memory)
config = {"configurable": {"thread_id": "user-123"}}
result = app.invoke({"messages": [("user", "Hi")]}, config)
For production persistence, see REFERENCE.md for PostgreSQL/SQLite setup.
# Install
pip install langgraph langgraph-checkpoint-postgres
# Verify
python -c "import langgraph; print(langgraph.__version__)"
# LangGraph Platform
langgraph init # Initialize project
langgraph dev # Start dev server
langgraph build # Build for deployment
langgraph up # Deploy with Docker
export LANGSMITH_API_KEY="ls-..."
export LANGCHAIN_TRACING_V2="true"
export OPENAI_API_KEY="sk-..."
| Error | Solution |
|---|---|
InvalidUpdateError |
Node returned keys not in state schema |
GraphRecursionError |
Add exit condition or increase recursion_limit |
ValueError: Checkpointer required |
Add checkpointer to compile() for thread ops |
Load REFERENCE.md for:
| Agent | Use Case |
|---|---|
backend-developer |
Graph implementation, state design |
deep-debugger |
Execution issues, state bugs |
tech-lead-orchestrator |
Multi-agent architecture |
code-reviewer |
Graph logic review |
Progressive Disclosure: Start here for quick implementation. Load REFERENCE.md for comprehensive patterns, production setup, and advanced features.