LangGraph StateGraph Workflows

Build cyclic, deterministic, multi-agent workflows with state persistence using LangGraph

⏱ 18 min⚡ Interactive Tool📊 Diagram

Standard linear chains (like LCEL or standard LangChain) fail on multi-step reasoning because they cannot loop backward. LangGraph models agent workflows as stateful graphs (`StateGraph`). Nodes are python functions that mutate state, edges route control dynamically, and checkpointers persist state across execution steps — enabling true autonomous agents that can reflect and correct their own mistakes.

Key Concepts

TypedDict State

The entire graph shares a single `State` object (defined via Python TypedDict). Every node function receives this state, performs work, and returns a dict that updates the state.

Conditional Edges (Loops)

Unlike DAGs (Directed Acyclic Graphs), LangGraph supports cyclic routing. A conditional edge can evaluate the state (e.g., 'is code valid?') and route back to a previous node (e.g., 'Coder') to try again.

Checkpointers (Memory)

MemorySaver or Postgres checkpointers save the graph's state after every node. This allows 'Time Travel' (rewinding to a previous state) and Human-in-the-Loop (pausing for approval).

Multi-Agent Collaboration

Each node can be a completely independent LLM agent with its own system prompt and tools. They collaborate by mutating the shared State (e.g., a 'Researcher' agent passing data to a 'Writer' agent).

⚡ Interactive Architecture Simulator

LangGraph StateGraph Workflow

agent

Planner Agent

waiting…

→

tool

Tool: PostgreSQL

waiting…

→

evaluator

Quality Evaluator

waiting…

→

end

END / Checkpoint

waiting…

Step: 0 / 7Reflection loops: 0

LangGraph Architecture & Reflection Loop

LangChain vs LangGraph Architecture

Feature	LangChain (LCEL)	LangGraph
Execution Path	Linear / DAG (Directed Acyclic)	Cyclic (Loops allowed)
Memory / State	Pass-through dicts or external memory	Centralized `TypedDict` Graph State
Self-Correction	Difficult (requires external loops)	Native via Conditional Edges
Human-in-the-loop	Manual implementation required	Native via Checkpointer pause/resume
Best For	RAG pipelines, Chatbots	Multi-Agent workflows, Autonomous Agents

Production Code 1: StateGraph Core Architecture

langgraph_agent_core.pypython
import operator
from typing import TypedDict, Annotated, Sequence
from langgraph.graph import StateGraph, END
from langchain_core.messages import BaseMessage, HumanMessage, AIMessage

# ── 1. Define Agent State Schema ──────────────────────────────────────────
class AgentState(TypedDict):
    # 'operator.add' means new messages append to the list instead of overwriting
    messages: Annotated[Sequence[BaseMessage], operator.add]
    loop_count: int
    is_valid: bool

# ── 2. Define Node Functions ──────────────────────────────────────────────
def planner_node(state: AgentState):
    """Generates the plan or code."""
    # In production, call your LLM here
    return {
        "messages": [AIMessage(content="Generated SQL query plan.")], 
        "loop_count": state.get("loop_count", 0) + 1
    }

def evaluator_node(state: AgentState):
    """Evaluates the planner's output."""
    # Mocking evaluation logic: passes only on the 2nd loop
    is_valid = state["loop_count"] >= 2
    return {"is_valid": is_valid}

# ── 3. Define Conditional Routing Logic ────────────────────────────────────
def route_quality_check(state: AgentState) -> str:
    if state["is_valid"]:
        return END        # Output is good, finish graph
    if state["loop_count"] > 3:
        return END        # Break infinite loops
    return "planner"      # Feedback loop: send back to planner to try again

# ── 4. Construct StateGraph ────────────────────────────────────────────────
workflow = StateGraph(AgentState)

workflow.add_node("planner", planner_node)
workflow.add_node("evaluator", evaluator_node)

workflow.set_entry_point("planner")
workflow.add_edge("planner", "evaluator")

# Add Conditional Edge (Cyclic Reflection Loop)
workflow.add_conditional_edges("evaluator", route_quality_check)

app = workflow.compile()

Production Code 2: Human-in-the-Loop & Persistence

langgraph_hitl.pypython
from langgraph.checkpoint.memory import MemorySaver
# from langgraph.checkpoint.postgres import PostgresSaver # For Production DB
from langgraph.graph import StateGraph
# ... (Assuming 'workflow' from previous snippet is defined) ...

# ── 1. Compile with Checkpointer and Interrupts ────────────────────────────
# MemorySaver stores state in RAM. PostgresSaver stores it in a DB table.
checkpointer = MemorySaver()

# 'interrupt_before' pauses execution BEFORE the evaluator node runs,
# allowing a human to approve or modify the state.
app = workflow.compile(
    checkpointer=checkpointer,
    interrupt_before=["evaluator"]
)

# ── 2. First Run (Pauses at interrupt) ─────────────────────────────────────
thread_config = {"configurable": {"thread_id": "user-session-999"}}
initial_input = {"messages": [HumanMessage(content="Write a React component")]}

print("Running graph...")
for event in app.stream(initial_input, thread_config):
    print(event)

# Graph execution pauses here! 
print("Graph state:", app.get_state(thread_config).next) # Returns: ('evaluator',)

# ── 3. Human Approval (Resume Execution) ───────────────────────────────────
user_approval = input("Approve the planner's output? (y/n): ")
if user_approval.lower() == 'y':
    # Resume graph execution with NO new input (passes None)
    print("Resuming graph...")
    for event in app.stream(None, thread_config):
        print(event)

💡

Senior Architect Insight: To prevent autonomous agents from burning through your API budget in an infinite 'hallucination reflection' loop, ALWAYS add a `loop_count` to your `AgentState` and enforce a hard limit (e.g. max 3 retries) in your conditional edge router.

PreviousRedis Vector Semantic Cache

Next LessonAdvanced Agents & Long-Term Memory