Dynamic Python Blocks

When the syntax for Workflow definitions was outlined, one key aspect was not covered: the ability to define blocks directly within the Workflow definition itself. This section can include the manifest and Python code for blocks defined in place, which are dynamically interpreted by the Execution Engine. These in-place blocks function similarly to those statically defined in plugins yet provide much more flexibility.

Execution Modes

Dynamic Python blocks support two execution modes:

Local Execution

When running inference locally on your own hardware, dynamic blocks execute directly in your environment. This provides the fastest performance for development and testing.

Cloud Execution (Roboflow Serverless v2)

When using Roboflow's cloud infrastructure with Serverless v2 API, dynamic blocks execute in secure, isolated containers. This ensures safe execution of custom code without compromising your infrastructure.

The cloud execution environment provides the same standard libraries and imports as local execution, ensuring your code works consistently across both modes.

State Management and Shared Data

Variables defined at the module level (outside of your run function) in your block's code are scoped to instances of that block. These variables:

• Persist across invocations of the same block (as long as the code doesn't change)
• Reset when the block's code changes — any modification to the block's code creates a new namespace
• Are lost when the server/container restarts

Example:

This block increments a counter each time the block is run and remembers the last result:

# This variable is block-scoped
counter = 0
last_result = None

def run(self, input_value):
    global counter, last_result

    counter += 1

    previous = last_result
    last_result = input_value * 2

    return {
        "run_count": counter,
        "current": last_result,
        "previous": previous
    }

Best Practices for State Management

Custom Block state is meant for caching expensive computations and optimization of artifact and dependency loading.

• Do not rely on state for critical data persistence — use external storage for important data.
• State may be lost at any time due to server restarts or container scaling.
• In cloud environments, subsequent requests may hit different servers with different state.
• Initialize block-scoped variables with default values to handle fresh starts.
• Keep state lightweight. Large objects consume memory and may impact performance.

Theory

The high-level overview of Dynamic Python blocks functionality:

• User provides definition of dynamic block in JSON.
• Definition contains information required by Execution Engine to construct WorkflowBlockManifest and WorkflowBlock out of the document.
• At runtime, the Compiler turns the definition into dynamically created Python classes — exactly the same as statically defined blocks.
• In the Workflow definition, you may declare steps that use dynamic blocks, as if dynamic blocks were standard static ones.

Example

Here is an example workflow with dynamic Python blocks that measures overlap between detected objects:

{
    "version": "1.0",
    "inputs": [
        {
            "type": "WorkflowImage",
            "name": "image"
        }
    ],
    "dynamic_blocks_definitions": [
        {
            "type": "DynamicBlockDefinition",
            "manifest": {
                "type": "ManifestDescription",
                "block_type": "OverlapMeasurement",
                "inputs": {
                    "predictions": {
                        "type": "DynamicInputDefinition",
                        "selector_types": ["step_output"]
                    },
                    "class_x": {
                        "type": "DynamicInputDefinition",
                        "value_types": ["string"]
                    },
                    "class_y": {
                        "type": "DynamicInputDefinition",
                        "value_types": ["string"]
                    }
                },
                "outputs": {
                    "overlap": {
                        "type": "DynamicOutputDefinition",
                        "kind": []
                    }
                }
            },
            "code": {
                "type": "PythonCode",
                "run_function_code": "..."
            }
        }
    ],
    "steps": [
        {
            "type": "RoboflowObjectDetectionModel",
            "name": "model",
            "image": "$inputs.image",
            "model_id": "rfdetr-small"
        },
        {
            "type": "OverlapMeasurement",
            "name": "overlap_measurement",
            "predictions": "$steps.model.predictions",
            "class_x": "dog",
            "class_y": "dog"
        }
    ],
    "outputs": [
        {
            "type": "JsonField",
            "name": "overlaps",
            "selector": "$steps.overlap_measurement.overlap"
        }
    ]
}

The analysis starts from dynamic_blocks_definitions — this is the part of the Workflow Definition that provides a list of dynamic blocks. Each block contains two sections:

• manifest — providing JSON representation of BlockManifest (refer to the blocks development guide)
• code — shipping Python code

Definition of Block Manifest

Manifest definition contains several fields, including:

• block_type — equivalent of type field in block manifest; must provide unique block identifier.
• inputs — dictionary with names and definitions of dynamic inputs.
• outputs — dictionary with names and definitions of dynamic outputs.
• output_dimensionality_offset — field specifies output dimensionality.
• accepts_batch_input — field dictates if input data in runtime is to be provided in batches by Execution Engine.
• accepts_empty_values — field deciding if empty inputs will be ignored while constructing step inputs.

Definition of Dynamic Input

Each input may define the following properties:

• has_default_value — flag to decide if dynamic manifest field has default.
• default_value — default value (used only if has_default_value=True).
• is_optional — flag to decide if dynamic manifest field is optional.
• is_dimensionality_reference — flag to decide if dynamic manifest field ships selector to be used in runtime as dimensionality reference.
• dimensionality_offset — dimensionality offset for configured input property.
• selector_types — type of selectors that may be used by property (one of input_image, step_output_image, input_parameter, step_output).
• selector_data_kind — dictionary with list of selector kinds specific for each selector type.
• value_types — definition of specific type to be placed in manifest. Selection of types: any, integer, float, boolean, dict, list, string.

Definition of Dynamic Output

Definitions of outputs are quite simple, holding an optional list of kinds declared for the given output.

Definition of Python Code

Python code is shipped in a JSON document with the following fields:

• run_function_code — code of run(...) method of your dynamic block.
• run_function_name — name of run function.
• init_function_code — optional code for your init function that will assemble step state; it is expected to return a dictionary available to the run() function under self._init_results.
• init_function_name — name of init function.
• imports — list of additional imports (you may only use libraries from your environment; no dependencies will be automatically installed).

How to Create the run(...) Method

You must know the following:

• run(...) function must be defined as if it were a class instance method — with the first argument being self and remaining arguments compatible with the dynamic block manifest.
• You should expect baseline symbols to be provided, including your import statements and the following:

from typing import Any, List, Dict, Set, Optional
import supervision as sv
import numpy as np
import math
import time
import json
import os
import requests
import cv2
import shapely
from inference.core.workflows.execution_engine.entities.base import Batch, WorkflowImageData
from inference.core.workflows.prototypes.block import BlockResult

An example run function:

def run(self, predictions: sv.Detections, class_x: str, class_y: str) -> BlockResult:
    bboxes_class_x = predictions[predictions.data["class_name"] == class_x]
    bboxes_class_y = predictions[predictions.data["class_name"] == class_y]
    overlap = []
    for bbox_x in bboxes_class_x:
        bbox_x_coords = bbox_x[0]
        bbox_overlaps = []
        for bbox_y in bboxes_class_y:
            if bbox_y[-1]["detection_id"] == bbox_x[-1]["detection_id"]:
                continue
            bbox_y_coords = bbox_y[0]
            x_min = max(bbox_x_coords[0], bbox_y_coords[0])
            y_min = max(bbox_x_coords[1], bbox_y_coords[1])
            x_max = min(bbox_x_coords[2], bbox_y_coords[2])
            y_max = min(bbox_x_coords[3], bbox_y_coords[3])
            intersection_area = max(0, x_max - x_min + 1) * max(0, y_max - y_min + 1)
            box_x_area = (bbox_x_coords[2] - bbox_x_coords[0] + 1) * (bbox_x_coords[3] - bbox_x_coords[1] + 1)
            local_overlap = intersection_area / (box_x_area + 1e-5)
            bbox_overlaps.append(local_overlap)
        overlap.append(bbox_overlaps)
    return {"overlap": overlap}

Usage of Dynamic Python Block as Step

As shown in the example Workflow definition, you may simply use the block as if it were a normal block exposed through a static plugin:

{
    "type": "OverlapMeasurement",
    "name": "overlap_measurement",
    "predictions": "$steps.model.predictions",
    "class_x": "dog",
    "class_y": "dog"
}