LangGraph — Stateful Multi-Agent Orchestration

What is LangGraph?

LangGraph is a library from the LangChain team for building stateful, cyclical agent workflows. While LangChain chains are linear (A → B → C), LangGraph uses a directed graph model where nodes can loop, branch, and hand off to other agents — matching the way real agentic systems need to work.

pip install langgraph

The core insight: agents aren’t chains, they’re loops. A code-writing agent might:

1. Write code
2. Run tests
3. If tests fail → go back to step 1
4. If tests pass → submit PR

LangChain can’t express “go back to step 1 on failure”. LangGraph can.

---

Core Concepts

State

Every LangGraph workflow has a shared State — a typed dictionary that flows through all nodes:

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

class AgentState(TypedDict):
    messages: Annotated[list, operator.add]   # list that accumulates (add new items)
    next_step: str
    attempts: int
    result: str

All nodes read from and write to this state. It’s the single source of truth for the workflow.

Nodes

A node is just a Python function that receives state and returns updated state:

from langchain_openai import ChatOpenAI

llm = ChatOpenAI(model="gpt-4o")

def write_code(state: AgentState) -> AgentState:
    response = llm.invoke(state["messages"])
    return {
        "messages": [response],
        "attempts": state["attempts"] + 1
    }

def run_tests(state: AgentState) -> AgentState:
    # In production: actually run tests
    test_result = "PASS" if state["attempts"] > 1 else "FAIL"
    return {"result": test_result}

Edges — Routing Logic

Edges connect nodes. Conditional edges choose the next node based on state:

def should_retry(state: AgentState) -> str:
    if state["result"] == "PASS":
        return "done"
    elif state["attempts"] >= 3:
        return "give_up"
    else:
        return "retry"

# Build the graph
graph = StateGraph(AgentState)
graph.add_node("write_code", write_code)
graph.add_node("run_tests", run_tests)

graph.set_entry_point("write_code")
graph.add_edge("write_code", "run_tests")

graph.add_conditional_edges(
    "run_tests",
    should_retry,
    {
        "retry": "write_code",   # loop back
        "done": END,
        "give_up": END
    }
)

app = graph.compile()
result = app.invoke({"messages": [("user", "Write a function to reverse a string")], "attempts": 0})

---

Human-in-the-Loop

LangGraph makes it easy to pause and wait for human approval:

from langgraph.checkpoint.memory import MemorySaver
from langgraph.graph import interrupt

checkpointer = MemorySaver()

def review_step(state: AgentState) -> AgentState:
    # Pause here — wait for human to resume
    human_input = interrupt({
        "question": "Does this plan look correct?",
        "plan": state["result"]
    })
    return {"messages": [("human", human_input)]}

graph.add_node("review", review_step)
app = graph.compile(checkpointer=checkpointer, interrupt_before=["review"])

# Run until the interrupt
config = {"configurable": {"thread_id": "session-1"}}
result = app.invoke(initial_state, config=config)
# → execution pauses at "review"

# Human approves and resumes:
app.invoke(Command(resume="Looks good, proceed"), config=config)

---

Multi-Agent Architectures

Supervisor Pattern

A coordinator agent routes tasks to specialist agents:

from langchain_core.messages import HumanMessage

def supervisor(state):
    """Decides which specialist agent to call next."""
    response = llm.invoke([
        ("system", "Route the task to the right specialist: researcher, writer, or reviewer."),
        *state["messages"]
    ])
    return {"next_step": response.content}   # e.g. "researcher"

def researcher(state):
    response = llm.invoke([
        ("system", "You research facts. Be accurate and cite sources."),
        *state["messages"]
    ])
    return {"messages": [response]}

def writer(state):
    response = llm.invoke([
        ("system", "You write clear, engaging content based on research."),
        *state["messages"]
    ])
    return {"messages": [response]}

# Build supervisor graph
graph = StateGraph(AgentState)
graph.add_node("supervisor", supervisor)
graph.add_node("researcher", researcher)
graph.add_node("writer", writer)

graph.set_entry_point("supervisor")
graph.add_conditional_edges(
    "supervisor",
    lambda s: s["next_step"],
    {"researcher": "researcher", "writer": "writer", "done": END}
)
graph.add_edge("researcher", "supervisor")   # always report back
graph.add_edge("writer", "supervisor")

Practical Example: LinkedIn Article Pipeline

class ArticleState(TypedDict):
    topic: str
    messages: Annotated[list, operator.add]
    research: str
    draft: str
    score: float
    final: str

def research_agent(state):
    """Searches news, blogs, and videos for the topic."""
    ...

def brainstorm_agent(state):
    """Generates angle and outline from research."""
    ...

def writer_agent(state):
    """Writes the full article draft."""
    ...

def scorer_agent(state):
    """Scores the draft for clarity, depth, and engagement (0-10)."""
    ...

def rewrite_agent(state):
    """Rewrites low-scoring sections based on scorer feedback."""
    ...

def route_after_score(state):
    return "publish" if state["score"] >= 8.0 else "rewrite"

# Graph: research → brainstorm → write → score → (rewrite loop) → publish

---

Persistence and Resumability

LangGraph supports multiple persistence backends so workflows survive restarts:

# In-memory (development)
from langgraph.checkpoint.memory import MemorySaver
checkpointer = MemorySaver()

# SQLite (simple production)
from langgraph.checkpoint.sqlite import SqliteSaver
checkpointer = SqliteSaver.from_conn_string("checkpoints.db")

# PostgreSQL (production at scale)
from langgraph.checkpoint.postgres import PostgresSaver
checkpointer = PostgresSaver.from_conn_string("postgresql://...")

app = graph.compile(checkpointer=checkpointer)

With checkpointing, you can:

• Pause a workflow and resume it hours later
• Replay a workflow from any checkpoint for debugging
• Support human-in-the-loop reviews with no timeout pressure

---

LangGraph vs LangChain

	LangChain	LangGraph
Structure	Linear chains	Directed graphs with cycles
State	No built-in state	Typed shared state
Loops	Not supported	First-class citizen
Human-in-loop	Manual workarounds	Built-in interrupt()
Multi-agent	Limited	Supervisor, swarm, handoff patterns
Best for	RAG, simple pipelines	Complex agents, production workflows

Use LangChain for RAG and simple linear workflows.
Use LangGraph when your workflow has loops, branches, multiple agents, or needs human review.

---

Deployment: LangGraph Platform

For production, LangGraph Platform provides:

• REST API for your graphs
• Horizontal scaling with managed queues
• Built-in persistence (no manual checkpointer setup)
• Streaming via SSE
• Studio UI for visualising and debugging graph execution

pip install langgraph-cli
langgraph dev   # local dev server with Studio UI

See also: LangChain for the foundational building blocks LangGraph builds on.