Sequential Workflows in LangGraph

LangGraph transforms LLM orchestration from rigid, linear abstractions into dynamic, stateful computational graphs. This module focuses on building foundational Sequential Workflows (linear DAGs), demonstrating how state flows deterministically across custom processing nodes, integrating external frameworks like LangChain, and implementing complex multi-stage design patterns such as Prompt Chaining.

1. Prerequisites & Environment Setup

Building agentic workflows in LangGraph requires a isolated environment containing both graph orchestration libraries (langgraph) and model abstraction dependencies (langchain, langchain-openai).

1.1 Virtual Environment & Dependency Installation

In your project directory (langgraph-tutorials), instantiate an isolated Python virtual environment and install the required core packages:

BASH

# Create and activate virtual environment
python -m venv myenv

# Activation (Windows PowerShell / Command Prompt)
myenv\Scripts\activate

# Activation (Linux / macOS)
source myenv/bin/activate

# Install core dependencies
pip install langgraph langchain langchain-openai python-dotenv ipykernel

1.2 Development Environment Rationale

langgraph: Core state-graph framework providing StateGraph, node execution engines, and graph compilers.
langchain & langchain-openai: Handles LLM interactions, prompts, document loaders, and output parsing inside graph nodes.
python-dotenv: Manages environment variables (such as OPENAI_API_KEY) securely via .env files.
ipykernel & Jupyter Notebooks: Preferred for prototyping LangGraph applications because Jupyter natively renders inline graph visualizations (e.g., via draw_mermaid_png()), which terminal scripts cannot display directly.

2. Core Mental Model: The 4-Step LangGraph Lifecycle

Every LangGraph program—regardless of complexity—strictly adheres to a four-step construction lifecycle:

Structure & Concept Flow

Step 1: Define Application State Schema (TypedDict)

Step 2: Instantiate StateGraph with Schema

Step 3: Register Nodes (Python Functions) and Edges (Transitions)

Step 4: Compile & Execute Workflow (Invoke with Input State)

Define State: Construct a schema using Python's TypedDict to explicitly declare all shared data variables accessible across nodes.
Instantiate StateGraph: Pass the state schema into StateGraph to register the global state type for graph validation.
Register Nodes & Edges:
- Nodes: Pure Python functions that accept the global State object, process data, and return a dictionary containing partial state updates.
- Edges: Control-flow paths connecting nodes, including predefined entry/exit nodes (START and END).
Compile & Execute: Call .compile() to produce an executable RunnableGraph, then invoke it using .invoke(initial_state).

3. Workflow 1: Non-LLM Pure State Graph (BMI Calculator)

To master LangGraph mechanics without model-latency overhead or API dependencies, we begin with a pure algorithmic sequential graph: a BMI Calculator.

3.1 Mathematical Foundation

Body Mass Index ( $BMI$ ) is computed by dividing weight in kilograms by height in meters squared:

BMI = \frac{\text{weight}}{\text{height}^2}

3.2 Single-Node Implementation

Step 1: Define State Schema & Node Function

PYTHON

from typing import TypedDict
from langgraph.graph import StateGraph, START, END

# Define shared graph state schema
class BMIState(TypedDict):
    weight: float  # Weight in kilograms
    height: float  # Height in meters
    bmi: float     # Calculated BMI value

# Node function: Receives state, returns partial state update
def calculate_bmi(state: BMIState) -> BMIState:
    weight = state["weight"]
    height = state["height"]
    
    # Calculate BMI
    bmi_value = weight / (height ** 2)
    
    # Return dictionary representing state mutations
    return {"bmi": round(bmi_value, 2)}

Step 2: Assemble and Compile Graph

PYTHON

# Instantiate graph bound to BMIState schema
graph = StateGraph(BMIState)

# Add node: Node name -> Executable function mapping
graph.add_node("calculate_bmi", calculate_bmi)

# Add directed edges defining sequential execution flow
graph.add_edge(START, "calculate_bmi")
graph.add_edge("calculate_bmi", END)

# Compile graph into executable Runnable object
workflow = graph.compile()

Step 3: Graph Execution Flow

CODE

START
   |
   | sends initial state {"weight": 80.0, "height": 1.73}
   v
calculate_bmi Node
   |
   | computes BMI = 80 / (1.73)^2 = 26.73
   v
END (Returns updated state)

Step 4: Execution

PYTHON

# Initial input state
initial_state: BMIState = {
    "weight": 80.0,
    "height": 1.73
}

# Execute graph synchronously
final_state = workflow.invoke(initial_state)
print("Final State:", final_state)
# Output: {'weight': 80.0, 'height': 1.73, 'bmi': 26.73}

4. Workflow 1 Evolution: Multi-Node Sequential Processing

Real-world applications require modular, multi-step pipelines. We now extend our BMI calculator by attaching a second downstream processing node (label_bmi) to categorize health status based on computed numerical values.

4.1 Schema Mutation & Node Implementation

PYTHON

from typing import TypedDict
from langgraph.graph import StateGraph, START, END

# Extended State Schema containing categorical classification
class BMIState(TypedDict):
    weight: float
    height: float
    bmi: float
    category: str  # Added field for categorization

# Node 1: Numerical Calculation
def calculate_bmi(state: BMIState) -> BMIState:
    weight = state["weight"]
    height = state["height"]
    bmi_val = weight / (height ** 2)
    return {"bmi": round(bmi_val, 2)}

# Node 2: Categorical Classification
def label_bmi(state: BMIState) -> BMIState:
    bmi = state["bmi"]
    
    if bmi < 18.5:
        category = "Underweight"
    elif 18.5 <= bmi < 25.0:
        category = "Normal"
    elif 25.0 <= bmi < 30.0:
        category = "Overweight"
    else:
        category = "Obese"
        
    return {"category": category}

4.2 Multi-Node Graph Wiring

PYTHON

# Instantiate State Graph
graph = StateGraph(BMIState)

# Register both nodes
graph.add_node("calculate_bmi", calculate_bmi)
graph.add_node("label_bmi", label_bmi)

# Connect linear sequential edges
graph.add_edge(START, "calculate_bmi")
graph.add_edge("calculate_bmi", "label_bmi")
graph.add_edge("label_bmi", END)

# Compile Workflow
workflow = graph.compile()

4.3 Visualizing the Graph Structure

Jupyter environments render LangGraph structure directly using visualizers:

PYTHON

from IPython.display import Image, display

# Display graph visualization inline
display(Image(workflow.get_graph().draw_mermaid_png()))

Visual Pipeline Flow

Structure & Concept Flow

START

calculate_bmi

label_bmi

END

4.4 Execution

PYTHON

initial_input = {"weight": 80.0, "height": 1.73}
final_state = workflow.invoke(initial_input)

print(f"Calculated BMI: {final_state['bmi']}")
print(f"Classification: {final_state['category']}")
# Output:
# Calculated BMI: 26.73
# Classification: Overweight

5. Workflow 2: Integrating LangChain & LangGraph (LLM QA)

LangGraph does not replace LangChain; rather, LangGraph orchestrates control flow while LangChain provides component primitives (models, prompt templates, tool definitions, output parsers).

5.1 Simple LLM Question-Answering Graph

This workflow constructs a minimal single-node graph where a node uses ChatOpenAI from langchain-openai to resolve user queries.

PYTHON

import os
from typing import TypedDict
from dotenv import load_dotenv
from langchain_openai import ChatOpenAI
from langgraph.graph import StateGraph, START, END

# Load environment variables (.env containing OPENAI_API_KEY)
load_dotenv()

# Initialize Chat Model via LangChain
model = ChatOpenAI(model="gpt-4o-mini", temperature=0.7)

# Step 1: State Schema
class LLMState(TypedDict):
    question: str
    answer: str

# Step 2: Node Function wrapping LangChain execution
def llm_qa(state: LLMState) -> LLMState:
    question = state["question"]
    prompt = f"Answer the following question succinctly:\n{question}"
    
    # LangChain model invocation
    response = model.invoke(prompt)
    
    # State mutation returning extracted text content
    return {"answer": response.content}

# Step 3: Graph Construction
graph = StateGraph(LLMState)
graph.add_node("llm_qa", llm_qa)

graph.add_edge(START, "llm_qa")
graph.add_edge("llm_qa", END)

workflow = graph.compile()

# Step 4: Execution
initial_state = {"question": "How far is the Moon from the Earth?"}
output_state = workflow.invoke(initial_state)

print(f"Q: {output_state['question']}")
print(f"A: {output_state['answer']}")

6. Workflow 3: Multi-Stage Prompt Chaining (Blog Generation Pipeline)

Prompt Chaining is an enterprise agentic pattern where complex operations are decomposed into sequential LLM calls. Each node handles a dedicated sub-task, feeding state forward to downstream nodes.

6.1 Architectural Breakdown

We build an automated blogging pipeline that converts a high-level title into a full article using two distinct LLM steps:

create_outline: Reads title, outputs a structured blog outline.
create_blog: Reads both title and outline, outputs complete article content.

CODE

START
   |
   | State: {"title": "Rise of AI in India"}
   v
create_outline Node
   |
   | Model generates structured sections
   | Mutates State: adds {"outline": "1. Intro..."}
   v
create_blog Node
   |
   | Model expands title + outline into full text
   | Mutates State: adds {"content": "# Rise of AI..."}
   v
END

6.2 Implementation

PYTHON

import os
from typing import TypedDict
from dotenv import load_dotenv
from langchain_openai import ChatOpenAI
from langgraph.graph import StateGraph, START, END

load_dotenv()
model = ChatOpenAI(model="gpt-4o-mini", temperature=0.7)

# 1. State Definition
class BlogState(TypedDict):
    title: str
    outline: str
    content: str

# 2. Node Functions
def create_outline(state: BlogState) -> BlogState:
    title = state["title"]
    prompt = f"Generate a detailed outline for a technical blog post on the topic: '{title}'."
    
    response = model.invoke(prompt)
    return {"outline": response.content}

def create_blog(state: BlogState) -> BlogState:
    title = state["title"]
    outline = state["outline"]
    
    prompt = (
        f"Write a comprehensive technical blog post titled '{title}'.\n\n"
        f"Use the following structured outline as your guide:\n{outline}"
    )
    
    response = model.invoke(prompt)
    return {"content": response.content}

# 3. Graph Assembly
graph = StateGraph(BlogState)

# Add Nodes
graph.add_node("create_outline", create_outline)
graph.add_node("create_blog", create_blog)

# Add Sequential Edges
graph.add_edge(START, "create_outline")
graph.add_edge("create_outline", "create_blog")
graph.add_edge("create_blog", END)

# Compile Workflow
blog_workflow = graph.compile()

# 4. Invoke Execution Pipeline
initial_input = {"title": "Rise of AI in India"}
final_output = blog_workflow.invoke(initial_input)

# Inspect Execution Artifacts
print("=== GENERATED OUTLINE ===")
print(final_output["outline"])
print("\n=== FINAL BLOG CONTENT ===")
print(final_output["content"])

7. LangGraph State Persistence vs. Classic LangChain Chains

A core advantage of LangGraph over traditional sequential chains (like LangChain's legacy SimpleSequentialChain) is state persistence and full visibility across the workflow lifecycle.

Architectural Dimension	Legacy LangChain Sequential Chains	LangGraph State Graphs
Data Propagation	Intermediate output passes to next step; prior context is frequently lost or requires complex string mapping.	Global state object persists all parameters across the entire execution life cycle.
State Mutability	Rigid input-output schemas per chain layer.	Flexible partial dictionary updates ( $O(1)$ keyword key assignment).
Debugging & Observability	Hard to inspect intermediate outputs mid-sequence without verbose callbacks.	Every step updates the state schema; every variable (`title`, `outline`, `content`) remains fully accessible in the output.
Control Flow Capability	strictly linear ( $A \rightarrow B \rightarrow C$ ).	Seamless expansion from linear paths to conditional branching, loops, and parallel paths.

Key Takeaway: In standard chains, intermediate steps are dropped unless specifically captured. In LangGraph, state acts as an evolving ledger that accumulates data across all processing stages.

8. Practical Code Execution & Debugging Workflows

8.1 Partial State Updates

In LangGraph node functions, you do not need to re-return the full state dictionary. Returning only modified or newly created key-value pairs performs a clean partial update:

PYTHON

# CORRECT: Returning partial update
def my_node(state: MyState) -> MyState:
    # Only return fields you modified or created
    return {"bmi": 24.5}

# UNNECESSARY: Returning full state manually
def my_node_verbose(state: MyState) -> MyState:
    new_state = state.copy()
    new_state["bmi"] = 24.5
    return new_state  # LangGraph handles merging automatically!

8.2 Common Error States & Remedies

KeyError inside Node: Occurs when a downstream node attempts to read a key (state["outline"]) before an upstream node has written it. Ensure all fields are added to the TypedDict schema and written in sequence.
Missing API Key: If ChatOpenAI fails on invoke(), verify load_dotenv() is run before model initialization and that .env contains OPENAI_API_KEY=sk-....

9. Homework & Directed Practical Exercises

To reinforce the concept of sequential state propagation, extend the Prompt Chaining Blog Generator by adding an evaluation layer.

Exercise Requirements:

Extend Schema: Add score: int and feedback: str fields to BlogState.
Implement Node 3 (evaluate_blog): Create a node function that accepts title, outline, and content, and prompts an LLM to evaluate article quality. The node should output an integer rating ( $1-10$ ) and text feedback.
Rewire Pipeline: Connect create_blog $\rightarrow$ evaluate_blog $\rightarrow$ END.

10. Summary and Key Takeaways

Deterministic Control Flow: Sequential workflows in LangGraph execute nodes in a strict linear sequence ( $N_1 \rightarrow N_2 \rightarrow \dots \rightarrow N_k$ ) without dynamic conditional branching.
State as Central Source of Truth: All variables managed by the graph are declared in a TypedDict schema. Nodes consume input state and return partial updates.
Modular Node Abstraction: LangGraph nodes are pure Python functions that encapsulate custom business logic, mathematical computations, or LLM interactions.
LangChain Integration: LangChain acts as the component provider (models, prompts, parsers), while LangGraph serves as the runtime orchestrator.
Prompt Chaining: Complex tasks are split into modular LLM steps where intermediate states accumulate safely in the graph state.
Native Observability: Workflows built on StateGraph can be compiled and visually rendered using draw_mermaid_png(), making graph architecture easy to audit and debug.

Sequential Workflows in LangGraph | Agentic AI using LangGraph | Video 5 | CampusX

Sequential Workflows in LangGraph

1. Prerequisites & Environment Setup

1.1 Virtual Environment & Dependency Installation

1.2 Development Environment Rationale

2. Core Mental Model: The 4-Step LangGraph Lifecycle

3. Workflow 1: Non-LLM Pure State Graph (BMI Calculator)

3.1 Mathematical Foundation

3.2 Single-Node Implementation

Step 1: Define State Schema & Node Function

Step 2: Assemble and Compile Graph

Step 3: Graph Execution Flow

Step 4: Execution

4. Workflow 1 Evolution: Multi-Node Sequential Processing

4.1 Schema Mutation & Node Implementation

4.2 Multi-Node Graph Wiring

4.3 Visualizing the Graph Structure

Visual Pipeline Flow

4.4 Execution

5. Workflow 2: Integrating LangChain & LangGraph (LLM QA)

5.1 Simple LLM Question-Answering Graph

6. Workflow 3: Multi-Stage Prompt Chaining (Blog Generation Pipeline)

6.1 Architectural Breakdown

6.2 Implementation

7. LangGraph State Persistence vs. Classic LangChain Chains

8. Practical Code Execution & Debugging Workflows

8.1 Partial State Updates

8.2 Common Error States & Remedies

9. Homework & Directed Practical Exercises

Exercise Requirements:

Recommended Solution Skeleton:

10. Summary and Key Takeaways