LangGraph — Build Stateful AI Agents with Graphs (2026)

Introduction

Why LangGraph?

LangGraph extends LangChain with a graph-based execution model for building stateful agents. Rather than running a fixed sequence of steps, LangGraph lets you define a directed graph where each node is a function (or LLM call) and edges can be conditional — routing to different nodes based on state values.

This architecture solves a key problem with simple agent loops: lack of explicit control flow. With LangGraph, you can build agents that retry on failure, branch based on confidence scores, loop until a condition is met, and persist state across sessions. It is the foundation for LangChain's production agent platform, LangGraph Cloud.

Installation

pip install langgraph langchain-openai

Core Concepts: State, Nodes, and Edges

Every LangGraph application has three components:

State — A typed dictionary that flows through the graph and accumulates changes
Nodes — Python functions that receive state, do work, and return state updates
Edges — Connections between nodes; can be fixed or conditional

from typing import TypedDict, Annotated, Literal
from langgraph.graph import StateGraph, END
from langchain_openai import ChatOpenAI
import operator
 
# Define the state schema
class ResearchState(TypedDict):
    question: str
    research_notes: Annotated[list[str], operator.add]  # Accumulates across nodes
    draft: str
    review_feedback: str
    iteration: int
    status: Literal["researching", "drafting", "reviewing", "approved"]
 
llm = ChatOpenAI(model="gpt-4o", temperature=0.1)

Building a Research-Write-Review Agent

from langchain_core.messages import HumanMessage, SystemMessage
 
def research_node(state: ResearchState) -> dict:
    """Node: Research the question and collect notes."""
    response = llm.invoke([
        SystemMessage(content="You are a research analyst. Find key facts, statistics, and insights."),
        HumanMessage(content=f"Research this topic and return 5 key findings:\n{state['question']}")
    ])
 
    return {
        "research_notes": [response.content],
        "status": "drafting"
    }
 
def draft_node(state: ResearchState) -> dict:
    """Node: Write a draft based on research notes."""
    notes = "\n\n".join(state["research_notes"])
    response = llm.invoke([
        SystemMessage(content="You are a technical writer. Write clear, well-structured content."),
        HumanMessage(content=f"Research notes:\n{notes}\n\nWrite a comprehensive answer to: {state['question']}")
    ])
 
    return {
        "draft": response.content,
        "status": "reviewing",
        "iteration": state.get("iteration", 0) + 1
    }
 
def review_node(state: ResearchState) -> dict:
    """Node: Review the draft and approve or request revision."""
    response = llm.invoke([
        SystemMessage(content="""You are an editorial reviewer.
If the draft is complete and accurate, respond with: APPROVED
If revisions are needed, respond with: REVISE: [specific feedback]"""),
        HumanMessage(content=f"Question: {state['question']}\n\nDraft:\n{state['draft']}")
    ])
 
    content = response.content.strip()
    if content.startswith("APPROVED"):
        return {"status": "approved", "review_feedback": ""}
    else:
        feedback = content.replace("REVISE:", "").strip()
        return {"status": "drafting", "review_feedback": feedback}
 
def revise_node(state: ResearchState) -> dict:
    """Node: Revise the draft based on feedback."""
    response = llm.invoke([
        SystemMessage(content="You are a technical writer revising content based on editorial feedback."),
        HumanMessage(content=f"""Original draft:
{state['draft']}
 
Editorial feedback:
{state['review_feedback']}
 
Write an improved version addressing all feedback.""")
    ])
 
    return {
        "draft": response.content,
        "status": "reviewing",
        "iteration": state.get("iteration", 0) + 1
    }
 
# Routing function: determines which node to visit after review
def route_after_review(state: ResearchState) -> str:
    if state["status"] == "approved":
        return "end"
    elif state.get("iteration", 0) &gt;= 3:
        return "end"  # Force stop after 3 iterations to prevent loops
    else:
        return "revise"

Assembling and Running the Graph

from langgraph.graph import StateGraph, END
 
def build_research_graph() -> StateGraph:
    graph = StateGraph(ResearchState)
 
    # Add nodes
    graph.add_node("research", research_node)
    graph.add_node("draft", draft_node)
    graph.add_node("review", review_node)
    graph.add_node("revise", revise_node)
 
    # Set entry point
    graph.set_entry_point("research")
 
    # Fixed edges
    graph.add_edge("research", "draft")
    graph.add_edge("draft", "review")
    graph.add_edge("revise", "review")
 
    # Conditional edge: after review, go to end or revise
    graph.add_conditional_edges(
        "review",
        route_after_review,
        {
            "end": END,
            "revise": "revise"
        }
    )
 
    return graph.compile()
 
# Run the graph
app = build_research_graph()
 
initial_state = {
    "question": "What are the key differences between RAG and fine-tuning for LLM customization?",
    "research_notes": [],
    "draft": "",
    "review_feedback": "",
    "iteration": 0,
    "status": "researching"
}
 
final_state = app.invoke(initial_state)
print(f"Status: {final_state['status']}")
print(f"Iterations: {final_state['iteration']}")
print(f"\nFinal answer:\n{final_state['draft']}")

Persistent State with Checkpointing

LangGraph's checkpointing system allows agents to resume from any point and maintain state across sessions:

from langgraph.checkpoint.sqlite import SqliteSaver
from langgraph.checkpoint.memory import MemorySaver
 
# In-memory for development
memory_checkpointer = MemorySaver()
 
# SQLite for single-instance production
sqlite_checkpointer = SqliteSaver.from_conn_string("./agent_state.db")
 
# Compile with persistence
persistent_app = build_research_graph().compile(
    checkpointer=sqlite_checkpointer
)
 
# Each thread_id is an independent session
config = {"configurable": {"thread_id": "user-session-abc123"}}
 
# First invocation starts the session
result1 = persistent_app.invoke(initial_state, config=config)
 
# Subsequent invocations continue from where the last left off
# (only meaningful if the graph has interrupt points)
state_snapshot = persistent_app.get_state(config)
print(f"Current state keys: {list(state_snapshot.values.keys())}")

Human-in-the-Loop with Interrupts

LangGraph supports pausing execution at any node to wait for human input:

from langgraph.graph import StateGraph, END
 
class ApprovalState(TypedDict):
    task: str
    plan: str
    human_approved: bool
    result: str
 
def plan_node(state: ApprovalState) -> dict:
    response = llm.invoke([
        HumanMessage(content=f"Create a detailed execution plan for: {state['task']}")
    ])
    return {"plan": response.content}
 
def execute_node(state: ApprovalState) -> dict:
    response = llm.invoke([
        HumanMessage(content=f"Execute this plan:\n{state['plan']}")
    ])
    return {"result": response.content}
 
def should_execute(state: ApprovalState) -> str:
    return "execute" if state.get("human_approved", False) else "end"
 
graph = StateGraph(ApprovalState)
graph.add_node("plan", plan_node)
graph.add_node("execute", execute_node)
graph.set_entry_point("plan")
graph.add_conditional_edges("plan", should_execute, {"execute": "execute", "end": END})
graph.add_edge("execute", END)
 
# Interrupt after 'plan' to wait for human approval
approval_app = graph.compile(
    checkpointer=MemorySaver(),
    interrupt_after=["plan"]
)
 
config = {"configurable": {"thread_id": "approval-flow-1"}}
 
# Run until interrupt
state = approval_app.invoke({"task": "Deploy new ML model to production", "human_approved": False}, config)
print("Plan generated, waiting for approval:")
print(approval_app.get_state(config).values["plan"])
 
# Human approves — resume execution
approval_app.update_state(config, {"human_approved": True})
final = approval_app.invoke(None, config)
print(f"\nResult: {final['result']}")

Streaming Graph Execution

# Stream intermediate results as each node completes
for event in app.stream(initial_state, stream_mode="updates"):
    for node_name, node_output in event.items():
        print(f"\n[{node_name}] completed:")
        if "status" in node_output:
            print(f"  Status: {node_output['status']}")
        if "draft" in node_output:
            print(f"  Draft length: {len(node_output.get('draft', ''))} chars")

Common Mistakes

Mutable default state values — Always use Annotated[list, operator.add] for lists that accumulate across nodes; plain lists get overwritten.
Missing termination conditions in loops — Conditional edges that loop must have a stopping condition (e.g., max iterations).
Forgetting to set entry point — graph.set_entry_point() is required; there is no default.
State type mismatches — All node return dicts must use keys that exist in the state TypedDict.
Not using checkpointing in production — Without a checkpointer, all state is lost if the process restarts.
Overly complex graphs — Start simple; add nodes and edges only when a simpler design fails.

Best Practices

Use Annotated[list, operator.add] for any list field that should accumulate values across nodes
Set max iteration guards in all routing functions that can loop
Stream graph execution during development to understand the execution path
Use SQLite checkpointing for single-instance apps; use Redis or PostgreSQL checkpointing for distributed deployments
Draw your graph visually using app.get_graph().draw_mermaid() before coding it

Key Takeaways

LangGraph models agents as directed graphs where state flows through nodes connected by conditional edges
State is typed with TypedDict; use Annotated[list, operator.add] for fields that accumulate values across nodes
Conditional edges with routing functions enable dynamic branching — retry on failure, escalate on low confidence, loop until done
Checkpointing with MemorySaver, SqliteSaver, or RedisSaver enables session persistence and resumable workflows
Human-in-the-loop is implemented with interrupt_after — the graph pauses at specified nodes and waits for external input
Always include a maximum iteration guard in any routing function that can loop to prevent infinite execution
Streaming (app.stream()) surfaces intermediate results as each node completes, enabling real-time progress updates
LangGraph is best suited for complex, stateful, multi-step agents where control flow matters — not for simple single-turn LLM calls

Introduction#

Why LangGraph?#

Installation#

Core Concepts: State, Nodes, and Edges#

Building a Research-Write-Review Agent#

Assembling and Running the Graph#

Persistent State with Checkpointing#

Human-in-the-Loop with Interrupts#

Streaming Graph Execution#

Common Mistakes#

Best Practices#

Key Takeaways#

Related reading