Skip to content

LangGraph

LangGraph is the framework for complex, stateful, multi-step LLM workflows. While LangChain handles single/few-step chains, LangGraph gives you explicit control over state, branching, loops, and human-in-the-loop. It models your workflow as a graph of nodes (functions) connected by edges (transitions) — think finite state machines for LLM applications.

Table of Contents

Installation / Setup

pip install langgraph                   # or: uv add langgraph
# langgraph pulls in langchain-core automatically

When to use LangGraph vs LangChain:

Use case Framework
Single-step call (prompt → model → parse) LangChain
Few-step chain (RAG, parallel) LangChain
Multi-step agent with loops LangGraph
Human-in-the-loop LangGraph
Persistent state / checkpointing LangGraph
Complex routing / branching LangGraph
Multi-agent coordination LangGraph

Core Syntax

StateGraph — the core abstraction

from langgraph.graph import StateGraph, START, END
from typing import TypedDict, Annotated
from operator import add

# 1) Define state
class State(TypedDict):
    messages: Annotated[list, add]      # append-only message list
    next_step: str

# 2) Create graph
graph = StateGraph(State)

# 3) Add nodes (functions that modify state)
def call_model(state: State):
    response = llm.invoke(state["messages"])
    return {"messages": [response]}

def decide_next(state: State):
    last = state["messages"][-1].content
    if "tool" in last.lower():
        return {"next_step": "use_tool"}
    return {"next_step": "finish"}

# 4) Register nodes
graph.add_node("agent", call_model)
graph.add_node("decide", decide_next)
graph.add_node("use_tool", execute_tool)

# 5) Add edges (transitions)
graph.add_edge(START, "agent")
graph.add_edge("agent", "decide")

# 6) Conditional routing
graph.add_conditional_edges(
    "decide",
    lambda s: s["next_step"],
    {"use_tool": "use_tool", "finish": END},
)
graph.add_edge("use_tool", "agent")     # loop back after tool

# 7) Compile
app = graph.compile()

# 8) Invoke
result = app.invoke({"messages": [("human", "What's the weather?")]})

Nodes

A node is a function that takes state and returns partial state updates.

def my_node(state: State):
    # read from state
    last_msg = state["messages"][-1].content
    # do something
    result = process(last_msg)
    # return state update (partial)
    return {"messages": [AIMessage(content=result)]}

Edges

# fixed edge: always go from A to B
graph.add_edge("node_a", "node_b")

# conditional edge: route based on state
graph.add_conditional_edges(
    "source_node",
    routing_function,      # state → string key
    {"key_a": "target_a", "key_b": "target_b", "exit": END},
)

# entry point
graph.add_edge(START, "first_node")

# end
graph.add_edge("last_node", END)

State

from typing import TypedDict, Annotated
from operator import add

class State(TypedDict):
    messages: Annotated[list, add]       # list with "add" reducer (append)
    count: int                           # plain field (overwritten)
    results: Annotated[list, add]        # append-only list
Reducer Behavior
add (operator.add) Append to list
Custom function Any merge logic
No annotation Overwrite on each update

Streaming

# stream events
for event in app.stream({"messages": [("human", "Hello")]}):
    for node_name, node_output in event.items():
        print(f"{node_name}: {node_output}")

# stream with updates
for chunk in app.stream({"messages": [("human", "Hello")]}, stream_mode="updates"):
    print(chunk)

Common Operations

Conditional routing

def should_continue(state: State):
    last_msg = state["messages"][-1]
    if hasattr(last_msg, "tool_calls") and last_msg.tool_calls:
        return "tools"
    return END

graph.add_conditional_edges("agent", should_continue, {"tools": "tools", END: END})

Tool node (prebuilt)

from langgraph.prebuilt import ToolNode

tools = [get_weather, search_web]
tool_node = ToolNode(tools)

graph.add_node("tools", tool_node)
# connect: agent → conditional → tools → agent

Human-in-the-loop

from langgraph.checkpoint.memory import MemorySaver

checkpointer = MemorySaver()
app = graph.compile(checkpointer=checkpointer)

# invoke with thread_id for persistence
config = {"configurable": {"thread_id": "user-123"}}
result = app.invoke({"messages": [("human", "Start a task")]}, config)

# interrupt before a node
from langgraph.types import Interrupt
def human_review(state: State):
    # this pauses execution
    raise Interrupt(value={"question": "Should I proceed?", "data": state})

# after human approves, resume
app.invoke(None, config)   # None = resume from interrupt

Persistence (checkpointing)

from langgraph.checkpoint.memory import MemorySaver

checkpointer = MemorySaver()
app = graph.compile(checkpointer=checkpointer)

# each invocation with the same thread_id continues the conversation
config = {"configurable": {"thread_id": "session-1"}}
app.invoke({"messages": [("human", "Hello")]}, config)
app.invoke({"messages": [("human", "Remember my name?")]}, config)  # has history

Multi-agent patterns

# define separate agents as nodes
graph.add_node("researcher", researcher_agent)
graph.add_node("writer", writer_agent)
graph.add_node("reviewer", reviewer_agent)

# researcher → writer → reviewer → (loop back or finish)
graph.add_edge("researcher", "writer")
graph.add_edge("writer", "reviewer")
graph.add_conditional_edges("reviewer", review_route, {"approve": END, "revise": "writer"})

Practical Examples

Example 1 — Simple ReAct agent

from langgraph.graph import StateGraph, START, END
from langgraph.prebuilt import ToolNode
from langchain_openai import ChatOpenAI
from langchain_core.tools import tool
from langchain_core.messages import HumanMessage
from typing import TypedDict, Annotated
from operator import add

@tool
def search(query: str) -> str:
    """Search the web for information."""
    return f"Results for '{query}': The answer is 42."

llm = ChatOpenAI(model="gpt-4o").bind_tools([search])

class State(TypedDict):
    messages: Annotated[list, add]

def agent(state: State):
    response = llm.invoke(state["messages"])
    return {"messages": [response]}

def should_continue(state: State):
    last = state["messages"][-1]
    if hasattr(last, "tool_calls") and last.tool_calls:
        return "tools"
    return END

graph = StateGraph(State)
graph.add_node("agent", agent)
graph.add_node("tools", ToolNode([search]))
graph.add_edge(START, "agent")
graph.add_conditional_edges("agent", should_continue, {"tools": "tools", END: END})
graph.add_edge("tools", "agent")

app = graph.compile()
result = app.invoke({"messages": [HumanMessage(content="What is the meaning of life?")]})
print(result["messages"][-1].content)

Example 2 — Research pipeline with routing

from langgraph.graph import StateGraph, START, END
from typing import TypedDict

class ResearchState(TypedDict):
    topic: str
    findings: str
    draft: str
    approved: bool

def research(state: ResearchState):
    findings = f"Research on {state['topic']}: key findings..."
    return {"findings": findings}

def write_draft(state: ResearchState):
    draft = f"Based on {state['findings']}, here is the article..."
    return {"draft": draft}

def review(state: ResearchState):
    approved = len(state["draft"]) > 100   # simple heuristic
    return {"approved": approved}

def route_after_review(state: ResearchState):
    return "end" if state["approved"] else "revise"

graph = StateGraph(ResearchState)
graph.add_node("research", research)
graph.add_node("write", write_draft)
graph.add_node("review", review)

graph.add_edge(START, "research")
graph.add_edge("research", "write")
graph.add_edge("write", "review")
graph.add_conditional_edges("review", route_after_review, {"end": END, "revise": "write"})

app = graph.compile()
result = app.invoke({"topic": "AI agents"})

Example 3 — Human-in-the-loop approval

from langgraph.graph import StateGraph, START, END
from langgraph.checkpoint.memory import MemorySaver
from langgraph.types import Interrupt
from typing import TypedDict

class ApprovalState(TypedDict):
    action: str
    approved: bool

def propose(state: ApprovalState):
    return {"action": "Deploy to production"}

def human_review(state: ApprovalState):
    raise Interrupt(value={"action": state["action"]})

def execute(state: ApprovalState):
    return {"approved": True}

graph = StateGraph(ApprovalState)
graph.add_node("propose", propose)
graph.add_node("review", human_review)
graph.add_node("execute", execute)

graph.add_edge(START, "propose")
graph.add_edge("propose", "review")
graph.add_edge("review", "execute")

checkpointer = MemorySaver()
app = graph.compile(checkpointer=checkpointer)
config = {"configurable": {"thread_id": "deploy-1"}}

# first call — pauses at human_review
app.invoke({}, config)
# resume after human approves
app.invoke(None, config)

Common Patterns

Pattern: loop with max iterations

def agent(state: State):
    # ... agent logic
    return {"messages": [response], "iterations": state.get("iterations", 0) + 1}

def should_continue(state: State):
    if state.get("iterations", 0) >= 5:
        return END
    if has_tool_calls(state["messages"][-1]):
        return "tools"
    return END

Pattern: map-reduce (parallel nodes)

from langgraph.graph import StateGraph

# fan-out: call multiple nodes in parallel
graph.add_edge("start", "node_a")
graph.add_edge("start", "node_b")    # both run in parallel
graph.add_edge("node_a", "aggregate")
graph.add_edge("node_b", "aggregate")

Pattern: subgraphs

# build a subgraph
subgraph = StateGraph(SubState)
subgraph.add_node("step1", ...)
subgraph.add_node("step2", ...)
sub_compiled = subgraph.compile()

# add as a node in the main graph
main_graph.add_node("subprocess", sub_compiled)

Best Practices

  • Start with LangChain; move to LangGraph only when you need loops, state, or human-in-the-loop.
  • State should be minimal — only what you need to pass between nodes.
  • Use Annotated[list, add] for append-only lists (messages, results).
  • Nodes are just functions — keep them small and testable.
  • Use ToolNode for tool execution — don't reimplement it.
  • MemorySaver is for development; use SqliteSaver or PostgresSaver for production.
  • Streaming is first-class — use app.stream() for real-time updates.
  • Interrupt is the modern way for human-in-the-loop — not the old interrupt_before.

Common Mistakes

Mistake Fix
Using LangGraph for simple chains Use LangChain's LCEL for single-step chains
Mutable state objects State updates must be dicts, not in-place mutations
Missing reducer on list fields Use Annotated[list, add] or lists are overwritten
No END edge Every path must terminate at END
Not compiling the graph Always call graph.compile() before invoke()
Complex graph for simple routing RunnableLambda or with_routing in LangChain is simpler

Troubleshooting

Symptom Cause / Fix
ModuleNotFoundError: No module named 'langgraph' pip install langgraph
State not updating between nodes Return a dict from nodes, not None
Infinite loop Add a max iteration check in your conditional edge
Interrupt not pausing Ensure checkpointer is set on compile()
Graph won't compile Check all nodes are added and edges reference valid node names
Streaming returns nothing Ensure you're using app.stream() not app.invoke()

Official Documentation

Last Updated

2026-09-06 — Initial version.