Bundling Workflows Blocks

To efficiently manage the Workflows ecosystem, a standardized method for building and distributing blocks is essential. This allows users to create their own blocks and bundle them into Workflow plugins. A Workflow plugin is essentially a Python library that implements a defined interface and can be structured in various ways.

This page outlines the mandatory interface requirements and suggests a structure for blocks that aligns with the Workflows versioning guidelines.

Proposed Structure of Plugin

.
├── requirements.txt
├── setup.py
├── {plugin_name}
│   ├── __init__.py    # main module that contains loaders
│   ├── kinds.py       # optionally - definitions of custom kinds
│   ├── {block_name}   # package for your block
│   │   ├── v1.py      # version 1 of your block
│   │   ├── ...
│   │   └── v5.py      # version 5 of your block
│   └── {block_name}   # package for another block
└── tests

Required Interface

Plugin must only provide a few extensions to __init__.py in the main package compared to a standard Python library:

• load_blocks() function to provide a list of blocks' classes (required)
• load_kinds() function to return all custom kinds the plugin defines (optional)
• REGISTERED_INITIALIZERS module property which is a dict mapping name of block init parameter into default value or parameter-free function constructing that value (optional)

load_blocks() Function

from typing import List, Type
from inference.core.workflows.prototypes.block import WorkflowBlock

from my_plugin.block_1.v1 import Block1V1
from my_plugin.block_2.v1 import Block2V1

def load_blocks() -> List[Type[WorkflowBlock]]:
    return [
        Block1V1,
        Block2V1,
    ]

load_kinds() Function

from typing import List
from inference.core.workflows.execution_engine.entities.types import Kind

from my_plugin.kinds import MY_KIND

def load_kinds() -> List[Kind]:
    return [MY_KIND]

REGISTERED_INITIALIZERS Dictionary

import os

def init_my_param() -> str:
    return "some-value"

REGISTERED_INITIALIZERS = {
    "param_1": 37,
    "param_2": init_my_param,
}

Serializers and Deserializers for Kinds

Support for custom serializers and deserializers was introduced in Execution Engine v1.3.0. From that version onward it is possible to point custom functions that the Execution Engine should use to serialize and deserialize any kind.

from typing import Any

def serialize_kind(value: Any) -> Any:
    pass

def deserialize_kind(parameter_name: str, value: Any) -> Any:
    pass

KINDS_SERIALIZERS = {
    "name_of_the_kind": serialize_kind,
}
KINDS_DESERIALIZERS = {
    "name_of_the_kind": deserialize_kind,
}

Tips and Tricks

• Each serializer must be a function taking the value to serialize and returning serialized value (accepted by default Python JSON encoder).
• Each deserializer must be a function accepting two parameters — name of Workflow input to be deserialized and the value to be deserialized.
• Kinds from roboflow_core plugin already have reasonable serializers and deserializers.
• If you do not like the way how data is serialized in roboflow_core plugin, feel free to alter the serialization methods for kinds in your plugin — the serializer/deserializer defined last will be in use.

Enabling Plugin in Your Workflows Ecosystem

To load a plugin you must:

• Install the Python package with the plugin in the environment you run Workflows.
• Export an environment variable named WORKFLOWS_PLUGINS set to a comma-separated list of names of plugins you want to load.

Example:

export WORKFLOWS_PLUGINS="plugin_a,plugin_b"