Keyboard shortcuts

Press ← or → to navigate between chapters

Press S or / to search in the book

Press ? to show this help

Press Esc to hide this help

Mini-project 6: build a workflow engine

Time to write the scheduler.

By the end of this chapter you will have a workflow engine in about two hundred lines of Python, with no framework anywhere in it, supporting:

  • sequential chaining
  • conditional routing
  • parallel fan-out with per-item payloads, and fan-in
  • typed shared state with declared reducers
  • per-step retry and recovery
  • hard budgets on supersteps, node runs, and wall-clock time

Everything runs offline. Every block of output below is real terminal output from the code as printed.

Same method as Part 1: build the simplest thing that could work, run it, watch it break, and add exactly the piece that fixes the break. Four versions. The failures are the curriculum.

Setup:

mkdir -p workflow-engine && cd workflow-engine
python3 -m venv .venv && source .venv/bin/activate

No dependencies until the LangGraph section at the end.

The workload

We need something with real structure, so: support ticket triage for Solaris Audio.

Load a ticket, classify it, gather evidence, draft a reply. The gathering step differs by category — a billing ticket needs the order record, the billing knowledge-base article, a sentiment read, and an enrichment lookup; a technical ticket needs a different article and the sentiment read, and no order at all.

That single sentence contains everything this chapter is about: a chain, a branch, a fan-out, a fan-in, and a step that can fail.

v1: a sequential chain

Start with the smallest thing that runs steps in order over shared state.

v1.py:

def run_chain(steps, state):
    for name, fn in steps:
        update = fn(state)
        state.update(update)
        print(f"  ran {name:12s} -> {sorted(update)}")
    return state

And the workflow — four plain functions over dict fixtures:

def load_ticket(state):
    return {"ticket": TICKETS[state["ticket_id"]]}

def classify(state):
    text = state["ticket"]["text"].lower()
    return {"category": "billing" if "charged" in text else "technical"}

def lookup_order(state):
    return {"order": ORDERS[state["ticket"]["order_id"]]}

def draft_reply(state):
    o = state["order"]
    return {"reply": f"We found {o['charges']} charges of ${o['total']}; refunding one."}

CHAIN = [("load_ticket", load_ticket), ("classify", classify),
         ("lookup_order", lookup_order), ("draft_reply", draft_reply)]

Two design choices already worth naming.

Steps return a partial update, not a mutated state. The engine owns the merge. That looks like ceremony now and becomes load-bearing in v3, when two steps return updates at the same time.

State is a plain dict. That is a bug we will fix in v2, and it is worth feeling it first.

Run it on the billing ticket, then on a technical one with no order ID:

$ python3 v1.py

--- happy path ---
  ran load_ticket  -> ['ticket']
  ran classify     -> ['category']
  ran lookup_order -> ['order']
  ran draft_reply  -> ['reply']
REPLY: We found 2 charges of $129.00; refunding one.

--- ticket with no order id ---
  ran load_ticket  -> ['ticket']
  ran classify     -> ['category']
Traceback (most recent call last):
  ...
  File "v1.py", line 33, in lookup_order
    return {"order": ORDERS[state["ticket"]["order_id"]]}
KeyError: 'order_id'

What v1 gets wrong

The crash is the obvious problem, but look at why it crashed.

lookup_order ran on a ticket that has no order, because a list has exactly one path through it. The category was computed one step earlier, sitting right there in the state, and the chain had no way to use it. There is no vocabulary for “it depends.”

Second, the state is a dict, so state["order"] in draft_reply is a runtime lookup into a bag with no schema. Nothing tells you which keys exist at which point, nothing stops a typo, and nothing stops two steps writing the same key.

Third, one bad step kills the run and takes all the completed work with it. load_ticket and classify both succeeded. That work is gone.

v2: typed state and a graph

Two fixes. Make the state a dataclass, and make the control flow data.

Start engine.py. State first:

from dataclasses import dataclass, field, fields, replace

def reduced(default_factory, reducer):
    """Declare a state field whose concurrent updates are combined, not clobbered."""
    return field(default_factory=default_factory, metadata={"reduce": reducer})


def merge(state, updates: list[dict]):
    reducers = {f.name: f.metadata.get("reduce") for f in fields(state)}
    out = {}
    for upd in updates:
        for key, value in upd.items():
            if key not in reducers:
                raise KeyError(f"update touches unknown state field {key!r}")
            red = reducers[key]
            if key in out:
                if red is None:
                    raise ValueError(
                        f"two parallel steps both wrote non-reduced field {key!r}"
                    )
                out[key] = red(out[key], value)
            else:
                out[key] = red(getattr(state, key), value) if red else value
    return replace(state, **out)

merge is thirty lines and it is the most important function in the engine. Read the three branches.

A key that is not a declared field raises immediately — that is your typo guard, and it fires at the step that made the mistake instead of three steps later when someone reads a None.

A key written twice in the same batch, with a reducer, folds.

A key written twice in the same batch, without a reducer, raises. That last one is the bug this whole design exists to prevent: two parallel branches both returning {"findings": [...]} and one silently winning. The engine’s position is that if you did not say how two concurrent writes combine, two concurrent writes are a programming error.

replace() returns a new state object rather than mutating, so a node can never see a half-updated state.

Now the graph. Nodes, edges, and one sentinel:

END = "__end__"

@dataclass(frozen=True)
class Send:
    """Dispatch one node with its own payload. A list of these fans out."""
    node: str
    payload: Any = None


@dataclass
class Node:
    name: str
    fn: Callable[..., dict]
    retries: int = 0
    on_error: Callable[[Exception, Any], dict] | None = None

And the builder, which is all bookkeeping:

class Workflow:
    def __init__(self, *, max_supersteps=25, max_node_runs=60, max_seconds=30.0):
        self.nodes: dict[str, Node] = {}
        self.edges: dict[str, Any] = {}      # name -> str | list[str] | router fn
        self.entry: str | None = None
        ...

    def node(self, name, fn=None, *, retries=0, on_error=None):
        def add(f):
            self.nodes[name] = Node(name, f, retries=retries, on_error=on_error)
            return f
        return add(fn) if fn else add

    def edge(self, src, dst):
        self.edges[src] = dst
        return self

    def branch(self, src, router):
        """router(state) returns a node name, a list of names, or a list of Sends."""
        self.edges[src] = router
        return self

Note self.edges maps a node name to either a string, a list, or a callable. A static edge and a conditional edge are the same field. That is the “control flow as data” idea from Chapter 1 made concrete: routing is a value in a dict, not a branch in the scheduler.

Add a validate() that runs before execution and checks three things: every edge source and every static target names a real node, every node has an outgoing edge, and the entry node exists. Twelve lines, and it turns “the workflow silently ended after step two” into an exception at startup. It cannot check the targets of a router function — those are only known at runtime — which is a genuine trade-off of dynamic routing and a good reason to keep routers small enough to unit-test.

What v2 still gets wrong

We have typed state and a graph, and we still execute one node at a time.

For the billing ticket, four evidence-gathering steps are completely independent — the order lookup, the article search, the sentiment read, the enrichment call. Each is I/O bound and takes about 200 ms. Run sequentially, that is 800 ms of latency the user waits for no reason.

v3: supersteps and parallel execution

The scheduler. Here is the whole thing:

    def run(self, state):
        self.validate()
        started = time.monotonic()
        frontier = [Send(self.entry)]
        runs = 0
        for step in range(1, self.max_supersteps + 1):
            if not frontier:
                return state
            if runs + len(frontier) > self.max_node_runs:
                raise BudgetExceeded(f"node-run budget ({self.max_node_runs}) exceeded")
            if time.monotonic() - started > self.max_seconds:
                raise BudgetExceeded(f"time budget ({self.max_seconds}s) exceeded")

            names = ", ".join(s.node for s in frontier)
            self.log(f"[superstep {step}] {names}")

            frozen = state                    # every node in a step sees the same state
            with ThreadPoolExecutor(max_workers=max(1, len(frontier))) as pool:
                futures = [
                    pool.submit(self._run_node, self.nodes[s.node], frozen, s.payload)
                    for s in frontier
                ]
                updates = [f.result() or {} for f in futures]
            runs += len(frontier)

            for send, upd in zip(frontier, updates):
                self.log(f"        {send.node:16s} -> {sorted(upd) or '(no update)'}")
            state = merge(state, updates)

            nxt, seen = [], set()
            for send in frontier:
                for s in self._next(send.node, state):
                    key = (s.node, repr(s.payload))
                    if key not in seen:
                        seen.add(key)
                        nxt.append(s)
            frontier = nxt
        raise BudgetExceeded(f"superstep budget ({self.max_supersteps}) exceeded")

Five things are happening and each is deliberate.

frozen = state before the pool. Every node in a superstep sees the identical input. No node can observe a sibling’s partial work, so the result does not depend on thread scheduling. This one line is what makes the parallelism deterministic.

updates = [f.result() ...] in frontier order, not completion order. merge folds in a fixed sequence, so findings comes out in the same order on every run. If you collected results with as_completed() you would get a faster-looking loop and a non-reproducible output list, which is a miserable thing to debug.

Budgets checked before the work, not after. Three of them: supersteps, total node runs, wall-clock seconds. The node-run budget is the one that saves you when a fan-out is bigger than you expected — twenty supersteps of one node each is cheap, but one superstep dispatching two thousand workers is not, and only max_node_runs catches that.

Deduplication of the next frontier. Four nodes all pointing at gather produce four Send("gather") entries. The seen set collapses them to one, which is exactly the fan-in join. You do not need a special “join node” concept — deduplication is the join.

BudgetExceeded is raised, not returned. Different from the Part 1 agent, where step exhaustion returned a polite string. Here the caller is your own code, not a user, and exhausting a workflow budget means the graph is wrong. It should be loud.

The edge evaluation is small enough to read in one go:

    def _next(self, name, state) -> list[Send]:
        dst = self.edges[name]
        if callable(dst):
            dst = dst(state)
        if isinstance(dst, Send):
            dst = [dst]
        if isinstance(dst, str):
            dst = [dst]
        out = []
        for item in dst:
            if isinstance(item, Send):
                out.append(item)
            elif item != END:
                out.append(Send(item))
        return out

Static edge, conditional edge, single Send, list of Send — all normalise to a list of Send. END normalises to nothing, which is how a run ends: the frontier goes empty.

v4: per-node error policy

One piece left. In v1 a failing step killed the run. Nodes are not all equally important, and the engine should let you say so.

    def _run_node(self, node, state, payload):
        attempt = 0
        while True:
            try:
                if payload is None:
                    return node.fn(state)
                return node.fn(state, payload)
            except Exception as exc:              # noqa: BLE001 - deliberate
                if attempt < node.retries:
                    attempt += 1
                    self.log(f"        retry {node.name} after {type(exc).__name__}")
                    continue
                if node.on_error is not None:
                    self.log(f"        {node.name} failed "
                             f"({type(exc).__name__}: {exc}) -> recovery")
                    return node.on_error(exc, payload)
                raise

Three policies, declared at the node:

  • retries=n — transient failures, retried in place. Add jitter and backoff here for anything hitting a real network.
  • on_error=fn — the node is optional or degradable. The handler returns a normal partial update, so a failure becomes data in the state rather than an exception. Downstream nodes can then decide what to do about it.
  • neither — the node is essential. It fails, the run fails, loudly.

That middle policy is the one that earns its keep. Notice the shape it takes in the triage workflow:

@wf.node("fetch_order", retries=1,
         on_error=lambda exc, _p: {"errors": [f"fetch_order: {type(exc).__name__}"]})
def fetch_order(s: Triage) -> dict:
    order_id = s.ticket["order_id"]                 # KeyError on a ticket with no order
    return {"order": ORDERS[order_id], "findings": [...]}

The failure lands in errors, which is a reduced field, so it accumulates alongside everything else and the drafting node can read it. This is the Part 1 rule — errors are observations — moved up a level. At the tool layer, a failure becomes text the model can read. At the workflow layer, a failure becomes state a downstream node can read. Same principle, same payoff: the system degrades instead of dying.

Running the whole thing

triage.py wires it up. State first:

@dataclass
class Triage:
    ticket_id: str
    ticket: dict = field(default_factory=dict)
    category: str = ""
    order: dict | None = None
    findings: list[str] = reduced(list, operator.add)
    errors: list[str] = reduced(list, operator.add)
    reply: str = ""

Two reduced fields, four single-writer fields. That declaration is the concurrency contract for the entire workflow, in seven lines, and you can review it.

The router, which does the branch and the fan-out in one move:

def route(s: Triage):
    if s.category == "billing":
        return [Send("fetch_order"), Send("search_kb", "billing"),
                Send("score_sentiment"), Send("flaky_enrich")]
    return [Send("search_kb", "crash"), Send("score_sentiment")]

Note Send("search_kb", "billing") versus Send("search_kb", "crash"). Same node, different payload, chosen at runtime. search_kb takes (state, topic) and never reads the category — it does not need to know why it was called.

And the wiring:

(wf.start("load")
   .edge("load", "classify")
   .branch("classify", route)
   .edge("fetch_order", "gather")
   .edge("search_kb", "gather")
   .edge("score_sentiment", "gather")
   .edge("flaky_enrich", "gather")
   .edge("gather", "draft")
   .edge("draft", END))

gather is an empty node that returns {}. Its only job is to be a place all four branches point at, so they collapse into one frontier entry.

Each of the gathering nodes sleeps 200 ms to stand in for I/O, and flaky_enrich fails 75% of the time on purpose, with retries=2 and an on_error handler.

$ python3 triage.py

===== T-901 =====
[superstep 1] load
        load             -> ['ticket']
[superstep 2] classify
        classify         -> ['category']
[superstep 3] fetch_order, search_kb, score_sentiment, flaky_enrich
        retry flaky_enrich after TimeoutError
        retry flaky_enrich after TimeoutError
        flaky_enrich failed (TimeoutError: upstream enrichment service timed out) -> recovery
        fetch_order      -> ['findings', 'order']
        search_kb        -> ['findings']
        score_sentiment  -> ['findings']
        flaky_enrich     -> ['errors']
[superstep 4] gather
        gather           -> (no update)
[superstep 5] draft
        draft            -> ['reply']
REPLY: [billing] R. Okafor: order 12345: 2 charges; kb[billing]: Duplicate charges are refunded automatically within 5 business days.; sentiment: neutral (degraded: 1 step(s) failed)
errors: ['flaky_enrich gave up: upstream enrichment service timed out']
wall clock: 0.20s

===== T-902 =====
[superstep 1] load
        load             -> ['ticket']
[superstep 2] classify
        classify         -> ['category']
[superstep 3] search_kb, score_sentiment
        search_kb        -> ['findings']
        score_sentiment  -> ['findings']
[superstep 4] gather
        gather           -> (no update)
[superstep 5] draft
        draft            -> ['reply']
REPLY: [technical] M. Diallo: kb[crash]: Known issue #44: crash on launch for firmware < 2.3. Fix: update firmware.; sentiment: frustrated
errors: []
wall clock: 0.20s

Read that output carefully, because four separate claims from this chapter are visible in it.

The routing decision fired. T-901 dispatched four nodes at superstep 3; T-902 dispatched two, and never touched fetch_order — which is the exact step that crashed v1 on this ticket.

The parallelism is real. Four nodes, 200 ms each, wall clock 0.20 s. Sequentially that superstep is 0.8 s.

The reducer worked. Three nodes wrote findings concurrently and all three survived, in frontier order, every run.

The error policy worked. flaky_enrich was retried twice, gave up, and its failure became an errors entry that draft read and reported as a degradation. The run produced a useful answer anyway.

The budget, proved

Give the engine a graph that never terminates — a refine step whose critic is never satisfied:

wf = Workflow(max_supersteps=4)

@wf.node("refine")
def refine(s: S) -> dict:
    return {"tries": s.tries + 1}

wf.start("refine").branch("refine", lambda s: "refine")

[superstep 1] refine
        refine           -> ['tries']
[superstep 2] refine
        refine           -> ['tries']
[superstep 3] refine
        refine           -> ['tries']
[superstep 4] refine
        refine           -> ['tries']
STOPPED: superstep budget (4) exceeded

Cycles are legal in this engine, which is what lets you build reflection and iterative-refinement loops. The budget is what stops a cycle from being a bug that costs you money overnight.

The same workflow in LangGraph

Now map it onto the real tool.

pip install langgraph          # 1.2.10 at the time of writing

import operator
from typing import Annotated

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


class Triage(TypedDict, total=False):
    ticket_id: str
    ticket: dict
    category: str
    order: dict
    findings: Annotated[list[str], operator.add]
    errors: Annotated[list[str], operator.add]
    reply: str

Your reduced(list, operator.add) is their Annotated[list[str], operator.add]. Identical idea, and if you got the reducer concept from the previous section you already understand LangGraph state, which is the piece people most often get wrong.

Nodes are the same plain functions returning partial updates:

def classify(state: Triage) -> dict:
    text = state["ticket"]["text"].lower()
    return {"category": "billing" if "charge" in text else "technical"}

The router returns Send objects, exactly as yours does:

def route(state: Triage):
    if state["category"] == "billing":
        return [Send("fetch_order", state), Send("search_kb", {"topic": "billing"}),
                Send("score_sentiment", state)]
    return [Send("search_kb", {"topic": "crash"}), Send("score_sentiment", state)]

One real difference: in LangGraph a node reached by Send receives the payload instead of the state, so if a node needs the state you pass the state as the payload. That trips people up once. Your engine passes both, which is friendlier but means your nodes have two signatures. Neither choice is wrong; know which one you are in.

Assembly:

builder = StateGraph(Triage)
for name, fn in [("load", load), ("classify", classify), ("fetch_order", fetch_order),
                 ("search_kb", search_kb), ("score_sentiment", score_sentiment),
                 ("draft", draft)]:
    builder.add_node(name, fn)

builder.add_edge(START, "load")
builder.add_edge("load", "classify")
builder.add_conditional_edges("classify", route,
                              ["fetch_order", "search_kb", "score_sentiment"])
builder.add_edge("fetch_order", "draft")
builder.add_edge("search_kb", "draft")
builder.add_edge("score_sentiment", "draft")
builder.add_edge("draft", END)

graph = builder.compile()

START and END are sentinels, same as your END. The third argument to add_conditional_edges is the list of possible targets — it is optional at runtime but it is what makes the graph drawable, so always pass it. Note there is no explicit gather: LangGraph joins at draft automatically because all three branches have an edge to it. Your engine does the same thing via frontier deduplication; you just had to write a no-op node to have somewhere to point.

$ python3 lg_triage.py

T-901: [billing] R. Okafor: order 12345: 2 charges; kb[billing]: Duplicate charges are refunded automatically within 5 business days.; sentiment: neutral
  wall clock: 0.21s
T-902: [technical] M. Diallo: kb[crash]: Known issue #44: crash on launch for firmware < 2.3. Fix: update firmware.; sentiment: frustrated
  wall clock: 0.20s

Same answers, same parallelism, same routing.

What LangGraph gives you that yours does not

An honest list, because this is why you would use it.

Durable execution. Compile with a checkpointer and every superstep boundary is persisted:

from langgraph.checkpoint.memory import InMemorySaver

graph = builder.compile(checkpointer=InMemorySaver())
config = {"configurable": {"thread_id": "ticket-901"}}
out = graph.invoke({"ticket_id": "T-901"}, config, durability="sync")

print(len(list(graph.get_state_history(config))))   # -> 6

Six checkpoints for a five-superstep run, one per boundary plus the initial state, each replayable. Use PostgresSaver in production; InMemorySaver loses everything on restart.

Streaming. graph.stream(...) yields state updates as each node finishes, which is how you drive a UI that shows progress instead of a spinner.

Interrupts. interrupt() inside a node suspends the run; Command(resume=value) continues it, possibly days later in a different process. That is the human-in-the-loop gate, and it only works because of the checkpointer.

Subgraphs, retries with backoff policies, node caching, and tracing integration.

None of that changes the model in your head. It is all built on nodes, edges, reduced state, and supersteps — which you now have written yourself.

Extensions worth doing

Each is under fifty lines and each teaches something.

A cost budget. Add max_cost and have nodes report tokens spent in a reduced field. Check it at the superstep barrier alongside the other budgets. Step counts are a bad proxy for money.

Structured tracing. Replace self.log with a JSON line per node: run ID, superstep, node, payload hash, duration, outcome. You will want exactly this when Part 5 gets to observability, and retrofitting it is annoying.

Checkpointing. Serialize (state, frontier) at each barrier and add a resume(run_id). It is about thirty lines for a JSON-file version and it will teach you more about durable execution than any blog post.

Async and per-node timeouts. Swap the thread pool for asyncio.gather; most workflow nodes are I/O bound, so this is the version you would actually ship. Then add per-node timeouts — and notice that future.result(timeout=...) on a thread pool does not actually stop the thread, which is a limitation better felt than read about.

Cycle detection. Warn when the same (node, payload) pair appears in a frontier for the third time. The budget catches runaway loops; this catches them with a message that says which node is spinning.

What you should be able to do now

  • Build a graph-based workflow engine from scratch with nodes returning partial updates, static and conditional edges, and a superstep scheduler.
  • Explain why freezing state at the superstep barrier and merging results in frontier order is what makes parallel execution deterministic, and predict the bug you get when either is missing.
  • Declare per-field reducers on a typed state object, and know why a collision on a non-reduced field must be a loud error rather than a last-write-wins.
  • Express a runtime-sized fan-out with per-item payloads, and get the fan-in for free through frontier deduplication.
  • Assign an error policy per node — retry, degrade into state, or fail the run — and design a workflow that produces a useful degraded answer when an optional step dies.
  • Enforce budgets on supersteps, total node runs, and wall-clock time, and say which budget catches which class of runaway.
  • Read a LangGraph graph definition and name the piece of your own engine each line corresponds to.

Further reading