Home

GraphK WIKI

GraphK is built around graph programming.

Execution is modeled as a graph of small units with clear responsibilities. Each unit does one thing. Pipelines define structure. Programs store pipelines. Sessions carry runtime state. Policies and context define execution conditions and scoped values.

The result is a package for building process architectures that are:

• explicit
• composable
• state-aware
• easy to inspect

GraphK gives you a small execution model:

• a Node defines one executable unit of behavior
• a Pipeline turns nodes into an architecture
• a Program stores pipelines and shared binding scope
• a Session carries runtime state
• an Emitter or Runner executes the graph

The easiest way to understand that model is to follow one small story.

Illustrative Scenario

Consider a small processing service.

A request arrives with one value. The service must process that value through a few explicit steps, keep execution state visible, and return one final response. The architecture must remain readable in the code.

The story starts with one node.

import graphk


class IncrementNode(graphk.Node):

    def __init__(self, name: str, increment: int = 1, **kwargs) -> None:
        super().__init__(**kwargs)
        self._name = name
        self._increment = increment

    def ping(self) -> bool:
        return True

    def info(self) -> dict:
        return {"name": self._name}

    def step(self):
        current = self.session.get("value", 0)
        current += self._increment
        self.session.set("value", current)
        self.session.set("response", {"value": current})
        yield current

That node defines one unit of behavior. The next step is to give it an architecture.

import graphk


pipeline = graphk.SequencePipe(
    nodes=[
        IncrementNode("Prepare", increment=1),
        IncrementNode("Enrich", increment=2),
        IncrementNode("Finalize", increment=3),
    ]
)

program = graphk.Program({"main": pipeline})

To execute that graph, create the runtime state and dispatch it through the program.

import graphk


session = graphk.Session(value=10)
emitter = graphk.Emitter(program)

emitter.request("main", session)

print(emitter.response())
print(session.to_dict())

That is the first complete GraphK pattern:

1. define behavior with nodes
2. compose behavior into a pipeline
3. register the pipeline in a program
4. create a session
5. execute through an emitter or runner
6. inspect the final session and response

If the program has exactly one pipeline, or a configured default pipeline, the emitter can also dispatch without an explicit id. If dispatch is ambiguous or the id does not exist, GraphK raises an exception instead of silently failing.

The package grows from this same structure. More advanced use cases do not replace this model. They extend it with richer context, policies, branching, and multi-route execution.

Where To Go Next

• Learn By Example
• Extend GraphK