LangGraph Tutorial — Stateful Multi-Agent Orchestration & Cyclic Workflows

01 Why LangGraph? Beyond Linear Chains & DAGs

Architecture Shift

Most legacy workflow engines (Airflow, Prefect, early LangChain chains) enforce Directed Acyclic Graphs (DAGs). But real-world intelligence is inherently non-linear and iterative:

↻

Cyclic Execution

Agents must loop between Reasoning, Tool Execution, and Self-Correction until an exit condition is met, without hacky recursion or while-loop duct tape.

💾

Fine-Grained State Reducers

State is not an opaque global blob. Fields use explicit reducers (such as operator.add or add_messages) to safely append, overwrite, or mutate sub-states across multiple nodes.

⏸

Durable Checkpointing & HITL

Every single step is persisted to durable storage. You can interrupt execution, request human authorization before sending a transaction, and rewind or edit past state.

02 The Core Primitives: State, Nodes, and Edges

Building Blocks

1. State & Nodes

A Node is just a standard Python function (sync or async) that takes the current state as input and returns updated keys. Nodes do not know about other nodes; they only read from and write to the graph state.

from typing import Annotated, TypedDict
from langgraph.graph.message import add_messages

# Define the shared state schema
class AgentState(TypedDict):
    # 'add_messages' ensures new messages append rather than overwrite
    messages: Annotated[list, add_messages]
    retry_count: int

def call_model(state: AgentState):
    response = model.invoke(state["messages"])
    return {"messages": [response]}

2. Edges & Conditional Routing

Edges dictate control flow. Normal edges connect node A to node B unconditionally. Conditional Edges evaluate a routing function to decide the next node dynamically.

from langgraph.graph import StateGraph, START, END
from langgraph.prebuilt import tools_condition

workflow = StateGraph(AgentState)

workflow.add_node("agent", call_model)
workflow.add_node("tools", tool_node)

workflow.add_edge(START, "agent")
# If model requested a tool call -> route to 'tools', else -> END
workflow.add_conditional_edges("agent", tools_condition)
workflow.add_edge("tools", "agent") # Cyclic loop!

app = workflow.compile()

03 Persistence & Thread Checkpointing

Reliability

By compiling your graph with a Checkpointer, every state transition is recorded as a snapshot identified by a thread_id. This provides fault-tolerance against network dropouts, conversational memory across sessions, and full time-travel debugging.

⚡ Local Testing: MemorySaver

In-memory checkpointer for local unit testing, rapid prototyping, and CLI experimentation.

from langgraph.checkpoint.memory import MemorySaver

memory = MemorySaver()
app = workflow.compile(checkpointer=memory)

config = {"configurable": {"thread_id": "user_42_conversation_1"}}
response = app.invoke({"messages": [("user", "Hello")]}, config=config)

🐘 Production: PostgresSaver

Persist snapshots directly into PostgreSQL tables for horizontally scalable agent microservices.

from langgraph.checkpoint.postgres import PostgresSaver

DB_URI = "postgresql://user:pass@localhost:5432/agents"
with PostgresSaver.from_conn_string(DB_URI) as checkpointer:
    checkpointer.setup()  # Auto-creates checkpoint tables
    app = workflow.compile(checkpointer=checkpointer)
    # Thread state survives server restarts and crashes

04 Human-in-the-Loop (HITL) & Approval Gates

Safety & Control

When agents perform destructive or financial actions (e.g. dropping a table, executing a trade, sending a contract), autonomous execution must pause for human confirmation. LangGraph provides first-class support via interrupt_before and update_state:

# 1. Compile with an interrupt before critical tool execution
app = workflow.compile(
    checkpointer=memory,
    interrupt_before=["tools"]  # Graph automatically halts when it reaches 'tools'
)

config = {"configurable": {"thread_id": "financial_tx_101"}}

# 2. Run graph - it pauses right before tools run
for event in app.stream({"messages": [("user", "Transfer $5,000 to Vendor B")]}, config):
    print(event)

# 3. Inspect pending state snapshot
state = app.get_state(config)
print("Pending tool call:", state.next) # ('tools',)

# 4. Human approver reviews and resumes (or edits state)
# Passing None continues execution with existing state
app.invoke(None, config=config)

05 Multi-Agent Patterns in LangGraph

Coordination

👑

Supervisor Architecture

Hierarchical Router

A top-level LLM router acts as an orchestrator, delegating sub-tasks to specialized sub-agents (e.g. Researcher, Coder, Reviewer) and synthesizing their results before delivering final output to the user.

Deterministic routing control
Sub-agents keep isolated, domain-specific tools
Easy to add quality-assurance review loops

🤝

Peer Handoff (Swarm)

Decentralized Transfer

Agents hand off control directly to one another via Command(goto="specialist_agent") without returning to a central supervisor. Popularized by OpenAI Swarm patterns and implemented natively in LangGraph.

Lower latency than multi-hop supervisor passes
Ideal for customer support triage trees
Requires strict handoff contracts to avoid infinite loops

Production Self-Corrective Search Agent

A complete, runnable LangGraph 0.2+ implementation with tools, conditional edges, and persistence.

Python 3.10+

import os
from typing import Annotated, Literal
from typing_extensions import TypedDict

from langchain_core.tools import tool
from langchain_openai import ChatOpenAI
from langgraph.graph import StateGraph, START, END
from langgraph.graph.message import add_messages
from langgraph.prebuilt import ToolNode, tools_condition
from langgraph.checkpoint.memory import MemorySaver

# 1. Define custom tools
@tool
def search_knowledge_base(query: str) -> str:
    """Searches corporate knowledge base for architecture and policy documents."""
    return f"Knowledge Base result for '{query}': AI agents must use LangGraph for cyclic state machines."

tools = [search_knowledge_base]
tool_node = ToolNode(tools)

# 2. Bind tools to LLM
model = ChatOpenAI(model="gpt-6-sol", temperature=0).bind_tools(tools)

# 3. Define Graph State
class AgentState(TypedDict):
    messages: Annotated[list, add_messages]

# 4. Define Agent Node
def call_agent(state: AgentState):
    response = model.invoke(state["messages"])
    return {"messages": [response]}

# 5. Build Graph
workflow = StateGraph(AgentState)

workflow.add_node("agent", call_agent)
workflow.add_node("tools", tool_node)

workflow.add_edge(START, "agent")
workflow.add_conditional_edges("agent", tools_condition)
workflow.add_edge("tools", "agent")  # Return to agent after tool execution

# 6. Compile with Checkpointer
memory = MemorySaver()
app = workflow.compile(checkpointer=memory)

# 7. Execute with Session Continuity
thread_config = {"configurable": {"thread_id": "session_alpha"}}
inputs = {"messages": [("user", "What is the company policy on AI agent state machines?")]}

for event in app.stream(inputs, config=thread_config, stream_mode="values"):
    latest_msg = event["messages"][-1]
    print(f"[{latest_msg.type}]: {latest_msg.content}")

07 Framework Comparison

Feature	LangGraph	CrewAI	OpenAI Agents SDK
Execution Model	Cyclic StateGraph (Nodes & Edges)	Role-based Crew (Sequential / Hierarchical)	Event loop with Agent Handoffs
State Management	First-Class Reducers (add_messages)	Shared Task Output / Memory Storage	Context object passed through runners
Persistence & Replay	Durable Checkpointers (Postgres/Redis)	SQLite / Chroma Long-term Memory	External Session Store
Human-in-the-Loop	Native interrupt_before & state edits	Human review flags per task	Guardrail hooks / Custom middleware
Best Use Case	Complex, deterministic, cyclic enterprise workflows	Collaborative research teams & role-play agents	Fast-turnaround triage, customer support, handoffs

08 Frequently Asked Questions

Is LangGraph tied only to LangChain models? →

No. LangGraph is a standalone orchestration graph. While it integrates seamlessly with langchain-core, nodes are just Python functions. You can call raw OpenAI, Anthropic, Google Vertex AI, or local Ollama SDKs directly inside any node.

How does LangGraph prevent infinite agent loops? →

LangGraph includes a configurable recursion_limit parameter in config={"recursion_limit": 25} (defaults to 25 steps). If a graph exceeds this threshold before reaching an END node, a GraphRecursionError is raised.

How do I stream token-by-token in LangGraph? →

Use app.astream_events() with version "v2". It streams events for LLM tokens (on_chat_model_stream), tool starts, tool ends, and node state transitions in real time.

Stateful Agent Orchestration with LangGraph