Chapter 10 · AI Architect

Pipeline / DAG design

DAG · dependencies · topological sort · cycle detectionARCH-90369 of 84

A simple pipeline is a straight line: step 1, step 2, step 3. A real AI pipeline is a DAG — a *directed acyclic graph*. fetch and embed have no dependencies and could run in either order; retrieve needs both of them; answer needs retrieve. You can't just pick any order — each step must run after the steps it depends on.

Working out a valid order is a topological sort. The clean way is Kahn's algorithm:

Build a map of each node to its set of unmet dependencies.
Repeatedly find a node with no unmet deps (it's *ready*), place it in the order, and remove it from everyone else's dependency set.
Repeat until every node is placed.

To make the result deterministic, when several nodes are ready pick the alphabetically-smallest one. And here's the cycle check, free of charge: if at some point no node is ready but nodes remain, they all depend on each other in a loop — that's a cycle, and a cyclic graph has *no* valid order. Raise an error rather than loop forever.

This is pure Python; in production it's Airflow / Dagster / Prefect — the same DAG idea driven by a real scheduler.

Syntax

def topo_order(stages):
    deps = {s["name"]: set(s["deps"]) for s in stages}
    order = []
    while deps:
        ready = sorted(n for n, d in deps.items() if not d)  # no unmet deps
        if not ready:
            raise ValueError("cycle detected")               # nothing ready, nodes remain
        node = ready[0]                                       # smallest-ready (deterministic)
        order.append(node)
        del deps[node]
        for d in deps.values():
            d.discard(node)                                   # node is now satisfied for others
    return order

Worked examples

A diamond: a → (b, c) → d, ordered correctly

def topo_order(stages):
    deps = {s["name"]: set(s["deps"]) for s in stages}
    order = []
    while deps:
        ready = sorted(n for n, d in deps.items() if not d)
        if not ready:
            raise ValueError("cycle detected")
        node = ready[0]
        order.append(node)
        del deps[node]
        for d in deps.values():
            d.discard(node)
    return order

stages = [
    {"name": "d", "deps": ["b", "c"]},
    {"name": "b", "deps": ["a"]},
    {"name": "c", "deps": ["a"]},
    {"name": "a", "deps": []},
]
print(topo_order(stages))
# Output:
# ['a', 'b', 'c', 'd']

A cycle has no valid order — detect it, don't loop forever

def topo_order(stages):
    deps = {s["name"]: set(s["deps"]) for s in stages}
    order = []
    while deps:
        ready = sorted(n for n, d in deps.items() if not d)
        if not ready:
            raise ValueError("cycle detected")
        node = ready[0]
        order.append(node)
        del deps[node]
        for d in deps.values():
            d.discard(node)
    return order

cyclic = [
    {"name": "x", "deps": ["y"]},
    {"name": "y", "deps": ["x"]},   # x needs y, y needs x — impossible
]
try:
    topo_order(cyclic)
except ValueError as e:
    print("raised:", e)
# Output:
# raised: cycle detected

run_dag threads each step's artifact to its dependants

def topo_order(stages):
    deps = {s["name"]: set(s["deps"]) for s in stages}
    order = []
    while deps:
        ready = sorted(n for n, d in deps.items() if not d)
        if not ready:
            raise ValueError("cycle detected")
        node = ready[0]
        order.append(node)
        del deps[node]
        for d in deps.values():
            d.discard(node)
    return order

def run_dag(stages):
    by_name = {s["name"]: s for s in stages}
    artifacts = {}
    for name in topo_order(stages):
        stage = by_name[name]
        inputs = {dep: artifacts[dep] for dep in stage["deps"]}
        artifacts[name] = stage["run"](inputs)
    return artifacts

stages = [
    {"name": "answer",   "deps": ["retrieve"],       "run": lambda i: f"answer({i['retrieve']})"},
    {"name": "retrieve", "deps": ["fetch", "embed"], "run": lambda i: f"hits[{i['fetch']}|{i['embed']}]"},
    {"name": "fetch",    "deps": [],                 "run": lambda i: "docs"},
    {"name": "embed",    "deps": [],                 "run": lambda i: "vecs"},
]
print(topo_order(stages))
print(run_dag(stages)["answer"])
# Output:
# ['embed', 'fetch', 'retrieve', 'answer']
# answer(hits[docs|vecs])

Common mistakes & gotchas

Forgetting to discard the placed node from others

The line d.discard(node) is the heart of Kahn's algorithm — once a node is placed, every node that depended on it no longer has to wait. Skip it and no further node ever becomes ready, so the loop either stalls into a false 'cycle detected' or (without the guard) spins forever.

A cycle that silently loops instead of raising

Without the if not ready: raise check, a cyclic graph leaves nodes that never become ready and the while deps loop never empties — an infinite loop. The check turns 'impossible to order' into a clear, fast ValueError. Always have a termination guard on graph algorithms.

Assuming a unique order

A DAG often has many valid topological orders (the diamond could be a,b,c,d or a,c,b,d). That's correct — any order that respects the dependencies is valid. Picking smallest-ready-first just makes the output reproducible; don't write code that depends on one specific order beyond the dependency constraints.

Why it matters

The moment a pipeline has more than one upstream input, it's a DAG, and ordering it by hand is how subtle 'ran before its data was ready' bugs creep in. Topological sort is the standard, provably-correct answer — every workflow engine (Airflow, Dagster, build systems like Make, even package managers) is doing this underneath. Knowing it means you can reason about, and debug, all of them.