beginner⏱35 minutesLesson 8 of 10

Sequential Chains

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

Sequential Chains

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"}

Practice Questions

Practice Question

How does state flow in a sequential LangGraph chain?

Practice Question

What is the default behavior when two nodes write to the same state key?

Practice Question

How do you make a list field append instead of replace?

Practice Question

What happens when two edges point to the same node (fan-in)?

Practice Question

In a sequential chain, when does the next node start?

Practice Question

What advantage does a LangGraph sequential chain have over a LangChain pipe chain?

Practice Question

If node A writes 'key1' and node B writes 'key2', can node C read both?

Practice Question

What operator does the 'add' reducer use for lists?

Practice Question

When building a fan-out pattern that fans back in, what node ordering is guaranteed?

Practice Question

What is the best use case for a sequential chain in LangGraph?

✅Success

Key Takeaways

Sequential chains are linear pipelines: each node runs after the previous one
State passes automatically between nodes through the shared state object
Default state update is replace; use Annotated[list, add] for appending
Fan-out lets you parallelize; fan-in synchronizes parallel branches
Sequential chains in LangGraph offer better debuggability than LangChain pipe chains
State reducers control how multiple writes to the same key are combined
Add conditional routing points to inject decision logic into sequential chains

Progress80%

Basic Tools

Conditional Branching

Course Content

1. What is LangGraph?2. LangChain Refresher 3. State, Graphs, and StateGraph 4. Nodes and Edges 5. Building a Simple Graph 6. Connecting LLMs to Nodes 7. Basic Tools 8. Sequential Chains 9. Conditional Branching 10. Beginner Project: Q&A Agent with Tool Use and Routing