Add Per-Class Confidence Filter workflow block (#2283)

Adds a new transformation block roboflow_core/per_class_confidence_filter@v1
that filters detection predictions using a different confidence threshold
per class. Detections are kept only if their confidence is at least the
threshold configured for their class, with a configurable default_threshold
fallback for classes that are not listed.

The existing detections_filter block can express this via its query language,
but requires a verbose StatementGroup/BinaryStatement configuration. This
block exposes a simple {"class_name": threshold} dictionary for the common
per-class case.

Supports object detection, instance segmentation, and keypoint detection
prediction kinds. Includes a unit test suite covering manifest validation,
boundary conditions, unknown-class fallback, empty inputs, and batch
processing, plus an integration test exercising the full ExecutionEngine
path with synthetic detections.

Co-authored-by: Paweł Pęczek <146137186+PawelPeczek-Roboflow@users.noreply.github.com>

Author: dcaroboflow

inference/core/workflows/core_steps/loader.py

@@ -476,6 +476,9 @@
 from inference.core.workflows.core_steps.transformations.image_slicer.v2 import (
     ImageSlicerBlockV2,
 )
+from inference.core.workflows.core_steps.transformations.per_class_confidence_filter.v1 import (
+    PerClassConfidenceFilterBlockV1,
+)
 from inference.core.workflows.core_steps.transformations.perspective_correction.v1 import (
     PerspectiveCorrectionBlockV1,
 )
@@ -744,6 +747,7 @@ def load_blocks() -> List[Type[WorkflowBlock]]:
         DynamicCropBlockV1,
         DetectionsFilterBlockV1,
         DetectionOffsetBlockV1,
+        PerClassConfidenceFilterBlockV1,
         DepthEstimationBlockV1,
         ByteTrackerBlockV1,
         RelativeStaticCropBlockV1,

inference/core/workflows/core_steps/transformations/per_class_confidence_filter/__init__.py

@@ -0,0 +1,168 @@
+from typing import Any, Dict, List, Literal, Optional, Type, Union
+
+import numpy as np
+import supervision as sv
+from pydantic import ConfigDict, Field
+
+from inference.core.workflows.execution_engine.entities.base import (
+    Batch,
+    OutputDefinition,
+)
+from inference.core.workflows.execution_engine.entities.types import (
+    DICTIONARY_KIND,
+    FLOAT_ZERO_TO_ONE_KIND,
+    INSTANCE_SEGMENTATION_PREDICTION_KIND,
+    KEYPOINT_DETECTION_PREDICTION_KIND,
+    OBJECT_DETECTION_PREDICTION_KIND,
+    Selector,
+)
+from inference.core.workflows.prototypes.block import (
+    BlockResult,
+    WorkflowBlock,
+    WorkflowBlockManifest,
+)
+
+SHORT_DESCRIPTION = "Filter detections by applying a per-class confidence threshold."
+
+LONG_DESCRIPTION = """
+Filter detection predictions by applying a different confidence threshold to each class, keeping only detections whose confidence meets or exceeds the threshold configured for their class (with a configurable fallback threshold for classes that are not listed).
+
+## How This Block Works
+
+This block applies class-aware confidence filtering to detection predictions, enabling precise control over which detections are retained based on per-class quality requirements. The block:
+
+1. Takes detection predictions (object detection, instance segmentation, or keypoint detection) and a dictionary mapping class names to confidence thresholds
+2. Iterates through each detection, looking up the threshold associated with the detection's class name
+3. If the class is not present in the dictionary, falls back to the configurable `default_threshold` value
+4. Keeps only the detections whose confidence is greater than or equal to the resolved threshold
+5. Returns the filtered detections while preserving all original metadata (class ids, masks, keypoints, tracker ids, etc.)
+
+Unlike a single global confidence threshold, this block lets you demand high-confidence predictions for classes that are prone to false positives while keeping a more permissive threshold for classes that are harder to detect. Unlike the generic detections filter, it exposes a purpose-built dictionary input that maps cleanly to a simple `{"class_name": threshold}` JSON object.
+
+## Common Use Cases
+
+- **Noise-prone classes**: Demand very high confidence (e.g. 0.9) for classes that frequently produce false positives, while accepting lower confidence for well-behaved classes
+- **Hard-to-detect classes**: Lower the threshold for classes that the model rarely detects with high confidence so that they are not filtered out entirely
+- **Production-grade filtering**: Apply domain-specific thresholds tuned during evaluation so that downstream analytics, alerts, or counting blocks only see detections that meet the project's quality bar
+- **Multi-class pipelines**: Combine with object detection models that predict many classes at once when a single global confidence threshold is too coarse
+
+## Connecting to Other Blocks
+
+The filtered predictions from this block can be connected to:
+
+- **Visualization blocks** (Bounding Box Visualization, Label Visualization, Polygon Visualization) to render only detections that cleared their per-class threshold
+- **Counting and analytics blocks** (Line Counter, Time in Zone, Velocity) so that metrics reflect only high-quality detections
+- **Tracking blocks** (Byte Tracker) so that tracker associations are not polluted by low-confidence noise
+- **Storage or sink blocks** (Roboflow Dataset Upload, Webhook Sink, CSV Formatter) so that only detections meeting the quality bar are persisted or transmitted
+- **Downstream transformation blocks** (Dynamic Crop, Detection Offset) for subsequent processing on the filtered subset
+"""
+
+
+class BlockManifest(WorkflowBlockManifest):
+    model_config = ConfigDict(
+        json_schema_extra={
+            "name": "Per-Class Confidence Filter",
+            "version": "v1",
+            "short_description": SHORT_DESCRIPTION,
+            "long_description": LONG_DESCRIPTION,
+            "license": "Apache-2.0",
+            "block_type": "transformation",
+            "ui_manifest": {
+                "section": "flow_control",
+                "icon": "far fa-filter",
+                "blockPriority": 2,
+            },
+        }
+    )
+    type: Literal["roboflow_core/per_class_confidence_filter@v1"]
+    predictions: Selector(
+        kind=[
+            OBJECT_DETECTION_PREDICTION_KIND,
+            INSTANCE_SEGMENTATION_PREDICTION_KIND,
+            KEYPOINT_DETECTION_PREDICTION_KIND,
+        ]
+    ) = Field(
+        description="Detection predictions to filter. Each detection is kept only if its confidence is greater than or equal to the threshold configured for its class (with a fallback to default_threshold for classes that are not listed in class_thresholds).",
+        examples=["$steps.object_detection_model.predictions"],
+    )
+    class_thresholds: Union[
+        Dict[str, float],
+        Selector(kind=[DICTIONARY_KIND]),
+    ] = Field(
+        description="Mapping of class name to minimum confidence threshold. Detections whose class name is present in this dictionary are kept only if their confidence is at least the corresponding threshold. Classes not present fall back to default_threshold. Thresholds should be in the [0.0, 1.0] range.",
+        examples=[{"person": 0.98, "car": 0.5}, "$inputs.class_thresholds"],
+    )
+    default_threshold: Union[float, Selector(kind=[FLOAT_ZERO_TO_ONE_KIND])] = Field(
+        default=0.3,
+        description="Confidence threshold applied to detections whose class name is not listed in class_thresholds. Must be in the [0.0, 1.0] range.",
+        examples=[0.3, "$inputs.default_threshold"],
+    )
+
+    @classmethod
+    def get_parameters_accepting_batches(cls) -> List[str]:
+        return ["predictions"]
+
+    @classmethod
+    def describe_outputs(cls) -> List[OutputDefinition]:
+        return [
+            OutputDefinition(
+                name="predictions",
+                kind=[
+                    OBJECT_DETECTION_PREDICTION_KIND,
+                    INSTANCE_SEGMENTATION_PREDICTION_KIND,
+                    KEYPOINT_DETECTION_PREDICTION_KIND,
+                ],
+            ),
+        ]
+
+    @classmethod
+    def get_execution_engine_compatibility(cls) -> Optional[str]:
+        return ">=1.3.0,<2.0.0"
+
+
+class PerClassConfidenceFilterBlockV1(WorkflowBlock):
+
+    @classmethod
+    def get_manifest(cls) -> Type[WorkflowBlockManifest]:
+        return BlockManifest
+
+    def run(
+        self,
+        predictions: Batch[sv.Detections],
+        class_thresholds: Dict[str, Any],
+        default_threshold: float = 0.3,
+    ) -> BlockResult:
+        return [
+            {
+                "predictions": filter_detections_by_class_confidence(
+                    detections=detections,
+                    class_thresholds=class_thresholds,
+                    default_threshold=default_threshold,
+                )
+            }
+            for detections in predictions
+        ]
+
+
+def filter_detections_by_class_confidence(
+    detections: sv.Detections,
+    class_thresholds: Dict[str, Any],
+    default_threshold: float = 0.3,
+) -> sv.Detections:
+    if detections is None or len(detections) == 0:
+        return detections
+    confidences = detections.confidence
+    if confidences is None:
+        return detections
+    class_names = detections.data.get("class_name", [])
+    thresholds = {str(k): float(v) for k, v in (class_thresholds or {}).items()}
+    default = float(default_threshold)
+    keep: List[int] = []
+    for i, confidence in enumerate(confidences):
+        class_name = class_names[i] if i < len(class_names) else None
+        threshold = thresholds.get(str(class_name), default)
+        if float(confidence) >= threshold:
+            keep.append(i)
+    if len(keep) == len(detections):
+        return detections
+    return detections[np.array(keep, dtype=int)]

inference/core/workflows/core_steps/transformations/per_class_confidence_filter/v1.py

@@ -0,0 +1,158 @@
+import numpy as np
+import supervision as sv
+
+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.workflows.execution_engine.core import ExecutionEngine
+
+
+PER_CLASS_CONFIDENCE_FILTER_WORKFLOW = {
+    "version": "1.3.0",
+    "inputs": [
+        {
+            "type": "WorkflowBatchInput",
+            "name": "predictions",
+            "kind": ["object_detection_prediction"],
+        },
+        {"type": "WorkflowParameter", "name": "class_thresholds"},
+        {"type": "WorkflowParameter", "name": "default_threshold", "default_value": 0.3},
+    ],
+    "steps": [
+        {
+            "type": "roboflow_core/per_class_confidence_filter@v1",
+            "name": "filter",
+            "predictions": "$inputs.predictions",
+            "class_thresholds": "$inputs.class_thresholds",
+            "default_threshold": "$inputs.default_threshold",
+        }
+    ],
+    "outputs": [
+        {
+            "type": "JsonField",
+            "name": "filtered",
+            "selector": "$steps.filter.predictions",
+        }
+    ],
+}
+
+
+def _make_detections(
+    class_names: list[str], confidences: list[float]
+) -> sv.Detections:
+    n = len(class_names)
+    return sv.Detections(
+        xyxy=np.array([[0, 0, 10, 10]] * n, dtype=np.float64),
+        class_id=np.arange(n),
+        confidence=np.array(confidences, dtype=np.float64),
+        data={
+            "class_name": np.array(class_names),
+            "detection_id": np.array([f"d{i}" for i in range(n)]),
+        },
+    )
+
+
+def test_per_class_confidence_filter_end_to_end(
+    model_manager: ModelManager,
+) -> None:
+    # given
+    workflow_init_parameters = {
+        "workflows_core.model_manager": model_manager,
+        "workflows_core.api_key": None,
+        "workflows_core.step_execution_mode": StepExecutionMode.LOCAL,
+    }
+    execution_engine = ExecutionEngine.init(
+        workflow_definition=PER_CLASS_CONFIDENCE_FILTER_WORKFLOW,
+        init_parameters=workflow_init_parameters,
+        max_concurrent_steps=WORKFLOWS_MAX_CONCURRENT_STEPS,
+    )
+    predictions = _make_detections(
+        class_names=["person", "person", "car", "dog"],
+        confidences=[0.99, 0.7, 0.6, 0.4],
+    )
+
+    # when
+    result = execution_engine.run(
+        runtime_parameters={
+            "predictions": [predictions],
+            "class_thresholds": {"person": 0.98, "car": 0.5},
+            "default_threshold": 0.5,
+        }
+    )
+
+    # then
+    assert isinstance(result, list)
+    assert len(result) == 1
+    filtered: sv.Detections = result[0]["filtered"]
+    assert list(filtered.data["class_name"]) == ["person", "car"]
+    assert list(filtered.confidence) == [0.99, 0.6]
+
+
+def test_per_class_confidence_filter_default_threshold_filters_unknown_class(
+    model_manager: ModelManager,
+) -> None:
+    # given
+    workflow_init_parameters = {
+        "workflows_core.model_manager": model_manager,
+        "workflows_core.api_key": None,
+        "workflows_core.step_execution_mode": StepExecutionMode.LOCAL,
+    }
+    execution_engine = ExecutionEngine.init(
+        workflow_definition=PER_CLASS_CONFIDENCE_FILTER_WORKFLOW,
+        init_parameters=workflow_init_parameters,
+        max_concurrent_steps=WORKFLOWS_MAX_CONCURRENT_STEPS,
+    )
+    predictions = _make_detections(
+        class_names=["cat", "cat"],
+        confidences=[0.2, 0.8],
+    )
+
+    # when
+    result = execution_engine.run(
+        runtime_parameters={
+            "predictions": [predictions],
+            "class_thresholds": {"person": 0.98},
+            "default_threshold": 0.5,
+        }
+    )
+
+    # then
+    filtered: sv.Detections = result[0]["filtered"]
+    assert list(filtered.data["class_name"]) == ["cat"]
+    assert list(filtered.confidence) == [0.8]
+
+
+def test_per_class_confidence_filter_handles_batch_of_images(
+    model_manager: ModelManager,
+) -> None:
+    # given
+    workflow_init_parameters = {
+        "workflows_core.model_manager": model_manager,
+        "workflows_core.api_key": None,
+        "workflows_core.step_execution_mode": StepExecutionMode.LOCAL,
+    }
+    execution_engine = ExecutionEngine.init(
+        workflow_definition=PER_CLASS_CONFIDENCE_FILTER_WORKFLOW,
+        init_parameters=workflow_init_parameters,
+        max_concurrent_steps=WORKFLOWS_MAX_CONCURRENT_STEPS,
+    )
+    predictions_image_1 = _make_detections(
+        class_names=["person", "car"], confidences=[0.99, 0.4]
+    )
+    predictions_image_2 = _make_detections(
+        class_names=["car"], confidences=[0.55]
+    )
+
+    # when
+    result = execution_engine.run(
+        runtime_parameters={
+            "predictions": [predictions_image_1, predictions_image_2],
+            "class_thresholds": {"person": 0.98, "car": 0.5},
+            "default_threshold": 0.3,
+        }
+    )
+
+    # then
+    assert len(result) == 2
+    assert list(result[0]["filtered"].data["class_name"]) == ["person"]
+    assert list(result[1]["filtered"].data["class_name"]) == ["car"]