Skip to content

How To Build a Pipeline

Jump to bottom
Fernando Koch edited this page Mar 30, 2026 · 3 revisions

How To Build A Pipeline

Purpose

Use a pipeline when you want to compose nodes into a larger executable graph.

GraphK provides three main public pipeline forms:

  • SequencePipe
  • BranchPipe
  • MultiPipe

In most cases, start with SequencePipe.

Pipeline Mental Model

A pipeline is also a Node, but instead of doing one piece of work directly, it coordinates traversal across child nodes or nested pipelines.

This means:

  • pipelines can be nested
  • pipelines can carry context and policies
  • pipelines can be executed manually
  • pipelines are usually executed through Runner or through Emitter after registration in a Program

The Abstract Pipeline

graphk.Pipeline is the base type. It is usually not instantiated directly.

Important public methods:

  • add(node, index=None)
  • start(session)
  • step()
  • complete()
  • sequence()
  • forward()

The one abstract method is:

  • forward()

forward() defines traversal order. Concrete pipeline classes implement it differently.

Sequence Pipeline

Use SequencePipe when execution should happen in order.

Simple example:

import graphk


pipeline = graphk.SequencePipe(
    nodes=[
        graphk.demo.IncrementNode("A", increment=1, target_key="total", response=True),
        graphk.demo.IncrementNode("B", increment=2, target_key="total", response=True),
        graphk.demo.IncrementNode("C", increment=3, target_key="total", response=True),
    ]
)

This is the default composition type for GraphK.

Nested Pipelines

Pipelines can contain other pipelines.

Example:

import graphk


inner = graphk.SequencePipe(
    nodes=[
        graphk.demo.IncrementNode("Inner-A", increment=2, target_key="total", order_key="order", response=True),
        graphk.demo.IncrementNode("Inner-B", increment=3, target_key="total", order_key="order", response=True),
    ]
)

outer = graphk.SequencePipe(
    nodes=[
        graphk.demo.IncrementNode("Outer-Start", increment=1, target_key="total", order_key="order", response=True),
        inner,
        graphk.demo.IncrementNode("Outer-End", increment=4, target_key="total", order_key="order", response=True),
    ]
)

This is the normal way to build larger execution graphs from smaller reusable pieces.

Branching Pipelines

Use BranchPipe when exactly one matching route should execute.

Main inputs:

  • nodes
  • tests
  • selection_policy

Each branch item should itself be a pipeline.

Example shape:

import graphk


branch = graphk.BranchPipe(
    selection_policy=graphk.BranchPipe.SELECT_FIRST,
    nodes=[
        graphk.SequencePipe(nodes=[graphk.demo.RouteNode("A")]),
        graphk.SequencePipe(nodes=[graphk.demo.RouteNode("B")]),
    ],
    tests=[
        graphk.Matcher(task="route", route="alpha"),
        graphk.Matcher(task="route", route="beta"),
    ],
)

Use SELECT_FIRST unless you explicitly want random selection.

Multi-Branch Pipelines

Use MultiPipe when all matching routes should execute.

Example shape:

import graphk


pipe = graphk.MultiPipe(
    order_policy=graphk.MultiPipe.ORDER_STORED,
    nodes=[
        graphk.SequencePipe(nodes=[graphk.demo.RouteNode("A")]),
        graphk.SequencePipe(nodes=[graphk.demo.RouteNode("B")]),
    ],
    tests=[
        graphk.Matcher(task="route"),
        graphk.Matcher(task="route"),
    ],
)

Manual Pipeline Lifecycle

Low-level lifecycle:

  1. create a session
  2. call session.start()
  3. call pipeline.start(session)
  4. iterate nodes from the pipeline
  5. call pipeline.step() until each node finishes
  6. call pipeline.complete()

Example:

import graphk


pipeline = graphk.SequencePipe(
    nodes=[
        graphk.demo.IncrementNode("A", increment=1, target_key="counter"),
        graphk.demo.IncrementNode("B", increment=2, target_key="counter"),
    ]
)

session = graphk.Session(counter=0)
session.start()
pipeline.start(session)

for node in pipeline:
    node.start(session)
    while True:
        try:
            pipeline.step()
        except StopIteration:
            node.complete()
            break

pipeline.complete()
print(session.to_dict())

Preferred Pipeline Execution

Prefer Runner or Emitter instead of manual stepping.

Recommended pattern:

import graphk


pipeline = graphk.SequencePipe(nodes=[...])
session = graphk.Session(...)
runner = graphk.Runner(pipeline, session)

runner.start()
runner.run()
print(session.to_dict())

Program-dispatch pattern:

import graphk


pipeline = graphk.SequencePipe(nodes=[...])
program = graphk.Program({"main": pipeline})
emitter = graphk.Emitter(program)

session = emitter.request("main", graphk.Session(...))
print(session.to_dict())

If the program has one registered pipeline, or a configured default, the emitter can dispatch without an explicit id:

session = emitter.request(session=graphk.Session(...))

Common Mistakes

  • Do not instantiate Pipeline directly unless you are implementing a custom subclass.
  • Do not put plain nodes directly into BranchPipe branches; use branch pipelines.
  • Do not manually manage traversal when Runner already fits the use case.
  • Do not confuse sequence() with execution; sequence() resolves traversal order, it does not run the pipeline.

Learn More

Recommended references:

Clone this wiki locally