Merge pull request #1350 from roboflow/codex/add-workflow-blocks-for-perception-encoder

Add perception encoder workflow blocks

Author: grzegorz-roboflow

docs/foundation/perception_encoder.md

@@ -97,8 +97,15 @@ Run the code to use Perception Encoder on your webcam.
 
 **Note:** The model will take a minute or two to load. You will not see output while the model is loading.
 
+## Using PE in Workflows
+
+Perception Encoder can be used in Roboflow Workflows via the
+**Perception Encoder Embedding Model** block. This block lets you generate
+embeddings for images or text without writing code.
+
 ## API Compatibility
 
 The Perception Encoder model uses the **same API as CLIP**. This means you can use all the same methods and request/response formats as you would with CLIP, including `embed_text`, `embed_image`, and `compare`.
 
 For more details and advanced usage, see the [CLIP documentation](./clip.md).
+

inference/core/version.py

@@ -1,4 +1,4 @@
-__version__ = "0.50.5"
+__version__ = "0.51.0"
 
 
 if __name__ == "__main__":

inference/core/workflows/core_steps/loader.py

@@ -203,6 +203,9 @@
 from inference.core.workflows.core_steps.models.foundation.openai.v3 import (
     OpenAIBlockV3,
 )
+from inference.core.workflows.core_steps.models.foundation.perception_encoder.v1 import (
+    PerceptionEncoderModelBlockV1,
+)
 from inference.core.workflows.core_steps.models.foundation.qwen.v1 import (
     Qwen25VLBlockV1,
 )
@@ -561,6 +564,7 @@ def load_blocks() -> List[Type[WorkflowBlock]]:
         ClipComparisonBlockV1,
         ClipComparisonBlockV2,
         ClipModelBlockV1,
+        PerceptionEncoderModelBlockV1,
         CogVLMBlockV1,
         ColorVisualizationBlockV1,
         ConvertGrayscaleBlockV1,

inference/core/workflows/core_steps/models/foundation/perception_encoder/__init__.py

@@ -0,0 +1,189 @@
+import hashlib
+from typing import List, Literal, Optional, Type, Union
+
+from pydantic import ConfigDict, Field
+
+from inference.core.cache.lru_cache import LRUCache
+from inference.core.entities.requests.perception_encoder import (
+    PerceptionEncoderImageEmbeddingRequest,
+    PerceptionEncoderTextEmbeddingRequest,
+)
+from inference.core.env import (
+    HOSTED_CORE_MODEL_URL,
+    LOCAL_INFERENCE_API_URL,
+    WORKFLOWS_REMOTE_API_TARGET,
+)
+from inference.core.managers.base import ModelManager
+from inference.core.workflows.core_steps.common.entities import StepExecutionMode
+from inference.core.workflows.core_steps.common.utils import load_core_model
+from inference.core.workflows.execution_engine.entities.base import (
+    OutputDefinition,
+    WorkflowImageData,
+)
+from inference.core.workflows.execution_engine.entities.types import (
+    EMBEDDING_KIND,
+    IMAGE_KIND,
+    STRING_KIND,
+    Selector,
+)
+from inference.core.workflows.prototypes.block import (
+    BlockResult,
+    WorkflowBlock,
+    WorkflowBlockManifest,
+)
+from inference_sdk import InferenceHTTPClient
+
+LONG_DESCRIPTION = """
+Use the Meta Perception Encoder model to create semantic embeddings of text and images.
+
+This block accepts an image or string and returns an embedding. The embedding can be used to compare
+similarity between different images or between images and text.
+"""
+
+
+class BlockManifest(WorkflowBlockManifest):
+    model_config = ConfigDict(
+        json_schema_extra={
+            "name": "Perception Encoder Embedding Model",
+            "version": "v1",
+            "short_description": "Generate an embedding of an image or string.",
+            "long_description": LONG_DESCRIPTION,
+            "license": "MIT",
+            "block_type": "model",
+            "ui_manifest": {
+                "section": "model",
+                "icon": "far fa-paperclip",
+                "blockPriority": 9.9,
+            },
+        }
+    )
+    type: Literal["roboflow_core/perception_encoder@v1"]
+    name: str = Field(description="Unique name of step in workflows")
+    data: Union[Selector(kind=[IMAGE_KIND, STRING_KIND]), str] = Field(
+        title="Data",
+        description="The string or image to generate an embedding for.",
+        examples=["$inputs.image", "$steps.cropping.crops"],
+    )
+    version: Union[
+        Literal[
+            "PE-Core-B16-224",
+            "PE-Core-L14-336",
+            "PE-Core-G14-448",
+        ],
+        Selector(kind=[STRING_KIND]),
+    ] = Field(
+        default="PE-Core-L14-336",
+        description="Variant of Perception Encoder model",
+        examples=["PE-Core-B16-224", "$inputs.variant"],
+    )
+
+    @classmethod
+    def describe_outputs(cls) -> List[OutputDefinition]:
+        return [OutputDefinition(name="embedding", kind=[EMBEDDING_KIND])]
+
+    @classmethod
+    def get_execution_engine_compatibility(cls) -> Optional[str]:
+        return ">=1.3.0,<2.0.0"
+
+
+text_cache = LRUCache()
+
+
+class PerceptionEncoderModelBlockV1(WorkflowBlock):
+    def __init__(
+        self,
+        model_manager: ModelManager,
+        api_key: Optional[str],
+        step_execution_mode: StepExecutionMode,
+    ):
+        self._model_manager = model_manager
+        self._api_key = api_key
+        self._step_execution_mode = step_execution_mode
+
+    @classmethod
+    def get_init_parameters(cls) -> List[str]:
+        return ["model_manager", "api_key", "step_execution_mode"]
+
+    @classmethod
+    def get_manifest(cls) -> Type[WorkflowBlockManifest]:
+        return BlockManifest
+
+    def run(
+        self,
+        data: Union[WorkflowImageData, str],
+        version: str,
+    ) -> BlockResult:
+        if self._step_execution_mode is StepExecutionMode.LOCAL:
+            return self.run_locally(data=data, version=version)
+        elif self._step_execution_mode is StepExecutionMode.REMOTE:
+            return self.run_remotely(data=data, version=version)
+        else:
+            raise ValueError(
+                f"Unknown step execution mode: {self._step_execution_mode}"
+            )
+
+    def run_locally(
+        self,
+        data: Union[WorkflowImageData, str],
+        version: str,
+    ) -> BlockResult:
+        if isinstance(data, str):
+            hash_key = hashlib.md5((version + data).encode("utf-8")).hexdigest()
+            cached_value = text_cache.get(hash_key)
+            if cached_value is not None:
+                return {"embedding": cached_value}
+            inference_request = PerceptionEncoderTextEmbeddingRequest(
+                perception_encoder_version_id=version,
+                text=[data],
+                api_key=self._api_key,
+            )
+            pe_model_id = load_core_model(
+                model_manager=self._model_manager,
+                inference_request=inference_request,
+                core_model="perception_encoder",
+            )
+            predictions = self._model_manager.infer_from_request_sync(
+                pe_model_id, inference_request
+            )
+            text_cache.set(hash_key, predictions.embeddings[0])
+            return {"embedding": predictions.embeddings[0]}
+        else:
+            inference_request = PerceptionEncoderImageEmbeddingRequest(
+                perception_encoder_version_id=version,
+                image=[data.to_inference_format(numpy_preferred=True)],
+                api_key=self._api_key,
+            )
+            pe_model_id = load_core_model(
+                model_manager=self._model_manager,
+                inference_request=inference_request,
+                core_model="perception_encoder",
+            )
+            predictions = self._model_manager.infer_from_request_sync(
+                pe_model_id, inference_request
+            )
+            return {"embedding": predictions.embeddings[0]}
+
+    def run_remotely(
+        self,
+        data: Union[WorkflowImageData, str],
+        version: str,
+    ) -> BlockResult:
+        api_url = (
+            LOCAL_INFERENCE_API_URL
+            if WORKFLOWS_REMOTE_API_TARGET != "hosted"
+            else HOSTED_CORE_MODEL_URL
+        )
+        client = InferenceHTTPClient(api_url=api_url, api_key=self._api_key)
+        if WORKFLOWS_REMOTE_API_TARGET == "hosted":
+            client.select_api_v0()
+        if isinstance(data, str):
+            result = client.get_perception_encoder_text_embeddings(
+                text=data,
+                perception_encoder_version=version,
+            )
+        else:
+            result = client.get_perception_encoder_image_embeddings(
+                inference_input=data.base64_image,
+                perception_encoder_version=version,
+            )
+        return {"embedding": result["embeddings"][0]}

inference/core/workflows/core_steps/models/foundation/perception_encoder/v1.py

@@ -1389,6 +1389,44 @@ async def clip_compare_async(
                 response.raise_for_status()
                 return await response.json()
 
+    @wrap_errors
+    def get_perception_encoder_image_embeddings(
+        self,
+        inference_input: Union[ImagesReference, List[ImagesReference]],
+        perception_encoder_version: Optional[str] = None,
+    ) -> Union[dict, List[dict]]:
+        """Get Perception Encoder embeddings for input image(s)."""
+        extra_payload = {}
+        if perception_encoder_version is not None:
+            extra_payload["perception_encoder_version_id"] = perception_encoder_version
+        result = self._post_images(
+            inference_input=inference_input,
+            endpoint="/perception_encoder/embed_image",
+            extra_payload=extra_payload,
+        )
+        return unwrap_single_element_list(result)
+
+    @wrap_errors
+    def get_perception_encoder_text_embeddings(
+        self,
+        text: Union[str, List[str]],
+        perception_encoder_version: Optional[str] = None,
+    ) -> Union[dict, List[dict]]:
+        """Get Perception Encoder embeddings for input text(s)."""
+        payload = self.__initialise_payload()
+        payload["text"] = text
+        if perception_encoder_version is not None:
+            payload["perception_encoder_version_id"] = perception_encoder_version
+        response = requests.post(
+            self.__wrap_url_with_api_key(
+                f"{self.__api_url}/perception_encoder/embed_text"
+            ),
+            json=payload,
+            headers=DEFAULT_HEADERS,
+        )
+        api_key_safe_raise_for_status(response=response)
+        return unwrap_single_element_list(sequence=response.json())
+
     @deprecated(
         reason="Please use run_workflow(...) method. This method will be removed end of Q2 2024"
     )

inference_sdk/http/client.py

@@ -110,6 +110,7 @@ nav:
           - L2CS-Net (Gaze Detection): foundation/gaze.md
           - Moondream2: foundation/moondream2.md
           - PaliGemma: foundation/paligemma.md
+          - Perception Encoder: foundation/perception_encoder.md
           - Segment Anything (Segmentation): foundation/sam.md
           - Segment Anything 2 (Segmentation): foundation/sam2.md
           - SmolVLM2: foundation/smolvlm.md