David Archanjo

Posted on Jul 19

LangGraph Explained: A Practical Guide For Beginners

#ai #agents #langgraph #python

In this article, I will explain what LangGraph is, why it exists, and how it enables the development of stateful, multi-step AI applications powered by Large Language Models (LLMs). Instead of presenting isolated concepts, we will build our understanding step by step — from a simple single-node graph to a complete, production-ready AI workflow with state management, conditional routing, memory, human oversight, and multi-agent collaboration.

By the end of this article, you will understand how all these building blocks fit together and how LangGraph orchestrates them into a cohesive, controllable, and scalable AI system.

What We Will Build

By the end of this guide, we will have built a complete AI research and reporting workflow that:

Researches a topic using multiple specialized agents
Loops back for revisions until quality is approved
Runs independent tasks in parallel
Persists state across sessions
Pauses for human review before finalizing

About Me

I am David Archanjo, a Full-Stack Engineer with 10+ years of experience in the technology industry. I am passionate about software development and AI Engineering, and I enjoy sharing practical knowledge through technical articles that help developers build real-world applications.

Prerequisites
What Is LangGraph?
Why Learn and Use LangGraph?
LangChain vs LangGraph
The Core Concept
Hello, World! — Our First Graph
State Management
Nodes
Edges
Conditional Routing
Cycles and Loops
Parallel Execution
Persistence and Checkpointing
Human-in-the-Loop
Subgraphs
Building a Complete AI Workflow
Multi-Agent Systems
Production Features
Next Steps
Wrap Up

Prerequisites

To get the most out of this guide, you should have:

Python 3.10 or later and pip installed.
Basic knowledge of Python.
A text editor or IDE such as Visual Studio Code or PyCharm.
LangChain Basics: LangGraph is built on top of LangChain. You should understand what is an LLM, how to invoke one, what a Prompt Template is, and how chains work. If you are new to LangChain, you can check my LangChain Explained: A Practical Guide For Beginners.
Basic knowledge on flowcharts and how to visualize them using tools like Mermaid.

Python Core Dependencies

pip install langgraph langchain-openai

API Key Setup

# OpenAI
export OPENAI_API_KEY="sk-your-openai-api-key"

# Anthropic (optional)
export ANTHROPIC_API_KEY="sk-ant-your-anthropic-api-key"

For this article, you can use a locally hosted language model exposed through an OpenAI-compatible API using, for instance using llama.cpp. This setup allows every example to run entirely on our machine while keeping the code identical to what we would write when using the official OpenAI service.

I personally use llamafile, a standalone distribution of llama.cpp, to serve open models locally, like the Gemma 4:

./llamafile \
  -m gemma-4-E2B-it-Q4_K_M.gguf \
  --server \
  --host 0.0.0.0 \
  --port 8000 \
  --alias gpt-4o

To make this local setup work transparently, you must also configure the following environment variable:

export OPENAI_BASE_URL="http://localhost:8000/v1"

If you already have an OpenAI API key, simply remove the OPENAI_BASE_URL environment variable and replace the placeholder value of OPENAI_API_KEY with your own key. The examples throughout this article will continue to work without modification.

What Is LangGraph?

LangGraph is a framework for building stateful, multi-step AI applications where the flow of logic is modeled as a graph.

Think of it as a flowchart for our AI application.

Each box in the flowchart is a Node (a step that does something), and each arrow is an Edge (a connection that determines what happens next).

On its own, a standard LangChain chain:

Always executes steps in a fixed, linear order
Cannot loop back and retry a step
Cannot make dynamic decisions about which path to take next
Does not natively support pausing for human input in the middle of execution

LangGraph gives our AI application the ability to branch, loop, pause, and coordinate multiple agents through a controllable graph structure.

Why Learn and Use LangGraph?

To understand why we would consider LangGraph, we first need to understand what LangChain already does well, and where its linear model reaches its limits.

LangChain is excellent at building linear pipelines. For instance, a prompt flows into an LLM, the output flows into a parser, and the result is returned. With LangChain Expression Language (LCEL), a chain also supports parallel branches via RunnableParallel, memory via RunnableWithMessageHistory, and tool-calling via AgentExecutor. Frankly speaking, for a large number of real-world applications, this is everything we need.

However, as applications grow in complexity, certain patterns become difficult or impossible to express with a linear chain:

1. Cycles and Self-Correction

LangChain chains are directed and non-cyclic by design, i.e. data flows forward and never loops back. But many advanced AI workflows require iterationm, for instance:

Generate a draft → evaluate it → if it fails quality checks → revise and try again
Attempt a task → detect an error → retry with a corrected approach

In other words:

With LangChain: A chain does not have a native way to loop back to a previous step. We would have to wrap the entire chain in a manual while loop in our Python code, managing state and exit conditions by ourself.
With LangGraph: Cycles are a native feature of the graph. A node can route back to a previous node, creating a clean, controlled loop with a well-defined exit condition, having all the details expressed within the graph itself.

2. Complex Stateful Workflows

LangChain's memory tools work well for conversational history, i.e. tracking what was said in a chat. But for complex workflows where many different pieces of data need to be created, transformed, and passed across many steps, managing state through LangChain's memory primitives becomes awkward, to say the least.

With LangChain: For a chain, state beyond chat history must be managed externally, i.e. passed manually between steps or stored in application-level variables.
With LangGraph: State is a first-class citizen. We define a typed State object that is automatically shared across every node in the graph, always reflecting the latest values without any manual wiring.

3. Dynamic Conditional Routing

While LangChain's RunnableParallel handles static parallel branches, expressing dynamic routing, where the next step depends on the current result, requires conditional logic that does not fit naturally into a chain.

With LangChain: We must implement branching logic outside the chain, in our application code, using if/else statements that inspect the chain's output and decide what to do/call next.
With LangGraph: Conditional routing is a native graph concept. A routing function inspects the current state and returns the name of the next node to execute, keeping all the flow logic inside the graph where it is visible and testable.

4. Human-in-the-Loop Control

Real-world AI applications often need a human to review, approve, or correct AI decisions before the workflow continues.

With LangChain: There is no built-in mechanism to pause mid-execution and wait for human input. The implementation of such feature would require building custom infrastructure around the chain, e.g. queues, state serialization, resume logic, etc.
With LangGraph: Human-in-the-Loop is a built-in feature. The graph can be configured to pause at any node, wait for human input, and resume exactly where it left off, with the full state preserved automatically.

5. Multi-Agent Orchestration at Scale

LangChain's AgentExecutor is powerful for single-agent scenarios where one LLM decides which tools to be invoked. But coordinating multiple specialized agents, having each with their own role, tools, and sub-workflows that communicate with each other is beyond what a single AgentExecutor is designed for.

With LangChain: Each agent is a standalone executor. Coordinating multiple agents requires a hardcoded orchestration that manages inter-agent communication, shared context, and execution order.
With LangGraph: Each agent is a node in the graph. LangGraph handles routing between agents, maintains the shared state that all agents read from and write to, and provides a clear, visual structure for the entire collaboration ecosystem.

LangChain vs LangGraph

One of the most common questions when comes to LangGraph is: "I already use LangChain. Why would I need LangGraph?".

The answer depends entirely on the complexity of our workflow.

LangGraph and LangChain are not competitors; they complement each other. LangGraph builds on top of LangChain by adding an orchestration layer for stateful and complex workflows. We can still use LangChain components such as Prompt Templates, Output Parsers, Tools, and Retrievers inside our LangGraph nodes, while LangGraph controls how and when those components are executed.

Think in this way:

LangChain gives us the individual instruments.
LangGraph is the maestro that tells each instrument when to play.

Side-by-Side Comparison

Feature	LangChain	LangGraph
Execution Flow	Linear, sequential	Graph-based, flexible
Branching Logic	Via application code `if/else`	Native conditional edges
Loops / Cycles	Not supported natively	Native feature
State Management	Chat history via `RunnableWithMessageHistory`	Typed `State` object shared across all nodes
Parallel Execution	Via `RunnableParallel` and `.batch()`	Via `Send` API for dynamic fan-out
Memory / Persistence	`InMemoryChatMessageHistory`, `RedisChatMessageHistory`, etc.	Built-in checkpointers (Memory, SQLite, PostgreSQL)
Human-in-the-Loop	Not supported natively	Built-in interrupt and resume support
Tool Calling / Agents	`AgentExecutor` for single-agent scenarios	Multi-agent coordination as graph nodes
Best For	Linear pipelines, RAG, single agents	Complex stateful workflows, cycles, multi-agent systems

When to Use Which

1. Question: Is our workflow linear with no branching or loops?

Answer: LangChain is sufficient

2. Question: Does our app need to loop back and self-correct?

Answer: LangGraph (native cycles)

3. Question: Does our workflow have complex shared state beyond chat history?

Answer: LangGraph (typed State object)

4. Question: Do we need dynamic routing based on runtime results?

Answer: LangGraph (conditional edges)

5. Question: Do we need a human to approve AI decisions mid-workflow?

Answer: LangGraph (Human-in-the-Loop)

6. Question: Do we need multiple specialized agents collaborating?

Answer: LangGraph (Multi-Agent graph)

Key Takeaway: LangChain and LangGraph are designed to work together. We start with LangChain for simpler use cases and introduce LangGraph when our workflow requires cycles, complex state management, dynamic routing, human oversight, or multi-agent coordination.

The Core Concept

At its core, LangGraph is built around one simple idea: model AI application as a graph.

A graph is a structure made of two things:

Nodes: They are the steps — each node is a Python function that performs one unit of work, for instance: calling an LLM, running a tool, making a decision, or transforming data.
Edges: They are the connections between nodes — each edge defines what happens after a node finishes, i.e. which node to execute next. Together, nodes and edges define the flow of logic through our application.

The Three Core Building Blocks

Building Block	What It Does
State	A shared data object that carries information through the entire graph. Every node can read from and write to it.
Nodes	Python functions that perform work. They receive the current state as input and return updates to the state as output.
Edges	As connections between nodes, they can be fixed (always go to node B after node A) or conditional (go to B or C depending on the state).

Types of Edges

Edge	Type Description
Normal Edge	Always routes to the same next node.
Conditional Edge	Calls a routing function that inspects the state and returns the name of the next node to execute.
Entry Point	Defines which node the graph starts from.
`END`	A special destination that tells the graph the workflow is complete.

Types of Nodes

Nodes can perform virtually any operation required by our workflow. Some common types include:

LLM Nodes: Call a language model to generate, analyze, classify, or transform content.
Tool Nodes: Execute external tools, APIs, database queries, or other services.
Decision Nodes: Evaluate the current state and produce information used to determine what should happen next.
Pre-built Nodes: Ready-to-use nodes provided by LangGraph, such as ToolNode, which automatically executes tools requested by an LLM.

Regardless of what a node does internally, the basic pattern remains the same: it receives the current state, performs a unit of work, and returns the fields that should be updated.

How It All Flows

A basic LangGraph execution workflow looks like this:

Unlike a LangChain chain which is linear, a LangGraph graph can loop back:

This ability to loop, branch, and converge is what makes LangGraph fundamentally different and complementary from a standard LangChain chain.

Hello, World! — Our First Graph

It's time for a Hello, World!. Let's build the simplest possible LangGraph application: a graph with a single node which calls an LLM and returns a response.

from typing import TypedDict
from langchain.chat_models import init_chat_model
from langgraph.graph import StateGraph, END

# Step 1: Define the State
class State(TypedDict):
    question: str
    answer: str

# Step 2: Initialize the LLM
llm = init_chat_model(model="gpt-4o", model_provider="openai")

# Step 3: Define a Node
def answer_question(state: State) -> dict:
    response = llm.invoke(state["question"])
    return {"answer": response.content}

# Step 4: Build the Graph
graph_builder = StateGraph(State)

graph_builder.add_node("answer_question", answer_question)
graph_builder.add_edge(START, "answer_question")
graph_builder.add_edge("answer_question", END)

graph = graph_builder.compile()

# Step 5: Run the Graph
result = graph.invoke({"question": "What is LangGraph in one sentence?"})
print(result["answer"])

Output:

LangGraph is a library that allows developers to build complex, stateful, and cyclical LLM agent workflows by modeling them as directed graphs.

Breaking Down The Code

We defined a State using TypedDict with two fields: question and answer.
We initialized an LLM using init_chat_model.
We created a Node as a plain Python function that reads the question from the state, calls the LLM, and writes the generated answer back to the state.
We built a Graph using StateGraph, connected START to our node, and connected the node to END.
We compiled the graph into a runnable workflow and invoked it with an initial state.

This is the foundation, and everything else in LangGraph builds on top of this pattern.

The Execution Flow

State Management

State is the most important concept in LangGraph. It is the shared memory of our graph, i.e. the single source of truth that every node reads from and writes to as the workflow progresses.

What is a State?

State is a Python TypedDict that we define. Think of it as a baton in a relay race, in which each runner (node) picks it up, does their part, and passes it on.

from typing import TypedDict, List, Optional

class ResearchState(TypedDict):
    topic: str                 # Set at the start
    search_results: List[str]  # Populated by the researcher node
    draft: str                 # Written by the writer node
    feedback: Optional[str]    # Provided by the reviewer node
    final_report: str          # Finalized output

How Nodes Update State

A node receives the full current state and returns only the field(s) it changed. LangGraph merges those updates automatically:

def fetch_search_results(state: ResearchState) -> dict:
    topic = state["topic"]
    results = [
        f"Article 1 about {topic}",
        f"Article 2 about {topic}",
        f"Article 3 about {topic}"
    ]
    # Only return the fields being updated
    return {"search_results": results}

Using Reducers for Accumulating Data

By default, LangGraph replaces a state field with the new value. When we need to append instead, for instance to accumulate chat messages, we should use a reducer:

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

class ChatState(TypedDict):
    # New messages are appended, not replaced
    messages: Annotated[list, add_messages]

State Evolution Through the Graph

Nodes

As stated previously, a Node is a trivial Python function which receives the current state and returns a dictionary of updates. It can be considered as the unit of work in a LangGraph graph.

Anatomy of a Node

def my_node(state: State) -> dict:
    # 1. Read from state
    input_data = state["some_field"]

    # 2. Do some work
    result = do_something_with_input_data(input_data)

    # 3. Return only the fields being updated
    return {"output_field": result}

Categories of Nodes

Although LangGraph nodes are generally Python functions or runnables, we can categorize them based on the role they perform within a graph. Some common categories include:

LLM Node: calls a language model

An LLM Node calls a language model to perform tasks that require natural language understanding or generation. It can be used to answer questions, generate content, summarize information, classify text, or make decisions based on the current graph state.

from langchain.chat_models import init_chat_model
from langchain_core.prompts import ChatPromptTemplate

llm = init_chat_model(model="gpt-4o", model_provider="openai")

def generate_summary(state: State) -> dict:
    prompt = ChatPromptTemplate.from_template("Summarize the following text in 3 bullet points:\n\n{text}")
    chain = prompt | llm
    response = chain.invoke({"text": state["raw_text"]})
    return {"summary": response.content}

Tool Node: calls an external API or service

A Tool Node executes an external operation that extends the capabilities of the workflow beyond the language model itself. It can call an API, query a database, search the web, access a file system, or execute any other function exposed as a tool.

import requests

def fetch_weather(state: State) -> dict:
    city = state["city"]
    response = requests.get(f"https://api.weather.com/current?city={city}")
    return {"weather_data": response.json()}

Decision Node: performs logic without calling an LLM

A Decision Node performs deterministic logic without calling a language model. It typically inspects the current state and returns an update that can be used by a conditional edge to determine which execution path the graph should follow.

def check_quality(state: State) -> dict:
    word_count = len(state["draft"].split())
    is_sufficient = word_count >= 200
    return {"quality_passed": is_sufficient}

Pre-built ToolNode: LangGraph's built-in node for executing LangChain tools automatically

LangGraph provides the pre-built ToolNode for executing LangChain tools requested by a language model. Instead of manually creating a node that identifies and executes each tool call, ToolNode handles the tool execution automatically and returns the results to the graph.

from langgraph.prebuilt import ToolNode
from langchain_core.tools import tool

@tool
def search_web(query: str) -> str:
    """Search the web for information about a query."""
    return f"Search results for: {query}"

@tool
def calculate(expression: str) -> str:
    """Evaluate a mathematical expression."""
    return str(eval(expression))

# ToolNode automatically executes whichever tool the LLM requests
tools = [search_web, calculate]
tool_node = ToolNode(tools)

graph_builder.add_node("tools", tool_node)

Edges

Edges are the connections between Nodes. They define the flow of execution through the graph.

Normal Edges

A fixed, unconditional connection. After node A finishes, always go to node B:

from typing import TypedDict
from langchain.chat_models import init_chat_model
from langgraph.graph import StateGraph, END, START

class State(TypedDict):
    topic: str
    title: str
    body: str
    conclusion: str

llm = init_chat_model(model="gpt-4o", model_provider="openai")

def generate_title(state: State) -> dict:
    response = llm.invoke(f"Generate a blog title about: {state['topic']}")
    return {"title": response.content}

def write_body(state: State) -> dict:
    response = llm.invoke(f"Write a blog post titled: {state['title under 30 words.']}")
    return {"body": response.content}

def write_conclusion(state: State) -> dict:
    response = llm.invoke(f"Write a conclusion for:\n{state['body']} under 20 words.")
    return {"conclusion": response.content}

graph_builder = StateGraph(State)

graph_builder.add_node("generate_title", generate_title)
graph_builder.add_node("write_body", write_body)
graph_builder.add_node("write_conclusion", write_conclusion)

# Normal edges — always go from A to B
graph_builder.add_edge(START, "generate_title")
graph_builder.add_edge("generate_title", "write_body")
graph_builder.add_edge("write_body", "write_conclusion")
graph_builder.add_edge("write_conclusion", END)

graph = graph_builder.compile()
result = graph.invoke({"topic": "LangGraph"})
print(result["title"])
print('\n---\n')
print(result["body"])
print('\n---\n')
print(result["conclusion"])

Breaking Down The Code

State: The State schema defines the data shared across the workflow: the blog topic, generated title, body, and conclusion.
Nodes: Each node performs a specific task and updates a different field in the shared state.
Normal Edges: The nodes are connected using add_edge(), creating a fixed execution sequence where each node always runs after the previous one.
Execution Flow: The graph starts at START, executes generate_title → write_body → write_conclusion, and finally reaches END.

The Execution Flow

Conditional Routing

Normal edges are great for linear workflows, but real-world AI applications need to make decisions: "Is this output good enough? Which path should we take?"

That's where Conditional Edges come in.

What is a Conditional Edge?

A conditional edge connects one node to multiple possible next nodes via a routing function. Its a plain Python function that inspects the current state and returns the name of the next node:

def routing_function(state: State) -> str:
    return "node_b" if state["some_condition"] else "node_c"

graph_builder.add_conditional_edges(
    "node_a",          # From this node
    routing_function,  # Call this function
    {
        "node_b": "node_b",   # If it returns "node_b", go here
        "node_c": "node_c"    # If it returns "node_c", go here
    }
)

Example: Intent-Based Routing

In the following example, we build a graph that classifies a user's message and dynamically routes the execution to the appropriate response node based on the detected intent:

from typing import TypedDict
from langchain.chat_models import init_chat_model
from langgraph.graph import StateGraph, START, END

llm = init_chat_model(model="gpt-4o", model_provider="openai")

class State(TypedDict):
    user_input: str
    intent: str
    response: str

def classify_intent(state: State) -> dict:
    prompt = (
        "Classify the user's primary intent as exactly one word: "
        "question, complaint, or compliment. "
        "A message expressing dissatisfaction is a complaint, even if phrased as a question.\n\n"
        f"Message: {state['user_input']}"
    )
    intent = llm.invoke(prompt).content.strip().lower()
    return {"intent": intent}

def handle_question(state: State) -> dict:
    response = llm.invoke(f"Answer helpfully: {state['user_input']}")
    return {"response": response.content}

def handle_complaint(state: State) -> dict:
    response = llm.invoke(f"Respond empathetically: {state['user_input']}")
    return {"response": response.content}

def handle_compliment(state: State) -> dict:
    response = llm.invoke(f"Respond warmly: {state['user_input']}")
    return {"response": response.content}

def route_by_intent(state: State) -> str:
    return state["intent"]

builder = StateGraph(State)

builder.add_node("classify", classify_intent)
builder.add_node("question", handle_question)
builder.add_node("complaint", handle_complaint)
builder.add_node("compliment", handle_compliment)

builder.add_edge(START, "classify")

builder.add_conditional_edges(
    "classify",
    route_by_intent,
    {
        "question": "question",
        "complaint": "complaint",
        "compliment": "compliment",
    },
)

builder.add_edge("question", END)
builder.add_edge("complaint", END)
builder.add_edge("compliment", END)

graph = builder.compile()

result = graph.invoke({
    "user_input": "Why is my order still not delivered?",
    "intent": "",
    "response": "",
})

print(f"Intent: {result['intent']}")
print(f"Response: {result['response']}")

Breaking Down The Code

State: The State schema stores the user's input, the classified intent, and the generated response.
Classification: The classify_intent node uses the LLM to classify the user's input as a question, complaint, or compliment.
Routing Function: The route_by_intent function reads the classified intent from the state and returns the route that should be followed.
Conditional Edges: add_conditional_edges() maps each possible route to its corresponding node: question, complaint, or compliment.
Execution Flow: After the selected node generates the appropriate response, execution follows a normal edge to END.

The Execution Flow

Cycles and Loops

One of LangGraph's most powerful capabilities, and one that is not natively possible in a linear LangChain chain, is the ability to create cycles: graphs where execution can loop back to the previous node.

Why Cycles Matter

Cycles enable self-correcting AI workflows:

Generate an output → evaluate it → if not good enough → generate again
Try a tool → check the result → if wrong or failed → try a different approach
Draft a plan → execute it → check for errors → revise and retry

Example: Self-Correcting Code Generator

In the following example, we build a graph that generates Python code, evaluates its quality, and loops back to generate an improved version until the code passes validation or reaches the maximum number of attempts:

from typing import TypedDict, Optional
from langchain.chat_models import init_chat_model
from langgraph.graph import StateGraph, START, END

llm = init_chat_model(model="gpt-4o", model_provider="openai")

class State(TypedDict):
    task: str
    code: str
    error: Optional[str]
    attempts: int
    final_code: str

def generate_code(state: State) -> dict:
    if state.get("error"):
        prompt = (
            f"Fix this Python code:\n{state['code']}\n\n"
            f"Error: {state['error']}\n"
            "Return only valid Python code without Markdown."
        )
    else:
        prompt = (
            f"Write a Python function that {state['task']}. "
            "Return only valid Python code without Markdown."
        )

    code = llm.invoke(prompt).content.strip()

    return {
        "code": code,
        "attempts": state.get("attempts", 0) + 1,
        "error": None,
    }

def validate_code(state: State) -> dict:
    try:
        compile(state["code"], "<string>", "exec")
        return {"error": None, "final_code": state["code"]}
    except SyntaxError as e:
        return {"error": str(e), "final_code": ""}

def route(state: State) -> str:
    if not state.get("error") or state["attempts"] >= 3:
        return "done"
    return "retry"

builder = StateGraph(State)
builder.add_node("generate", generate_code)
builder.add_node("validate", validate_code)
builder.add_edge(START, "generate")
builder.add_edge("generate", "validate")
builder.add_conditional_edges(
    "validate",
    route,
    {"retry": "generate", "done": END},
)

graph = builder.compile()

result = graph.invoke({
    "task": "calculate the factorial of a number recursively",
    "code": "",
    "error": None,
    "attempts": 0,
    "final_code": "",
})

print("Final Code:")
print(result["final_code"])
print(f"Generated in {result['attempts']} attempt(s)")

The Execution Flow

We must always add an exit condition to our cycles. A cycle without a maximum attempt counter or a clear success condition creates an infinite loop.

Parallel Execution

LangGraph supports parallel execution, i.e. run multiple nodes simultaneously and merging their results. This is ideal when we have independent tasks that do not depend on each other. Parallel execution in LangGraph is powered by the Send API. It allows a node to dispatch multiple parallel tasks dynamically, each with its own state slice.

Example: Parallel Content Generation

In the following example, we build a graph that dynamically dispatches multiple content-generation tasks to run in parallel and combines their results into the shared graph state:

from typing import TypedDict, List, Annotated
from langchain.chat_models import init_chat_model
from langgraph.graph import StateGraph, START, END
from langgraph.types import Send
import operator

# Step 1: Initialize the LLM
llm = init_chat_model(model="gpt-4o", model_provider="openai")

# Step 2: Define the States
class State(TypedDict):
    topic: str
    results: Annotated[List[str], operator.add]

class BranchState(TypedDict):
    topic: str
    task: str

# Step 3: Define the Nodes
def write_section(state: BranchState) -> dict:
    response = llm.invoke(
        f"For a blog post about '{state['topic']}', write {state['task']}. "
        "Keep it concise."
    )
    return {"results": [f"{state['task']}:\n{response.content}"]}

def dispatch_tasks(state: State):
    tasks = ["a catchy title", "a 5-point outline", "an engaging introduction"]
    return [
        Send("write_section", {"topic": state["topic"], "task": task})
        for task in tasks
    ]

# Step 4: Define the Edges
builder = StateGraph(State)
builder.add_node("write_section", write_section)
builder.add_conditional_edges(
    START,
    dispatch_tasks,
    ["write_section"],
)
builder.add_edge("write_section", END)
graph = builder.compile()

# Step 5: Run the graph
result = graph.invoke({
    "topic": "LangGraph for AI Developers",
    "results": [],
})

# Print the result
for section in result["results"]:
    print(section)
    print("---")

The Execution Flow

When to Use Parallel Execution

Scenario	Benefit
Generating multiple independent sections	Significantly faster than sequential
Calling multiple APIs simultaneously	Reduces total wait time
Running the same task on multiple inputs	Processes all inputs at once
Evaluating output from multiple perspectives	Gets parallel viewpoints simultaneously

Persistence and Checkpointing

By default, a LangGraph graph is stateless between separate .invoke() calls. Each run starts fresh and real-world applications (e.g. chatbots, long-running workflows, multi-session assistants) need to remember previous interactions and resume from where they left off.

That's where Checkpointers come in.

What is a Checkpointer?

A Checkpointer automatically saves the full graph state after every node execution. This means:

If the graph is interrupted, it can be resumed from the last saved checkpoint.
Multiple .invoke() calls on the same thread_id share and continue the same state.
The complete execution history is available for inspection and debugging.

Example: In-Memory Checkpointing

In the following example, we build a conversational graph that uses an in-memory checkpointer to persist message history across multiple graph invocations sharing the same thread_id:

from typing import TypedDict, Annotated
from langchain.chat_models import init_chat_model
from langgraph.checkpoint.memory import MemorySaver
from langgraph.graph import StateGraph, START, END
from langgraph.graph.message import add_messages
from langchain_core.messages import HumanMessage

# Step 1: Initialize the LLM
llm = init_chat_model(model="gpt-4o", model_provider="openai")

# Step 2: Define the State
class ChatState(TypedDict):
    messages: Annotated[list, add_messages]

# Step 3: Define the Node
def chat(state: ChatState) -> dict:
    response = llm.invoke(state["messages"])
    return {"messages": [response]}

# Step 4: Define the Edge and Build the Graph
graph_builder = StateGraph(ChatState)
graph_builder.add_node("chat", chat)
graph_builder.add_edge(START, "chat") 
graph_builder.add_edge("chat", END)

# Step 5: Compile the Graph assigning a Checkpointer with Memory
memory = MemorySaver()
graph = graph_builder.compile(checkpointer=memory)

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

# Testing the graph
## Turn 1
result1 = graph.invoke(
    {"messages": [HumanMessage(content="My name is David.")]},
    config=config
)
print(result1["messages"][-1].content)
# Output: Hello David! It's nice to meet you. How can I help you today?

## Turn 2 — the graph remembers the full conversation
result2 = graph.invoke(
    {"messages": [HumanMessage(content="What is my name?")]},
    config=config
)
print(result2["messages"][-1].content)
# Output: Your name is David.

The Execution Flow

For applications where state must survive application restarts, we can use a database-backed checkpointer. LangGraph provides different checkpointer implementations depending on the application's requirements:

Checkpointer	Package	Use Case
`InMemorySaver`	`langgraph-checkpoint`	Development, debugging, and testing
`SqliteSaver`	`langgraph-checkpoint-sqlite`	Experimentation and lightweight local workflows
`AsyncSqliteSaver`	`langgraph-checkpoint-sqlite`	Async experimentation and lightweight local workflows
`PostgresSaver`	`langgraph-checkpoint-postgres`	Production workloads
`AsyncPostgresSaver`	`langgraph-checkpoint-postgres`	Async production workloads

For instance, we can use SqliteSaver to persist graph checkpoints in a physical SQLite database file:

import sqlite3
from langgraph.checkpoint.sqlite import SqliteSaver

connection = sqlite3.connect(
    "checkpoints.db",
    check_same_thread=False
)

checkpointer = SqliteSaver(connection)

graph = graph_builder.compile(
    checkpointer=checkpointer
)

Unlike InMemorySaver, checkpoints stored in SQLite are persisted to disk and can be retrieved after the application restarts. When invoking the graph, we must provide a thread_id, which the checkpointer uses to identify and retrieve the persisted state associated with a particular execution thread:

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

result = graph.invoke(
    {"messages": [...]},
    config=config
)

By reusing the same thread_id across graph invocations, LangGraph can retrieve the previously persisted state and continue the same execution thread.

Inspecting Saved State

Once a graph uses a checkpointer, we can inspect both the current state of a thread and its previous checkpoints. This is useful for debugging, monitoring long-running workflows, and understanding how the state evolved throughout the graph execution:

# View the current state of a thread
current_state = graph.get_state(config)
print(current_state.values)

# View the full history of a thread
for checkpoint in graph.get_state_history(config):
    print(f"Step: {checkpoint.metadata.get('step')}")
    print(f"State: {checkpoint.values}")
    print("---")

How Persistence Works

The thread_id acts as the identifier for a persistent execution thread. When we invoke the graph with a particular thread_id, the checkpointer retrieves the latest state associated with that thread, applies the new input, executes the graph, and saves the updated state. Reusing the same thread_id allows subsequent invocations to continue from the previously persisted state:

Human-in-the-Loop

One of LangGraph's most distinctive features is native support for Human-in-the-Loop (HITL), which is the ability to pause a running graph, present information to a human, collect their input or approval, and then resume exactly where the graph left off.

This is critical for production AI systems where autonomous decisions carry real consequences.

Why Human-in-the-Loop Matters

Safety: A human can catch and correct AI mistakes before they cause harm.
Trust: Users trust AI systems more when they know a human is in the control loop.
Compliance: Many industries require human approval before acting on AI recommendations.

How It Works: Interrupt

LangGraph implements HITL through interrupts — checkpoints in the graph where execution pauses and waits:

from langgraph.checkpoint.memory import MemorySaver

memory = MemorySaver()
graph = graph_builder.compile(
    checkpointer=memory,
    interrupt_before=["execute_action"]  # Pause before this node
)

Example: AI Action Approval Workflow

In the following example, we build a graph that pauses execution before performing a proposed action, allowing a human to review and approve or reject it before the workflow continues:

from typing import TypedDict, Optional
from langchain.chat_models import init_chat_model
from langgraph.checkpoint.memory import MemorySaver
from langgraph.graph import StateGraph, START, END

llm = init_chat_model(model="gpt-4o", model_provider="openai")

class State(TypedDict):
    user_request: str
    proposed_action: str
    human_approved: Optional[bool]
    result: str

def propose_action(state: State) -> dict:
    response = llm.invoke(
        f"A user wants to: '{state['user_request']}'\n\n"
        f"Propose a specific action to fulfill this request. "
        f"Be concise and specific."
    )
    return {"proposed_action": response.content}

def execute_action(state: State) -> dict:
    if not state.get("human_approved"):
        return {"result": "Action was rejected by the human reviewer."}

    response = llm.invoke(
        f"Execute the following action and describe what was done:\n"
        f"{state['proposed_action']}"
    )
    return {"result": response.content}

graph_builder = StateGraph(State)
graph_builder.add_node("propose_action", propose_action)
graph_builder.add_node("execute_action", execute_action)

graph_builder.add_edge(START, "propose_action")
graph_builder.add_edge("propose_action", "execute_action")
graph_builder.add_edge("execute_action", END)

memory = MemorySaver()
graph = graph_builder.compile(
    checkpointer=memory,
    interrupt_before=["execute_action"]
)

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

# Step 1: Run until the interrupt
print("=== Running until human review point ===")
graph.invoke(
    {
        "user_request": "Delete all files older than 30 days from the archive folder",
        "proposed_action": "",
        "human_approved": None,
        "result": ""
    },
    config=config
)

# Show the proposed action to the human
current_state = graph.get_state(config)
proposed = current_state.values["proposed_action"]
print(f"\nProposed Action:\n{proposed}")

# Step 2: Human reviews and provides input
print("\n=== Waiting for human approval ===")
human_decision = input("Do you approve this action? (yes/no): ").strip().lower()
approved = human_decision == "yes"

# Step 3: Update state with human decision and resume
graph.update_state(config, {"human_approved": approved})

print("\n=== Resuming graph after human input ===")
final_result = graph.invoke(None, config=config)
print(f"\nResult: {final_result['result']}")

The Execution Flow

Subgraphs

As our LangGraph applications grow in complexity, a single flat graph becomes difficult to manage. Subgraphs allow us to break a large, complex graph into smaller, reusable, self-contained graphs, being each responsible for one part of the workflow.

What is a Subgraph?

A subgraph is a fully compiled LangGraph graph that is used as a node inside a parent graph. From the parent's perspective, it looks and behaves like any other node, i.e. it receives state, does its work, and returns updates.

Why Use Subgraphs?

Benefit	Description
Modularity	Break complex workflows into focused, manageable units
Reusability	Use the same subgraph in multiple parent graphs
Testability	Test each subgraph independently before composing them
Readability	Parent graph stays clean and high-level

Example: Research and Writing Subgraphs

In the following example, we build a graph that consists of two subgraphs: one for research and another for writing. Each subgraph is responsible for a different part of the workflow, and they are composed into a single parent graph:

from typing import TypedDict
from langgraph.graph import StateGraph, START, END
from langchain.chat_models import init_chat_model

llm = init_chat_model(model="gpt-4o", model_provider="openai")

# ─────────────────────────────────────────
# Subgraph 1: Research Subgraph
# ─────────────────────────────────────────
class ResearchState(TypedDict):
    topic: str
    research_notes: str
    key_facts: str

def gather_notes(state: ResearchState) -> dict:
    response = llm.invoke(
        f"Research '{state['topic']}' and provide detailed notes."
    )
    return {"research_notes": response.content}

def extract_key_facts(state: ResearchState) -> dict:
    response = llm.invoke(
        f"Extract the 5 most important facts from these notes:\n{state['research_notes']}"
    )
    return {"key_facts": response.content}

research_builder = StateGraph(ResearchState)
research_builder.add_node("gather_notes", gather_notes)
research_builder.add_node("extract_key_facts", extract_key_facts)
research_builder.add_edge(START, "gather_notes")
research_builder.add_edge("gather_notes", "extract_key_facts")
research_builder.add_edge("extract_key_facts", END)

research_subgraph = research_builder.compile()

# ─────────────────────────────────────────
# Subgraph 2: Writing Subgraph
# ─────────────────────────────────────────
class WritingState(TypedDict):
    topic: str
    key_facts: str
    draft: str
    polished_report: str

def write_draft(state: WritingState) -> dict:
    response = llm.invoke(
        f"Write a report about '{state['topic']}' using these facts:\n{state['key_facts']}"
    )
    return {"draft": response.content}

def polish_report(state: WritingState) -> dict:
    response = llm.invoke(
        f"Polish and improve this report for clarity and flow:\n{state['draft']}"
    )
    return {"polished_report": response.content}

writing_builder = StateGraph(WritingState)
writing_builder.add_node("write_draft", write_draft)
writing_builder.add_node("polish_report", polish_report)
writing_builder.add_edge(START, "write_draft")
writing_builder.add_edge("write_draft", "polish_report")
writing_builder.add_edge("polish_report", END)

writing_subgraph = writing_builder.compile()

# ─────────────────────────────────────────
# Parent Graph: Orchestrates both subgraphs
# ─────────────────────────────────────────
class ParentState(TypedDict):
    topic: str
    research_notes: str
    key_facts: str
    draft: str
    polished_report: str

parent_builder = StateGraph(ParentState)

parent_builder.add_node("research", research_subgraph)
parent_builder.add_node("writing", writing_subgraph)

parent_builder.add_edge(START, "research")
parent_builder.add_edge("research", "writing")
parent_builder.add_edge("writing", END)

parent_graph = parent_builder.compile()

result = parent_graph.invoke({
    "topic": "The future of AI agents",
    "research_notes": "",
    "key_facts": "",
    "draft": "",
    "polished_report": ""
})

print("=== FINAL REPORT ===")
print(result["polished_report"])

The Execution Flow

Building a Complete AI Workflow

Now let's bring everything together. We will build a complete, end-to-end AI research assistant that combines state management, conditional routing, cycles, parallel execution, persistence, and human-in-the-loop into a single cohesive workflow.

The Workflow

Our complete AI workflow will:

Research the topic (gather notes in parallel from multiple angles)
Write a draft report from the research
Review the draft and loop back for revisions if needed
Pause for human approval before finalizing
Finalize the approved report

State Definition

class WorkflowState(TypedDict):
    topic: str
    research_angles: List[str]
    research_results: Annotated[List[str], operator.add]
    combined_research: str
    draft: str
    review_feedback: Optional[str]
    revision_count: int
    human_approved: Optional[bool]
    final_report: str

Node Definitions

# Node 1: Plan research angles
def plan_research(state: WorkflowState) -> dict:
    response = llm.invoke(
        f"List exactly 3 distinct research angles for the topic: '{state['topic']}'. "
        f"Return them as a simple numbered list."
    )
    angles = [
        line.strip()
        for line in response.content.split("\n")
        if line.strip() and line.strip()[0].isdigit()
    ]
    return {"research_angles": angles[:3]}

# Node 2: Research one angle (used in parallel)
def research_angle(state: dict) -> dict:
    response = llm.invoke(
        f"Research this specific angle about '{state['topic']}':\n"
        f"{state['angle']}\n\nProvide 3-4 key insights."
    )
    return {"research_results": [f"[{state['angle']}]\n{response.content}"]}

# Dispatcher: fan out to parallel research nodes
def dispatch_research(state: WorkflowState):
    return [
        Send("research_angle", {"topic": state["topic"], "angle": angle})
        for angle in state["research_angles"]
    ]

# Node 3: Combine parallel research results
def combine_research(state: WorkflowState) -> dict:
    combined = "\n\n".join(state["research_results"])
    response = llm.invoke(
        f"Synthesize these research findings into a cohesive summary:\n\n{combined}"
    )
    return {"combined_research": response.content}

# Node 4: Write the draft
def write_draft(state: WorkflowState) -> dict:
    feedback = state.get("review_feedback", "")
    revision = state.get("revision_count", 0)

    if feedback and revision > 0:
        prompt = (
            f"Revise this draft about '{state['topic']}' based on feedback.\n\n"
            f"Original Draft:\n{state['draft']}\n\n"
            f"Feedback:\n{feedback}\n\nWrite the improved version."
        )
    else:
        prompt = (
            f"Write a professional report about '{state['topic']}' "
            f"using this research:\n\n{state['combined_research']}"
        )

    response = llm.invoke(prompt)
    return {
        "draft": response.content,
        "revision_count": revision + 1,
        "review_feedback": None
    }

# Node 5: Review the draft
def review_draft(state: WorkflowState) -> dict:
    response = llm.invoke(
        f"Review this report and respond with either:\n"
        f"'APPROVED' — if complete and high quality\n"
        f"'REVISE: <feedback>' — if improvements are needed\n\n"
        f"Report:\n{state['draft']}"
    )
    feedback_text = response.content

    if "APPROVED" in feedback_text:
        return {"review_feedback": None}
    else:
        return {"review_feedback": feedback_text}

# Node 6: Finalize the report
def finalize_report(state: WorkflowState) -> dict:
    return {"final_report": state["draft"]}

# Routing: after AI review
def route_after_review(state: WorkflowState) -> str:
    if state.get("revision_count", 0) >= 3:
        return "finalize"
    if state.get("review_feedback"):
        return "revise"
    return "finalize"

Graph Definition

graph_builder = StateGraph(WorkflowState)

graph_builder.add_node("plan_research",    plan_research)
graph_builder.add_node("research_angle",   research_angle)
graph_builder.add_node("combine_research", combine_research)
graph_builder.add_node("write_draft",      write_draft)
graph_builder.add_node("review_draft",     review_draft)
graph_builder.add_node("finalize_report",  finalize_report)

graph_builder.add_edge(START, "plan_research")

graph_builder.add_conditional_edges(
    "plan_research",
    dispatch_research,
    ["research_angle"]
)

graph_builder.add_edge("research_angle", "combine_research")
graph_builder.add_edge("combine_research", "write_draft")
graph_builder.add_edge("write_draft", "review_draft")

graph_builder.add_conditional_edges(
    "review_draft",
    route_after_review,
    {
        "revise":   "write_draft",
        "finalize": "finalize_report"
    }
)

graph_builder.add_edge("finalize_report", END)

memory = MemorySaver()
graph = graph_builder.compile(
    checkpointer=memory,
    interrupt_before=["finalize_report"]
)

Running the Workflow

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

print("=== PHASE 1: AI Research and Writing ===")
graph.invoke(
    {
        "topic": "The impact of AI agents on software development",
        "research_angles": [],
        "research_results": [],
        "combined_research": "",
        "draft": "",
        "review_feedback": None,
        "revision_count": 0,
        "human_approved": None,
        "final_report": ""
    },
    config=config
)

current_state = graph.get_state(config)
print("\n=== PHASE 2: Human Review ===")
print("Draft Report:")
print(current_state.values["draft"])
print(f"\nRevisions made by AI: {current_state.values['revision_count']}")

decision = input("\nApprove this report? (yes/no): ").strip().lower()
graph.update_state(config, {"human_approved": decision == "yes"})

print("\n=== PHASE 3: Finalizing ===")
final = graph.invoke(None, config=config)
print("\n=== FINAL REPORT ===")
print(final["final_report"])

The Execution Flow

Multi-Agent Systems

LangGraph's most advanced capability is enabling multi-agent systems, i.e. architectures where multiple specialized AI agents collaborate, each focusing on what it does best, coordinated by LangGraph's graph structure.

What is a Multi-Agent System?

Instead of one LLM trying to do everything, a multi-agent system divides responsibilities:

Agent	Role
Orchestrator / Supervisor	Coordinates the other agents, decides who works next
Researcher	Gathers and summarizes information
Writer	Drafts and revises content
Reviewer	Critiques and provides quality feedback
Executor	Takes actions in the real world (APIs, databases)

Supervisor Pattern

The most common multi-agent pattern in LangGraph is the Supervisor Pattern, in which one supervisor agent routes work to specialized worker agents:

from typing import TypedDict
from langchain.chat_models import init_chat_model
from langchain_core.prompts import ChatPromptTemplate
from langgraph.checkpoint.memory import MemorySaver
from langgraph.graph import StateGraph, START, END
from langgraph.types import interrupt, Command

# Step 1: Initialize the LLM
llm = init_chat_model(model="gpt-4o", model_provider="openai")


# Step 2: Define the state
class AgentState(TypedDict):
    task: str
    research: str
    draft: str
    feedback: str
    final: str
    next_agent: str
    iteration: int


# Step 3: Define Supervisor Agent
def supervisor(state: AgentState) -> dict:
    iteration = state.get("iteration", 0)

    prompt = ChatPromptTemplate.from_messages([
        ("system",
         "You are a supervisor managing a team of AI agents.\n"
         "Available agents: researcher, writer, reviewer, human_review\n\n"
         "Rules:\n"
         "- Start with 'researcher' if there is no research yet\n"
         "- Use 'writer' after research is complete\n"
         "- Use 'reviewer' after a draft exists\n"
         "- Use 'human_review' after review or after 3 iterations\n\n"
         "Respond with only the agent name to call next."),
        ("human",
         "Task: {task}\n"
         "Research done: {has_research}\n"
         "Draft done: {has_draft}\n"
         "Last feedback: {feedback}\n"
         "Iterations: {iteration}\n\n"
         "Which agent should work next?")
    ])

    response = (prompt | llm).invoke({
        "task": state["task"],
        "has_research": "yes" if state.get("research") else "no",
        "has_draft": "yes" if state.get("draft") else "no",
        "feedback": state.get("feedback", "none"),
        "iteration": iteration
    })

    return {"next_agent": response.content.strip().lower()} # type: ignore


# Step 4: Define Worker Agents
def researcher(state: AgentState) -> dict:
    response = llm.invoke(
        f"Research this task thoroughly and provide key findings:\n{state['task']}"
    )
    return {
        "research": response.content,
        "iteration": state.get("iteration", 0) + 1
    }


def writer(state: AgentState) -> dict:
    response = llm.invoke(
        f"Write a professional report for this task:\n{state['task']}\n\n"
        f"Based on this research:\n{state['research']}"
    )
    return {
        "draft": response.content,
        "iteration": state.get("iteration", 0) + 1
    }


def reviewer(state: AgentState) -> dict:
    response = llm.invoke(
        f"Review this draft critically. Be specific about what is good "
        f"and what needs improvement:\n{state['draft']}"
    )
    return {
        "feedback": response.content,
        "iteration": state.get("iteration", 0) + 1
    }


# Human-in-the-Loop Agent: pause the graph and request human approval
def human_review(state: AgentState) -> dict:
    approved = interrupt({
        "draft": state["draft"],
        "feedback": state["feedback"],
        "question": "Do you approve this report?"
    })

    # Route to finalizer if approved, otherwise return to the writer
    return {
        "next_agent": "finalizer" if approved else "writer"
    }


def finalizer(state: AgentState) -> dict:
    response = llm.invoke(
        f"Polish this draft into a final, publication-ready report:\n{state['draft']}"
    )
    return {"final": response.content}


# Routing: supervisor decides next agent
def route_by_supervisor(state: AgentState) -> str:
    next_agent = state.get("next_agent", "researcher")
    valid_agents = ["researcher", "writer", "reviewer", "human_review"]
    return next_agent if next_agent in valid_agents else "human_review"


# Routing: human decision determines whether to finalize or revise
def route_after_human_review(state: AgentState) -> str:
    return state["next_agent"]


# Step 5: Build the Multi-Agent Graph
graph_builder = StateGraph(AgentState)

graph_builder.add_node("supervisor",   supervisor)
graph_builder.add_node("researcher",   researcher)
graph_builder.add_node("writer",       writer)
graph_builder.add_node("reviewer",     reviewer)
graph_builder.add_node("human_review", human_review)
graph_builder.add_node("finalizer",    finalizer)

graph_builder.add_edge(START, "supervisor") # or `graph_builder.set_entry_point("supervisor")`

graph_builder.add_conditional_edges(
    "supervisor",
    route_by_supervisor,
    {
        "researcher":   "researcher",
        "writer":       "writer",
        "reviewer":     "reviewer",
        "human_review": "human_review"
    }
)

# All workers report back to supervisor (except human_review and finalizer)
graph_builder.add_edge("researcher", "supervisor")
graph_builder.add_edge("writer",     "supervisor")
graph_builder.add_edge("reviewer",   "supervisor")

# Human approval routes to finalizer; rejection routes back to writer
graph_builder.add_conditional_edges(
    "human_review",
    route_after_human_review,
    {
        "writer":    "writer",
        "finalizer": "finalizer"
    }
)

graph_builder.add_edge("finalizer", END)


# Step 6: Compile the Graph with a checkpointer
# A checkpointer is required to pause and resume execution with interrupt()
memory = MemorySaver()
graph = graph_builder.compile(checkpointer=memory)

# The same thread_id must be used when starting and resuming the graph
config = {"configurable": {"thread_id": "report_001"}}


# Step 7: Run the graph until the Human-in-the-Loop interrupt
result = graph.invoke({
    "task": "Write a comprehensive overview of LangGraph for a technical blog",
    "research": "",
    "draft": "",
    "feedback": "",
    "final": "",
    "next_agent": "",
    "iteration": 0
}, config=config) # type: ignore


# Step 8: Display the information sent by the Human-in-the-Loop interrupt
interrupt_data = result["__interrupt__"][0].value

print("=== HUMAN REVIEW ===")
print("\nDraft:")
print(interrupt_data["draft"])

print("\nAI Reviewer Feedback:")
print(interrupt_data["feedback"])


# Step 9: Collect the human decision
decision = input("\nApprove report? (yes/no): ").strip().lower()
approved = decision == "yes"


# Step 10: Resume the graph with the human decision
# Command(resume=...) becomes the return value of interrupt()
result = graph.invoke(
    Command(resume=approved),
    config=config # type: ignore
)

# Step 11: Print the final result
print("\n=== FINAL OUTPUT ===")
print(result["final"])

The Execution Flow

Production Features

Taking a LangGraph application from prototype to production requires attention to reliability, observability, scalability, and error handling.

1. LangSmith Tracing

LangSmith automatically traces every step of our graph, making it easy to debug failures, measure latency, and monitor quality:

export LANGCHAIN_TRACING_V2=true
export LANGCHAIN_API_KEY="my-langsmith-api-key"
export LANGCHAIN_PROJECT="my-langgraph-app"

Once set, every graph invocation is automatically traced — no code changes required.

2. Error Handling and Retries

LLM calls and external services can fail temporarily due to network errors, rate limits, timeouts, or unavailable dependencies. Production nodes should handle these failures gracefully and retry transient errors instead of immediately failing the entire workflow. A common strategy is to retry the operation a limited number of times using exponential backoff:

import time

def resilient_node(state: State) -> dict:
    max_retries = 3
    for attempt in range(max_retries):
        try:
            response = llm.invoke(state["input"])
            return {"output": response.content, "error": ""}
        except Exception as e:
            if attempt < max_retries - 1:
                time.sleep(2 ** attempt)  # Exponential backoff
                continue
            return {"output": "", "error": str(e)}

3. Streaming Graph Output

For long-running workflows, waiting until the entire graph finishes before displaying any result can create a poor user experience. LangGraph supports streaming, allowing us to observe state updates and intermediate results as the graph executes:

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

for event in graph.stream(
    {"question": "Explain LangGraph in detail"},
    config=config,
    stream_mode="values"
):
    if "messages" in event:
        last_message = event["messages"][-1]
        if hasattr(last_message, "content"):
            print(last_message.content, end="", flush=True)

4. Async Execution

LangGraph also supports asynchronous execution through methods such as .ainvoke() and .astream(). Async execution is particularly useful for applications that perform I/O-bound operations, such as calling LLMs, APIs, databases, or other external services, because it allows the application to perform other work while waiting for those operations to complete:

import asyncio

async def async_node(state: State) -> dict:
    response = await llm.ainvoke(state["question"])
    return {"answer": response.content}

async def main():
    result = await graph.ainvoke(
        {"question": "What is LangGraph?"},
        config=config
    )
    print(result["answer"])

asyncio.run(main())

5. Runtime Configuration

Not every execution of a graph needs to use exactly the same configuration. LangGraph allows runtime values to be passed through the config object, making it possible to dynamically configure parameters such as the model, temperature, user identifiers, or other application-specific settings without modifying the graph state itself:

from langchain_core.runnables import RunnableConfig

def configurable_node(state: State, config: RunnableConfig) -> dict:
    model_name = config.get("configurable", {}).get("model", "gpt-4o")
    model_provider = config.get("configurable", {}).get("provider", "openai")
    temperature = config.get("configurable", {}).get("temperature", 0.7)

    llm = init_chat_model(model=model_name, model_provider=model_provider, temperature=temperature)
    response = llm.invoke(state["question"])
    return {"answer": response.content}

result = graph.invoke(
    {"question": "Explain AI agents"},
    config={
        "configurable": {
            "thread_id": "session_001",
            "model": "gpt-4o",
            "provider": "openai",
            "temperature": 0.2
        }
    }
)

6. Production Checkpointing with Persistent Storage

MemorySaver is useful during development, but its checkpoints exist only in application memory and are lost when the process restarts. Production applications should use a persistent checkpointer backed by durable storage, such as PostgreSQL, allowing graph state to survive application restarts and enabling workflows to be resumed across different application instances:

from langgraph.checkpoint.postgres import PostgresSaver

DB_URI = "postgresql://user:password@localhost:5432/langgraph_db"

with PostgresSaver.from_conn_string(DB_URI) as checkpointer:
    checkpointer.setup()
    graph = graph_builder.compile(checkpointer=checkpointer)

    result = graph.invoke(
        {"question": "What is LangGraph?"},
        config={"configurable": {"thread_id": "prod_session_001"}}
    )

Production Readiness Checklist

Feature	Tool	Purpose
Observability	LangSmith, Langfuse	Trace, debug, and monitor graph execution
Error Handling	`try/except` + retries	Resilience against transient failures
Streaming	`graph.stream()`	Responsive real-time user experience
Async	`graph.ainvoke()`	High-throughput concurrent processing
Persistence	Persistent storage checkpointer	Durable state across server restarts
Configuration	`RunnableConfig`	Runtime flexibility without code changes

Next Steps

Now that you have a solid understanding of LangGraph's core concepts and have built real-world applications with them, here are the natural next steps to expand your knowledge:

Explore the Ecosystem

LangGraph Studio: A visual IDE for building, running, and debugging LangGraph graphs interactively.
LangSmith: Set up evaluation pipelines, monitor production traces, and measure the quality of AI outputs.
Langfuse: An open-source alternative to LangSmith for LLM observability, tracing, evaluation, and debugging, with support for self-hosting.
LangGraph Cloud: Deploy your LangGraph applications to a fully managed, scalable cloud infrastructure.

Advanced Patterns to Learn

Reflection Agents: Agents that critique their own outputs and iteratively self-improve.
Plan-and-Execute: Agents that first create a full plan and then execute each step systematically.
Corrective RAG: A RAG pattern where the agent checks if retrieved documents are relevant and searches again if they are not.
Multi-Agent Debate: Multiple agents with opposing perspectives debate a topic to reach a higher-quality conclusion.

Wrap Up

Throughout this guide, we built our understanding of LangGraph from the ground up, from a single-node graph to a complete, production-ready multi-agent AI workflow.

Here is a summary of everything we covered:

Topic	Purpose	Key Takeaway
Prerequisites	Foundation knowledge	Know Python, LangChain basics, and LLM concepts
What Is LangGraph?	Core definition	A graph-based framework for stateful AI workflows
Why LangGraph Exists?	Motivation	Extends LangChain with cycles, complex state, and human oversight
LangChain vs LangGraph	Choosing the right tool	Complementary tools — LangGraph extends LangChain for complex workflows
The Core Concept	Mental model	Nodes do the work; edges control the flow
Our First Graph	Getting started	State + Node + Edge = a working LangGraph app
State Management	Shared memory	A TypedDict that evolves as it flows through the graph
Nodes	Units of work	Plain Python functions that read state and return updates
Edges	Flow control	Fixed or conditional connections between nodes
Conditional Routing	Dynamic decisions	Routing functions that inspect state and choose the next node
Cycles and Loops	Self-correction	Native looping — not possible in a linear LangChain chain
Parallel Execution	Speed and efficiency	Fan out independent tasks with the Send API
Persistence	Memory across sessions	Checkpointers save and restore full graph state automatically
Human-in-the-Loop	Safety and control	Pause, collect human input, and resume seamlessly
Subgraphs	Modularity	Compose large workflows from reusable sub-workflows
Complete Workflow	Everything together	Research → Write → Review → Human Approval → Finalize
Multi-Agent Systems	Collaboration	Specialized agents orchestrated by a supervisor
Production Features	Going to production	Tracing, streaming, async, error handling, and persistence

The Big Picture

LangGraph represents a natural evolution of LangChain, not a replacement, but an extension that unlocks capabilities that linear chains cannot express: cycles, complex shared state, dynamic conditional routing, human oversight, and multi-agent collaboration.

The best way to master LangGraph is to start simple and grow deliberately:

Begin with a single node and a clear state
Add edges to connect your steps
Introduce conditional routing when decisions are needed
Add cycles when self-correction is required
Layer in persistence, human oversight, and multiple agents as complexity demands

Remember, every production AI system we admire is built on exactly these principles.

All the Python examples presented in this guide are available in this accompanying GitHub repository, making it easy to run, modify, and experiment with each example on your own.

Happy AI coding 🚀

What We Will Build

About Me

Contents

Prerequisites

Python Core Dependencies

API Key Setup

What Is LangGraph?

Why Learn and Use LangGraph?

1. Cycles and Self-Correction

2. Complex Stateful Workflows

3. Dynamic Conditional Routing

4. Human-in-the-Loop Control

5. Multi-Agent Orchestration at Scale

LangChain vs LangGraph

Side-by-Side Comparison

When to Use Which

The Core Concept

The Three Core Building Blocks

Types of Edges

Types of Nodes

How It All Flows

Hello, World! — Our First Graph

Breaking Down The Code

The Execution Flow

State Management

What is a State?

How Nodes Update State

Using Reducers for Accumulating Data

State Evolution Through the Graph

Nodes

Anatomy of a Node

Categories of Nodes

LLM Node: calls a language model

Tool Node: calls an external API or service

Decision Node: performs logic without calling an LLM

Pre-built ToolNode: LangGraph's built-in node for executing LangChain tools automatically

Edges

Normal Edges

Breaking Down The Code

The Execution Flow

Conditional Routing

What is a Conditional Edge?

Example: Intent-Based Routing

Breaking Down The Code

The Execution Flow

Cycles and Loops

Why Cycles Matter

Example: Self-Correcting Code Generator

The Execution Flow

Parallel Execution

Example: Parallel Content Generation

The Execution Flow

When to Use Parallel Execution

Persistence and Checkpointing

What is a Checkpointer?

Example: In-Memory Checkpointing

The Execution Flow

Inspecting Saved State

How Persistence Works

Human-in-the-Loop

Why Human-in-the-Loop Matters

How It Works: Interrupt

Example: AI Action Approval Workflow

The Execution Flow

Subgraphs

What is a Subgraph?

Why Use Subgraphs?

Example: Research and Writing Subgraphs

The Execution Flow

Building a Complete AI Workflow

The Workflow

State Definition

Node Definitions

Graph Definition

Running the Workflow

The Execution Flow

Multi-Agent Systems

What is a Multi-Agent System?

Supervisor Pattern

The Execution Flow