Skip to content

02 – Graph Execution

A DAG lets states fan out (one node, many children) and merge (many nodes, one child). Both are plain State flow — the engine handles the plumbing.

🎯 Goal

Resolve an entity candidate into a name through two alternative resolvers, then tag the result — while a Direct path bypasses resolution entirely.

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
┌─────────────────┐      ├─────────────────────┤
│ entity.extract  │ ──┬──▶ resolve_amzn ──┐
│                 │   │  └────────────────┴─┐
└─────────────────┘   │                     ├──▶ entity.tag
┌─────────────────┐   │  ┌────────────────┐ │
│ entity.direct   │ ──┼──▶ entity.direct  ─┤ (no children → terminal)
└─────────────────┘   │        │              
┌─────────────────┐   │        └─────────────┘
│ entity.kill     │ ──┘  (yields nothing → branch pruned)
└─────────────────┘

🧱 Nodes

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
from dagpipe import Graph, Node

class ExtractCandidate(Node):
    id = "entity.extract"

    def resolve(self, state):
        # fan-out: two candidate states for one input
        yield self.fork(state, payload_update={"entity.candidate": "amzn"})
        yield self.fork(state, payload_update={"entity.candidate": "flipkart"})

class ResolveAmazon(Node):
    id = "entity.resolve_amzn"

    def resolve(self, state):
        if state.get("entity.candidate") == "amzn":
            yield self.fork(state, payload_update={"entity.name": "Amazon"})

class ResolveFlipkart(Node):
    id = "entity.resolve_fk"

    def resolve(self, state):
        if state.get("entity.candidate") == "flipkart":
            yield self.fork(state, payload_update={"entity.name": "Flipkart"})

class TagNode(Node):
    id = "entity.tag"

    def resolve(self, state):
        # only runs when a name was resolved
        if state.get("entity.name"):
            tags = state.get("tags") or []
            yield self.fork(state, payload_update={"tags": tags + ["resolved"]})

class DirectNode(Node):
    id = "entity.direct"

    def resolve(self, state):
        yield self.fork(state, payload_update={"entity.name": "Direct"})

class KillNode(Node):
    id = "entity.kill"

    def resolve(self, state):
        return ()   # branch pruned: no states, no children visited

🏗️ Wire the graph

1
2
3
4
5
6
7
graph = Graph()
graph.add_edge(ExtractCandidate(), ResolveAmazon())
graph.add_edge(ExtractCandidate(), ResolveFlipkart())
graph.add_edge(ResolveAmazon(), TagNode())
graph.add_edge(ResolveFlipkart(), TagNode())
graph.add_root(DirectNode())
graph.add_root(KillNode())

🏃 Run it

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14
15
16
17
18
19
from dagpipe import Engine, Payload, Schema, State

class ItemState(State):
    schema = Schema({
        "raw": object,
        "entity": Schema({"candidate": str | None, "name": str | None}),
        "tags": list | None,
    })

root = ItemState(payload=Payload({"raw": object(), "tags": []}))
results = Engine(graph).run(root)

names = sorted(s.get("entity.name") for s in results)
assert names == ["Amazon", "Direct", "Flipkart"]

tagged = {s.get("entity.name"): s.get("tags") for s in results}
assert tagged["Amazon"] == ["resolved"]
assert tagged["Flipkart"] == ["resolved"]
assert tagged["Direct"] is None     # direct path never reached tag

🔍 What just happened

  • The engine enqueued all roots with the same root State.
  • ExtractCandidate forked two states → each resolver ran once per branch.
  • ResolveAmazon/ResolveFlipkart each produced at most one state (the wrong candidate branch pruned itself by yielding nothing).
  • TagNode merged both successful branches and appended "resolved".
  • DirectNode had no children → its state is terminal immediately.
  • KillNode yielded nothing → pruned, never contributes.

💡 Tips

  • Root states are shared. Every root receives the same initial State — they don't interact.
  • Merge = shared child. Both resolvers point at TagNode; the tag step runs once per incoming state.
  • Pruning happens naturally: yield nothing and the branch dies.
  • Cycle detection is automatic — ValueError on a self/indirect cycle at add_edge time.