Skip to content

Sequential Workflows

1. Why this matters

90% of LangGraph apps start sequential. Even when you eventually need branches or loops, getting the linear backbone working first is the right move.

You'd reach for sequential over a plain LCEL chain when: - You want named, observable intermediate state (each node's output is in the snapshot). - You want persistence (checkpointing) across steps. - You may later add branches/loops — building the linear version first makes that natural.

2. Mental model

A pipeline where state grows step-by-step:

flowchart LR
    S0[State 0<br/> topic, ...] --> N1[Node 1<br/>+ outline]
    N1 --> S1[State 1<br/> topic, outline, ...]
    S1 --> N2[Node 2<br/>+ blog]
    N2 --> S2[State 2<br/> topic, outline, blog, ...]
    S2 --> N3[Node 3<br/>+ summary]
    N3 --> S3[State 3<br/>Final]

Each node: - Reads what it needs from state. - Writes new keys via a partial dict return. - Doesn't touch anything else.

3. Architecture / Flow

flowchart LR
    ST((START)) --> A[outline]
    A --> B[write]
    B --> C[edit]
    C --> D[publish]
    D --> EN((END))

4. Core concepts

  • Linear edge chain — every add_edge(X, Y) makes Y run after X completes.
  • Single-writer fields — sequential nodes usually each own a distinct state field, so no reducer is needed.
  • Sharing data across non-adjacent nodes — just keep the field in state; intermediate nodes ignore it.
  • set_finish_point("last") — shortcut for add_edge("last", END).

5. Code — minimal working example

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

class S(TypedDict):
    name: str
    greeting: str
    farewell: str

def greet(state: S):     return {"greeting": f"Hi, {state['name']}!"}
def farewell(state: S):  return {"farewell": f"Bye, {state['name']}."}

b = StateGraph(S)
b.add_node("greet", greet)
b.add_node("bye", farewell)
b.add_edge(START, "greet")
b.add_edge("greet", "bye")
b.add_edge("bye", END)

print(b.compile().invoke({"name": "Alice", "greeting": "", "farewell": ""}))
# {'name': 'Alice', 'greeting': 'Hi, Alice!', 'farewell': 'Bye, Alice.'}

6. Code — real-world pattern

Cricket player performance summarizer — the CampusX-style multi-node pipeline:

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

class PlayerState(TypedDict):
    runs: int
    balls: int
    fours: int
    sixes: int
    sr: float                  # strike rate
    bpb: float                 # balls per boundary
    boundary_pct: float
    summary: str

def calc_sr(state: PlayerState):
    return {"sr": (state["runs"] / state["balls"]) * 100}

def calc_bpb(state: PlayerState):
    boundaries = state["fours"] + state["sixes"]
    return {"bpb": state["balls"] / boundaries if boundaries else 0.0}

def calc_boundary_pct(state: PlayerState):
    boundary_runs = state["fours"] * 4 + state["sixes"] * 6
    return {"boundary_pct": (boundary_runs / state["runs"]) * 100}

def summarize(state: PlayerState):
    return {"summary": (
        f"Strike Rate: {state['sr']:.1f}\n"
        f"Balls per Boundary: {state['bpb']:.1f}\n"
        f"Boundary %: {state['boundary_pct']:.1f}"
    )}

b = StateGraph(PlayerState)
b.add_node("sr", calc_sr)
b.add_node("bpb", calc_bpb)
b.add_node("bpct", calc_boundary_pct)
b.add_node("sum", summarize)

b.add_edge(START, "sr")
b.add_edge("sr", "bpb")
b.add_edge("bpb", "bpct")
b.add_edge("bpct", "sum")
b.add_edge("sum", END)

graph = b.compile()

initial = {"runs": 100, "balls": 50, "fours": 6, "sixes": 4,
           "sr": 0, "bpb": 0, "boundary_pct": 0, "summary": ""}
print(graph.invoke(initial)["summary"])

LLM-flavored sequential (blog-writing pipeline):

from langchain_openai import ChatOpenAI
from langchain_core.messages import HumanMessage

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

class BlogState(TypedDict):
    topic: str
    outline: str
    draft: str
    final: str

def outline(s: BlogState):
    r = llm.invoke([HumanMessage(f"Make a 5-point outline for a blog on: {s['topic']}")])
    return {"outline": r.content}

def draft(s: BlogState):
    r = llm.invoke([HumanMessage(f"Write the blog using:\n{s['outline']}")])
    return {"draft": r.content}

def polish(s: BlogState):
    r = llm.invoke([HumanMessage(f"Edit for clarity & flow:\n{s['draft']}")])
    return {"final": r.content}

b = StateGraph(BlogState)
for name, fn in [("outline", outline), ("draft", draft), ("polish", polish)]:
    b.add_node(name, fn)
b.add_edge(START, "outline")
b.add_edge("outline", "draft")
b.add_edge("draft", "polish")
b.add_edge("polish", END)

print(b.compile().invoke({"topic": "RAG vs fine-tuning", "outline": "", "draft": "", "final": ""})["final"])

7. Common pitfalls

  • ❗ Returning the full state from each node. Wastes work and bypasses reducers. Return only what changed.
  • ❗ Stuffing all logic into one giant node. That defeats the point. Each node should do one observable thing — easier to debug and trace.
  • ❗ Hardcoding the LLM inside every node. Build the model once at module scope; reference it from nodes. Saves connection overhead.
  • ❗ No defaults in the initial state. TypedDict doesn't require fields but the nodes might. Either give defaults (Pydantic) or include all keys in initial state.
  • ❗ Sequential when parallel would be faster. If two nodes are independent (don't depend on each other's outputs), run them in parallel — see Chapter 7.

8. When to use vs not use

Use sequential when Don't when
Each step needs the previous step's output Nodes are independent → use parallel
You want observable, named intermediate state A single LCEL pipe is fine and simpler
The order is fixed (no branching) Dynamic routing needed → conditional
You may add branches later — start linear The whole flow is just one LLM call

9. Cheatsheet

# The whole pattern in 6 lines
b = StateGraph(StateSchema)
for name, fn in [("a", a), ("b", b_fn), ("c", c)]:
    b.add_node(name, fn)
b.set_entry_point("a")
b.add_edge("a", "b"); b.add_edge("b", "c")
b.set_finish_point("c")
graph = b.compile()

10. Q&A — recall test

  • Q: When is sequential the right shape? A: When every step depends on the previous one, the order is fixed, and there's no branching or looping.

  • Q: Should a sequential node return the full state or just changes? A: Just changes — return a dict with only the keys that node modified.

  • Q: Sequential graph vs LCEL a | b | c? A: LCEL is simpler and pipe-natural; LangGraph sequential gives you named nodes, observable intermediate state, and a path to add branches/loops later. Use LCEL unless you need those.

  • Q: How do you make two adjacent sequential nodes share data they both need? A: Put the field in the state schema; both nodes read it. State persists across all nodes within a run.

Practice

What does this print?

Expected: [1, 2, 3]

# Sequential nodes: each receives the output of the previous
def step1(s): return {"val": [1]}
def step2(s): return {"val": s["val"] + [2]}
def step3(s): return {"val": s["val"] + [3]}
s = {"val": []}
s.update(step1(s)); s.update(step2(s)); s.update(step3(s))
print(s["val"])

Connect node1 → node2 with an explicit edge (not implicit ordering)

Expected: True

edge_exists = False        # bug: must explicitly add_edge(node1, node2)
print(not edge_exists)

Quiz — Quick check

What you remember

Q1. In a sequential workflow A → B → C, what does each node see?

  • The cumulative state including changes from all prior nodes
  • Only the original input
  • Only the previous node's return value
  • An empty state

Why: State is global within a run. Node B sees A's changes; node C sees both A and B's changes. That's how data flows.

Q2. How do you connect two nodes sequentially?

  • graph.add_edge("nodeA", "nodeB")
  • Just call them in order
  • Add a next field
  • Python imports

Why: Edges are explicit. Without an edge, LangGraph doesn't know to run B after A. add_edge is the simplest form; add_conditional_edges is the branching form.

Q3. When does a sequential workflow stop?

  • When the last node's edge points to END
  • After a fixed number of steps
  • When state is empty
  • When a node returns None

Why: The graph executes until it reaches END. Forgetting the END edge causes infinite execution or "no edges from current node" errors.

Common doubts

Should every workflow start sequential and add branches later?

Yes — start simple. Build the happy path as a linear sequence, validate it works, then add conditional edges, retries, and parallel branches as needed. LangGraph makes incremental complexity easy.

What's the difference between LCEL and a sequential LangGraph?

LCEL chain1 | chain2 | chain3 is sequential too. The difference: LangGraph adds state persistence (checkpointing), built-in retries, and the ability to LATER add cycles or branches. If you're sure your workflow stays linear forever, LCEL is simpler.

Can I add async nodes in a sequential graph?

Yes — define your node as async def my_node(state): and invoke with await graph.ainvoke(...). Async nodes integrate seamlessly with sync nodes in the same graph.