Sequential Chains

Learn sequential execution patterns in LangGraph, passing state between nodes, and using state reducers for advanced state management.

While LangGraph supports complex graph topologies, many applications are built on simple sequential chains: a linear sequence of nodes where each step builds on the previous one.

Sequential Execution in LangGraph

A sequential chain is a linear pipeline: each node runs after the previous one completes, passing state along the way.

python

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

class PipelineState(TypedDict):
    input_text: str
    cleaned: str
    analyzed: str
    result: str

def clean(state: PipelineState) -> dict:
    return {"cleaned": state["input_text"].strip().lower()}

def analyze(state: PipelineState) -> dict:
    analysis = f"Analysis of '{state['cleaned']}': {len(state['cleaned'])} chars"
    return {"analyzed": analysis}

def format(state: PipelineState) -> dict:
    return {"result": f"=== RESULT ===\n{state['analyzed']}\n=== END ==="}

builder = StateGraph(PipelineState)
builder.add_node("clean", clean)
builder.add_node("analyze", analyze)
builder.add_node("format", format)
builder.add_edge(START, "clean")
builder.add_edge("clean", "analyze")
builder.add_edge("analyze", "format")
builder.add_edge("format", END)
app = builder.compile()

ℹ️Note

Each node only reads the fields it needs from state and writes the fields it produces. The pipeline topology ensures they run in the correct order.

Passing State Between Nodes

State flows automatically. When node A writes {"cleaned": value}, node B can read state["cleaned"]. This is the core mechanism for inter-node communication.

python

def node_a(state: State) -> dict:
    # Writes to state
    return {"intermediate": "Processed by A"}

def node_b(state: State) -> dict:
    # Reads from state (written by node_a)
    intermediate = state["intermediate"]  # "Processed by A"
    return {"result": f"B received: {intermediate}"}

State Flow Rules

Scenario	Behavior
A writes key X, B reads X	B sees A's value
B writes key X after A	A's value is overwritten
C writes key Y, D doesn't touch Y	D sees C's value for Y
No node writes key Y	Y keeps its initial value

Sequential Pipeline Example: Document Processor

python

from langchain_openai import ChatOpenAI
from langchain.prompts import ChatPromptTemplate
from langchain_core.output_parsers import StrOutputParser
from langgraph.graph import StateGraph, START, END
from typing_extensions import TypedDict
from typing import List

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

class DocState(TypedDict):
    raw_text: str
    cleaned_text: str
    summary: str
    bullet_points: List[str]
    final_doc: str

def clean_text(state: DocState) -> dict:
    cleaned = state["raw_text"].strip()
    # Remove extra whitespace
    import re
    cleaned = re.sub(r'\s+', ' ', cleaned)
    return {"cleaned_text": cleaned}

def summarize(state: DocState) -> dict:
    prompt = ChatPromptTemplate.from_messages([
        ("system", "Summarize the following text in 2-3 sentences."),
        ("human", "{text}")
    ])
    chain = prompt | llm | StrOutputParser()
    summary = chain.invoke({"text": state["cleaned_text"]})
    return {"summary": summary}

def extract_bullets(state: DocState) -> dict:
    prompt = ChatPromptTemplate.from_messages([
        ("system", "Extract 5 key points from the text as a numbered list."),
        ("human", "{text}")
    ])
    chain = prompt | llm | StrOutputParser()
    bullets_text = chain.invoke({"text": state["cleaned_text"]})
    bullets = [b.strip() for b in bullets_text.split("\n") if b.strip()]
    return {"bullet_points": bullets}

def format_document(state: DocState) -> dict:
    doc = f"""# Summary
{state['summary']}

# Key Points
{chr(10).join(f'- {b}' for b in state['bullet_points'])}
"""
    return {"final_doc": doc}

builder = StateGraph(DocState)
builder.add_node("clean", clean_text)
builder.add_node("summarize", summarize)
builder.add_node("bullets", extract_bullets)
builder.add_node("format", format_document)

builder.add_edge(START, "clean")
builder.add_edge("clean", "summarize")
builder.add_edge("summarize", "bullets")
builder.add_edge("bullets", "format")
builder.add_edge("format", END)

app = builder.compile()

✅Success

This is a real-world pipeline: clean → summarize → extract bullets → format. Each step is a separate, testable node.

State Reducers

By default, when a node writes to a state key, it replaces the value. State reducers change this behavior.

The Default Reducer: Replace

python

def node_a(state: State) -> dict:
    return {"messages": ["Hello"]}    # Replaces messages

def node_b(state: State) -> dict:
    return {"messages": ["World"]}    # Replaces messages (loses "Hello")

The Add Reducer: Append

Use Annotated with add to append to lists instead of replacing:

python

from typing import Annotated
from typing_extensions import TypedDict
from operator import add

class AccumState(TypedDict):
    messages: Annotated[list, add]  # Appends instead of replaces
    total: int

def step_1(state: AccumState) -> dict:
    return {"messages": ["Step 1 complete"], "total": 10}

def step_2(state: AccumState) -> dict:
    return {"messages": ["Step 2 complete"], "total": state["total"] + 20}

# Result: messages = ["Step 1 complete", "Step 2 complete"]
# Result: total = 30 (no reducer — last write wins)

⚠️Warning

The add reducer only works with lists. It uses Python's + operator, so both sides must be lists. If you need custom merge logic, define a custom reducer (covered in Intermediate course).

Parallel Sequential: Fan-Out

Sometimes you want sequential processing with parallel branches:

python

def validate(state: State) -> dict:
    return {"validated": True}

def search_web(state: State) -> dict:
    return {"web_results": "..."}

def search_db(state: State) -> dict:
    return {"db_results": "..."}

def merge(state: State) -> dict:
    combined = f"Web: {state['web_results']}\nDB: {state['db_results']}"
    return {"merged": combined}

builder.add_edge(START, "validate")
builder.add_edge("validate", "search_web")
builder.add_edge("validate", "search_db")  # Runs parallel to search_web
builder.add_edge("search_web", "merge")
builder.add_edge("search_db", "merge")     # Merge waits for both
builder.add_edge("merge", END)

✅Success

Fan-out with parallel branches, then fan-in to merge results. This pattern combines the power of sequential chains with parallel processing.

Sequential Chain with Conditionals

Inject decision points into your sequential chain:

python

def check_quality(state: State) -> str:
    if len(state.get("errors", [])) > 0:
        return "fix_errors"
    return "continue"

builder.add_edge(START, "process")
builder.add_edge("process", "validate")
builder.add_conditional_edges("validate", check_quality, {
    "fix_errors": "error_handler",
    "continue": "finalize"
})
builder.add_edge("error_handler", "process")  # Loop back for retry
builder.add_edge("finalize", END)

Comparing Sequential Chains: LangChain vs LangGraph

Aspect	LangChain Chain	LangGraph Sequential
Definition	`chain = A \| B \| C`	`add_edge(A, B); add_edge(B, C)`
State passing	Each output is next input	Shared state object
Intermediate access	Lost after chain runs	Available in state
Debugging	Hard (pipeline internals)	Easy (per-node streaming)
Adding steps	Requires chain rebuild	Just add edge
Error recovery	Chain fails entirely	Per-node handling

State Reducer: Merge for Dicts

python

from typing import Annotated
from typing_extensions import TypedDict

def merge_dicts(a: dict, b: dict) -> dict:
    """Custom reducer that deep-merges dictionaries."""
    result = a.copy()
    for k, v in b.items():
        if k in result and isinstance(result[k], dict) and isinstance(v, dict):
            result[k] = merge_dicts(result[k], v)
        else:
            result[k] = v
    return result

class NestedState(TypedDict):
    metadata: Annotated[dict, merge_dicts]
    logs: Annotated[list, add]

def node_a(state: NestedState) -> dict:
    return {"metadata": {"step": "a", "timestamp": "2024-01-01"}}

def node_b(state: NestedState) -> dict:
    return {"metadata": {"status": "done"}, "logs": ["Step B executed"]}
# Final metadata: {"step": "a", "timestamp": "2024-01-01", "status": "done"}