graphk 1.0.7

Framework for Graph-based execution and pipeline programming

GraphK

---

Overview

GraphK is a Python framework for graph-based execution and pipeline programming. It models execution as explicit nodes, pipelines, runtime sessions, scoped belief state, policy checks, and a program hub that stores pipelines separately from execution.

The public surface is centered on:

• Store and Matcher
• Session, Belief, and Policy
• Node
• Pipeline, SequencePipe, BranchPipe, and MultiPipe
• Program
• Runner
• Emitter

Report issues at GraphK Issues.

Getting Started

Install from PyPI when published:

pip install graphk

All examples use the public import form:

import graphk

Core Mental Model

The preferred architecture is:

1. build one or more pipelines
2. register them in a Program
3. create an Emitter(program) or Runner(pipeline, session)
4. dispatch a payload to a pipeline id
5. read the final session from emitter.response

Program is the storage and binding hub. Emitter is a dispatcher. Runner is the transient execution engine.

Experiment 1: First Request And Response

What is being presented?

The smallest useful public GraphK flow: one program, one registered pipeline, one emitter, one session, one response.

What does it mean?

This example shows the preferred request/response surface after separating pipeline storage from execution.

import graphk


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

emitter.request("echo", payload={"value": 5})

print(emitter.response)

Expected result:

{"echo": 5, "double": 10}

Experiment 2: Sequential Pipeline

What is being presented?

Several nodes running in a fixed order inside one pipeline registered under a program id.

What does it mean?

The pipeline is the execution architecture. The program stores it. The emitter dispatches a session to it.

import graphk


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

program = graphk.Program({"sequence": pipeline})
emitter = graphk.Emitter(program)
emitter.request("sequence", payload={"counter": 0})

print(emitter.response)

Expected result:

{"counter": 6, "order": ["A", "B", "C"]}

Experiment 3: Belief And Policy

What is being presented?

Scoped runtime values and rule-based execution checks flowing through Program, Runner, Pipeline, and Node.

What does it mean?

Program can hold shared belief state, Belief carries scoped execution data, and Policy checks entry and exit conditions against the active state.

import graphk


allow_entry = graphk.Policy(task="go", **{"@Node/stage": "work"})

pipeline = graphk.SequencePipe(
    nodes=[
        graphk.demo.IncrementNode(
            "Prepare",
            increment=2,
            target_key="counter",
            response=True,
            belief=graphk.Belief(stage="work"),
            entry_policy=allow_entry,
            exit_policy=graphk.Policy(counter=">=2"),
        )
    ]
)

program = graphk.Program(
    {"policy": pipeline},
    belief=graphk.Belief(mode="strict"),
)

emitter = graphk.Emitter(program)
emitter.request("policy", payload={"task": "go", "counter": 0})

print(emitter.response)

Experiment 4: Managed Execution With Runner

What is being presented?

The explicit execution engine for a selected pipeline.

What does it mean?

Runner is still the main execution API when session lifecycle and pipeline control need to remain visible. Program is only required when you want dispatch by pipeline id.

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)
runner = graphk.Runner(pipeline, session)

runner.start()
runner.run()

print(session.to_dict())

Experiment 5: Branching And Fan-Out

What is being presented?

Two advanced routing forms: exclusive routing with BranchPipe and cumulative routing with MultiPipe, both registered under a Program.

What does it mean?

BranchPipe selects one matching branch. MultiPipe executes every matching branch. Emitter chooses which top-level pipeline to run by program id.

import graphk


branch = graphk.BranchPipe(
    nodes=[
        (graphk.Matcher(task="route-a"), graphk.SequencePipe(nodes=[graphk.demo.RouteNode("A")])),
        (graphk.Matcher(task="route-b"), graphk.SequencePipe(nodes=[graphk.demo.RouteNode("B")])),
    ]
)

multi = graphk.MultiPipe(
    nodes=[
        (graphk.Matcher(group="alpha"), graphk.SequencePipe(nodes=[graphk.demo.RouteNode("A")])),
        (graphk.Matcher(group="alpha"), graphk.SequencePipe(nodes=[graphk.demo.RouteNode("B")])),
    ]
)

program = graphk.Program(
    {
        "route": graphk.SequencePipe(nodes=[branch]),
        "fanout": graphk.SequencePipe(nodes=[multi]),
    }
)

emitter = graphk.Emitter(program)

route_session = emitter.request("route", payload={"task": "route-b"})
print(route_session.to_dict())

fanout_session = emitter.request("fanout", payload={"group": "alpha"})
print(fanout_session.to_dict())

Program And Emitter

Program stores pipelines in a registry:

program = graphk.Program()
program.set("main", pipeline)
pipeline = program.get("main")
ids = program.ids()

Emitter dispatches by pipeline id:

emitter = graphk.Emitter(program)
session = emitter.request("main", payload={"value": 10})
response = emitter.response

If the pipeline id is missing, Emitter.request() returns a session marked as error instead of crashing.

Learn More

The repository examples and tests work as learn-by-example material.

Examples:

• examples/example_1.py
• examples/example_4.py
• examples/example_5.py
• examples/example_6.py
• examples/example_7.py
• examples/example_8.py

Tests:

• tests/emitter/
• tests/sequence/
• tests/branch/
• tests/multi/

Wiki:

• GraphK Wiki