A sequential pipeline: LangGraph lesson 2 - AI Tester Blueprint

01Real-world example

Illustration: Every Playwright test has a lifecycle: set up the browser, run the test, clean up. Each stage records what it did. — Every Playwright test has a lifecycle: set up the browser, run the test, clean up. Each stage records what it did.

02The graph

A sequential pipeline as a graph — setup, execute and teardown each add one key. stream() lets you watch every step.

03How the code does it

total=False lets the state start with only test_name; the other keys are filled in on the way.
Each node returns one key. Keys a node does not return keep their current value.
stream(..., stream_mode="updates") yields one dict per step: {node_name: what_it_changed}.

Key points

Edges run the steps in order
Keys you don't return stay as they were
stream(stream_mode="updates") shows each step

04Run it

terminal

cd chapter_18_LangGraph/src/chapters
python 002_Linear_Pipeline.py

output

setup     changed -> {'browser': 'chromium'}
execute   changed -> {'status': 'passed'}
teardown  changed -> {'cleaned_up': True}

DDrills for lesson 2

Work in chapter_18_LangGraph/src/chapters. Change one thing, run it, and compare with the expected result.

Insert a step
Add a report node between execute and teardown that returns {"report": "1 passed"}.

Expected result
The stream prints four lines in order: setup, execute, report, teardown.
Watch whole states instead of changes
Loop over app.stream(..., stream_mode="values") and print each item.

Expected result
Each line is the full state so far, growing by one key per step. The first line is your input: {'test_name': 'login_works'}.
Fail the test
Make execute return {"status": "failed"}. Does teardown still run?

Expected result
Yes. A plain edge always runs. To skip or branch you need a conditional edge, which is lesson 3.

SThe full script

The exact file from the course repo, plus the output it printed when this page was written.

002_Linear_Pipeline.py

"""002 - Three nodes in a row, like a test's setup -> execute -> teardown.

Each node returns a *partial* update. Keys it does not return are left alone.
stream_mode="updates" shows you exactly what each node changed, step by step.
"""
from typing import TypedDict

from langgraph.graph import END, START, StateGraph


class State(TypedDict, total=False):
    test_name: str
    browser: str
    status: str
    cleaned_up: bool


def setup(state: State) -> dict:
    return {"browser": "chromium"}


def execute(state: State) -> dict:
    return {"status": "passed"}


def teardown(state: State) -> dict:
    return {"cleaned_up": True}


builder = StateGraph(State)
builder.add_node("setup", setup)
builder.add_node("execute", execute)
builder.add_node("teardown", teardown)
builder.add_edge(START, "setup")
builder.add_edge("setup", "execute")
builder.add_edge("execute", "teardown")
builder.add_edge("teardown", END)

app = builder.compile()

if __name__ == "__main__":
    for step in app.stream({"test_name": "login_works"}, stream_mode="updates"):
        for node, update in step.items():
            print(f"{node:<9} changed -> {update}")

    print("\nFinal state:", app.invoke({"test_name": "login_works"}))

Expected output

output

setup     changed -> {'browser': 'chromium'}
execute   changed -> {'status': 'passed'}
teardown  changed -> {'cleaned_up': True}