What are the options for running workflows?¶

from inference import InferencePipeline
from inference.core.interfaces.camera.entities import VideoFrame

def my_sink(result: dict, video_frame: VideoFrame):
    print(result) # here you can find dictionary with outputs from your workflow


# initialize a pipeline object
pipeline = InferencePipeline.init_with_workflow(
    api_key="<YOUR-API-KEY>",
    workspace_name="<your-workspace-name>",
    workflow_id="<your-workflow-id>",
    video_reference=0, # Path to video, device id (int, usually 0 for built in webcams), or RTSP stream url
    on_prediction=my_sink,
    image_input_name="image",  # this parameter holds the name of Workflow input that represents 
    # image to be processed - please ADJUST it to your Workflow Definition 
)
pipeline.start() #start the pipeline
pipeline.join() #wait for the pipeline thread to finish

inference workflows process-images-directory \
    -i {your_input_directory} \
    -o {your_output_directory} \[workflows.py](..%2F..%2Finference_cli%2Fworkflows.py)
    --workspace_name {your-roboflow-workspace-url} \
    --workflow_id {your-workflow-id} \
    --api-key {your_roboflow_api_key}

from inference.core.env import WORKFLOWS_MAX_CONCURRENT_STEPS
from inference.core.managers.base import ModelManager
from inference.core.workflows.core_steps.common.entities import StepExecutionMode
from inference.core.env import MAX_ACTIVE_MODELS
from inference.core.managers.base import ModelManager
from inference.core.managers.decorators.fixed_size_cache import WithFixedSizeCache
from inference.core.registries.roboflow import RoboflowModelRegistry
from inference.models.utils import ROBOFLOW_MODEL_TYPES

# initialisation of Model registry to manage models load into memory 
# (required by core blocks exposing Roboflow models)
model_registry = RoboflowModelRegistry(ROBOFLOW_MODEL_TYPES)
model_manager = ModelManager(model_registry=model_registry)
model_manager = WithFixedSizeCache(model_manager, max_size=MAX_ACTIVE_MODELS)

# workflow definition
OBJECT_DETECTION_WORKFLOW = {
    "version": "1.0",
    "inputs": [
        {"type": "WorkflowImage", "name": "image"},
        {"type": "WorkflowParameter", "name": "model_id"},
        {"type": "WorkflowParameter", "name": "confidence", "default_value": 0.3},
    ],
    "steps": [
        {
            "type": "RoboflowObjectDetectionModel",
            "name": "detection",
            "image": "$inputs.image",
            "model_id": "$inputs.model_id",
            "confidence": "$inputs.confidence",
        }
    ],
    "outputs": [
        {"type": "JsonField", "name": "result", "selector": "$steps.detection.*"}
    ],
}

# example init paramaters for blocks - dependent on set of blocks 
# used in your workflow
workflow_init_parameters = {
    "workflows_core.model_manager": model_manager,
    "workflows_core.api_key": "<YOUR-API-KEY>,
    "workflows_core.step_execution_mode": StepExecutionMode.LOCAL,
}

# instance of Execution Engine - init(...) method invocation triggers
# the compilation process
execution_engine = ExecutionEngine.init(
    workflow_definition=OBJECT_DETECTION_WORKFLOW,
    init_parameters=workflow_init_parameters,
    max_concurrent_steps=WORKFLOWS_MAX_CONCURRENT_STEPS,
)

# runing the workflow
result = execution_engine.run(
    runtime_parameters={
        "image": [<your-image>],
        "model_id": "yolov8n-640",
    }
)