diff --git a/src/main/python/systemds/scuro/drsearch/hyperparameter_tuner.py b/src/main/python/systemds/scuro/drsearch/hyperparameter_tuner.py
index 094a4b9585e..8df2b403213 100644
--- a/src/main/python/systemds/scuro/drsearch/hyperparameter_tuner.py
+++ b/src/main/python/systemds/scuro/drsearch/hyperparameter_tuner.py
@@ -38,6 +38,8 @@
 from systemds.scuro.drsearch.task import PerformanceMeasure
 import pickle
 from systemds.scuro.utils.checkpointing import CheckpointManager
+from systemds.scuro.drsearch.process_cache import HPOExecutionCache
+from systemds.scuro.drsearch.operator_registry import is_expensive_representation
 
 
 def _wandb_safe_tag(s: str, max_len: int = 64) -> str:
@@ -319,6 +321,7 @@ def __init__(
         wandb_group: Optional[str] = None,
         wandb_tags: Optional[List[str]] = None,
         enable_checkpointing: bool = False,
+        representation_cache_size: int = 128,
     ):
         self.tasks = tasks
         self.unimodal_optimization_results = optimization_results
@@ -357,6 +360,14 @@ def __init__(
         self._wandb_run = None
         self.enable_checkpointing = enable_checkpointing
 
+        self.execution_cache = HPOExecutionCache(
+            representation_entries=representation_cache_size,
+            should_cache_representation=self._should_cache_representation,
+        )
+
+    def _should_cache_representation(self, operation: Any) -> bool:
+        return is_expensive_representation(operation)
+
     def get_modalities_by_id(self, modality_ids: List[int]) -> Modality:
         modalities = []
         for mod in self.modalities:
@@ -1004,15 +1015,14 @@ def evaluate_dag_config(
                 if modalities_override is not None
                 else self.get_modalities_by_id(modality_ids)
             )
-            modified_modality = dag_copy.execute(modalities, task)
+            modified_modality = dag_copy.execute(
+                modalities, task, execution_cache=self.execution_cache
+            )
             score = task.run(modified_modality.data)
 
             return params, score
         except Exception as e:
-            import traceback
-
-            traceback.print_exc()
-            self.logger.error(f"Error evaluating DAG with params {params}: {e}")
+            self.logger.warning("Skipping invalid DAG candidate %s: %s", params, e)
             return params, [np.nan, np.nan, np.nan]
 
     def tune_multimodal_representations(
diff --git a/src/main/python/systemds/scuro/drsearch/operator_registry.py b/src/main/python/systemds/scuro/drsearch/operator_registry.py
index d9ae798f867..df5483d5332 100644
--- a/src/main/python/systemds/scuro/drsearch/operator_registry.py
+++ b/src/main/python/systemds/scuro/drsearch/operator_registry.py
@@ -37,12 +37,14 @@ class Registry:
     _fusion_operators = []
     _context_representation_operators = {}
     _dimensionality_reduction_operators = {}
+    _expensive_representations = {}
 
     def __new__(cls):
         if not cls._instance:
             cls._instance = super().__new__(cls)
             for m_type in ModalityType:
                 cls._representations[m_type] = []
+                cls._expensive_representations[m_type] = []
         return cls._instance
 
     def set_fusion_operators(self, fusion_operators):
@@ -57,6 +59,12 @@ def set_representations(self, modality_type, representations):
         else:
             self._representations[modality_type] = [representations]
 
+    def set_expensive_representations(self, modality_type, representations):
+        if isinstance(representations, list):
+            self._expensive_representations[modality_type] = representations
+        else:
+            self._expensive_representations[modality_type] = [representations]
+
     def set_context_operators(self, modality_type, context_operators):
         if isinstance(context_operators, list):
             self._context_operators[modality_type] = context_operators
@@ -115,6 +123,11 @@ def add_context_representation_operator(
                 context_representation
             )
 
+    def add_expensive_representation(
+        self, representation: Representation, modality: ModalityType
+    ):
+        self._expensive_representations[modality].append(representation)
+
     def get_representations(self, modality: ModalityType):
         return self._representations[modality]
 
@@ -232,6 +245,17 @@ def _drop_windows_below_min_count(self, effective_window_lenghts, num_windows):
         lengths, counts = zip(*filtered)
         return list(lengths), list(counts)
 
+    def get_expensive_representations(self, modality: ModalityType):
+        return self._expensive_representations[modality]
+
+
+def is_expensive_representation(representation: Representation):
+    registry = Registry()
+    for modality in registry._expensive_representations:
+        if representation in registry._expensive_representations[modality]:
+            return True
+    return False
+
 
 def register_representation(modalities: Union[ModalityType, List[ModalityType]]):
     """
@@ -300,3 +324,21 @@ def decorator(cls):
         return cls
 
     return decorator
+
+
+def register_expensive_representation(
+    modalities: Union[ModalityType, List[ModalityType]],
+):
+    """
+    Decorator to register an expensive representation for a specific modality.
+    :param modalities: The modalities for which the representation is to be registered
+    """
+    if isinstance(modalities, ModalityType):
+        modalities = [modalities]
+
+    def decorator(cls):
+        for modality in modalities:
+            Registry().add_expensive_representation(cls, modality)
+        return cls
+
+    return decorator
diff --git a/src/main/python/systemds/scuro/drsearch/process_cache.py b/src/main/python/systemds/scuro/drsearch/process_cache.py
new file mode 100644
index 00000000000..361db144ff0
--- /dev/null
+++ b/src/main/python/systemds/scuro/drsearch/process_cache.py
@@ -0,0 +1,257 @@
+# -------------------------------------------------------------
+#
+# Licensed to the Apache Software Foundation (ASF) under one
+# or more contributor license agreements.  See the NOTICE file
+# distributed with this work for additional information
+# regarding copyright ownership.  The ASF licenses this file
+# to you under the Apache License, Version 2.0 (the
+# "License"); you may not use this file except in compliance
+# with the License.  You may obtain a copy of the License at
+#
+#   http://www.apache.org/licenses/LICENSE-2.0
+#
+# Unless required by applicable law or agreed to in writing,
+# software distributed under the License is distributed on an
+# "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+# KIND, either express or implied.  See the License for the
+# specific language governing permissions and limitations
+# under the License.
+#
+# -------------------------------------------------------------
+import hashlib
+import os
+import threading
+from collections import OrderedDict
+from typing import Any, Callable, Hashable, Optional, Tuple
+
+import numpy as np
+
+_MISSING = object()
+
+
+def freeze_cache_value(value: Any) -> Hashable:
+    if value is None or isinstance(value, (str, int, bool, bytes)):
+        return value
+    if isinstance(value, float):
+        return ("float", value.hex())
+    if isinstance(value, np.generic):
+        return freeze_cache_value(value.item())
+    if isinstance(value, type):
+        return ("class", value.__module__, value.__qualname__)
+    if isinstance(value, dict):
+        return tuple(
+            (str(key), freeze_cache_value(item))
+            for key, item in sorted(value.items(), key=lambda pair: str(pair[0]))
+        )
+    if isinstance(value, (list, tuple)):
+        return tuple(freeze_cache_value(item) for item in value)
+    if isinstance(value, set):
+        return tuple(sorted((freeze_cache_value(item) for item in value), key=repr))
+    if hasattr(value, "get_current_parameters"):
+        return (
+            type(value).__module__,
+            type(value).__qualname__,
+            freeze_cache_value(value.get_current_parameters()),
+        )
+    return (type(value).__module__, type(value).__qualname__, repr(value))
+
+
+def _update_fingerprint(digest, value: Any) -> None:
+    if isinstance(value, np.ndarray):
+        digest.update(b"array\0")
+        digest.update(value.dtype.str.encode())
+        digest.update(repr(value.shape).encode())
+        if value.dtype.hasobject:
+            for item in value.flat:
+                _update_fingerprint(digest, item)
+        else:
+            digest.update(value.tobytes(order="C"))
+        return
+    if isinstance(value, np.generic):
+        _update_fingerprint(digest, value.item())
+        return
+    if isinstance(value, str):
+        encoded = value.encode("utf-8", errors="surrogatepass")
+        digest.update(b"str\0")
+        digest.update(len(encoded).to_bytes(8, "little"))
+        digest.update(encoded)
+        return
+    if isinstance(value, bytes):
+        digest.update(b"bytes\0")
+        digest.update(len(value).to_bytes(8, "little"))
+        digest.update(value)
+        return
+    if value is None or isinstance(value, (bool, int, float)):
+        digest.update(repr(value).encode())
+        digest.update(b"\0")
+        return
+    if isinstance(value, dict):
+        digest.update(b"dict\0")
+        for key in sorted(value, key=str):
+            _update_fingerprint(digest, str(key))
+            _update_fingerprint(digest, value[key])
+        return
+    if isinstance(value, (list, tuple)):
+        digest.update(b"sequence\0")
+        digest.update(len(value).to_bytes(8, "little"))
+        for item in value:
+            _update_fingerprint(digest, item)
+        return
+    digest.update(repr(freeze_cache_value(value)).encode())
+
+
+def data_fingerprint(*values: Any) -> str:
+    digest = hashlib.sha256()
+    for value in values:
+        _update_fingerprint(digest, value)
+    return digest.hexdigest()
+
+
+class BoundedProcessCache:
+    def __init__(self, max_entries: int):
+        self.max_entries = max(0, int(max_entries))
+        self._values = OrderedDict()
+        self._pending = {}
+        self._lock = threading.RLock()
+
+    def get(self, key: Hashable, default=None):
+        with self._lock:
+            value = self._values.pop(key, _MISSING)
+            if value is _MISSING:
+                return default
+            self._values[key] = value
+            return value
+
+    def put(self, key: Hashable, value: Any) -> None:
+        if self.max_entries == 0:
+            return
+        with self._lock:
+            self._values.pop(key, None)
+            self._values[key] = value
+            while len(self._values) > self.max_entries:
+                self._values.popitem(last=False)
+
+    def get_or_compute(self, key: Hashable, compute: Callable[[], Any]) -> Any:
+        if self.max_entries == 0:
+            return compute()
+
+        while True:
+            with self._lock:
+                value = self._values.pop(key, _MISSING)
+                if value is not _MISSING:
+                    self._values[key] = value
+                    return value
+                pending = self._pending.get(key)
+                if pending is None:
+                    pending = threading.Event()
+                    self._pending[key] = pending
+                    owner = True
+                else:
+                    owner = False
+
+            if owner:
+                try:
+                    value = compute()
+                    self.put(key, value)
+                    return value
+                finally:
+                    with self._lock:
+                        self._pending.pop(key, None)
+                        pending.set()
+            pending.wait()
+
+    def clear(self) -> None:
+        with self._lock:
+            self._values.clear()
+
+    def __len__(self) -> int:
+        with self._lock:
+            return len(self._values)
+
+
+class HPOExecutionCache:
+    def __init__(
+        self,
+        representation_entries: int = 128,
+        should_cache_representation: Optional[Callable[[Any], bool]] = None,
+    ):
+        self.representations = BoundedProcessCache(representation_entries)
+        self.should_cache_representation = should_cache_representation
+        self._modality_keys = {}
+        self._lock = threading.RLock()
+
+    def modality_key(self, modality) -> Tuple[str, str]:
+        object_key = id(modality)
+        with self._lock:
+            cached = self._modality_keys.get(object_key)
+            if cached is not None:
+                return cached
+
+        loader = getattr(modality, "data_loader", None)
+        data = getattr(modality, "data", None)
+        has_data = data is not None and len(data) > 0
+        if has_data:
+            source = (
+                getattr(modality, "modality_type", None),
+                data,
+                getattr(modality, "metadata", None),
+            )
+        elif loader is not None:
+            source_path = getattr(loader, "source_path", None)
+            source_files = None
+            try:
+                file_names = loader.get_file_names(getattr(loader, "indices", None))
+                if isinstance(file_names, str):
+                    file_names = [file_names]
+                source_files = []
+                for file_name in file_names:
+                    if os.path.isfile(file_name):
+                        stat = os.stat(file_name)
+                        source_files.append(
+                            (os.path.abspath(file_name), stat.st_size, stat.st_mtime_ns)
+                        )
+            except (AttributeError, OSError, TypeError):
+                source_files = None
+            source = (
+                type(loader).__module__,
+                type(loader).__qualname__,
+                source_path,
+                source_files,
+                getattr(loader, "indices", None),
+                getattr(loader, "data", None),
+                getattr(loader, "metadata", None),
+                getattr(loader, "test_data", None),
+                getattr(loader, "_full_metadata", None),
+                getattr(modality, "modality_type", None),
+            )
+        else:
+            source = (type(modality).__module__, type(modality).__qualname__, data)
+
+        key = ("dataset", data_fingerprint(source))
+        with self._lock:
+            self._modality_keys[object_key] = key
+        return key
+
+    def representation_key(
+        self, operation: Any, parameters: Any, input_keys: Any
+    ) -> Hashable:
+        return (
+            "representation",
+            operation.__module__,
+            operation.__qualname__,
+            freeze_cache_value(parameters or {}),
+            tuple(input_keys),
+        )
+
+    def get_or_compute_representation(self, operation, key, compute):
+        if (
+            self.should_cache_representation is not None
+            and not self.should_cache_representation(operation)
+        ):
+            return compute()
+        return self.representations.get_or_compute(key, compute)
+
+    def clear(self) -> None:
+        self.representations.clear()
+        with self._lock:
+            self._modality_keys.clear()
diff --git a/src/main/python/systemds/scuro/drsearch/representation_dag.py b/src/main/python/systemds/scuro/drsearch/representation_dag.py
index 6315e853dda..fcafe5d2a8d 100644
--- a/src/main/python/systemds/scuro/drsearch/representation_dag.py
+++ b/src/main/python/systemds/scuro/drsearch/representation_dag.py
@@ -330,8 +330,31 @@ def execute(
         rep_cache: Dict[Any, TransformedModality] = None,
         consumer_count: Dict[str, int] = None,
         gpu_id: int = None,
+        execution_cache=None,
     ) -> Union[Dict[str, TransformedModality], TransformedModality]:
 
+        execution_keys = {}
+
+        def execution_key(node_id: str):
+            if execution_cache is None:
+                return None
+            if node_id in execution_keys:
+                return execution_keys[node_id]
+            node = self.get_node_by_id(node_id)
+            if not node.inputs:
+                modality = get_modality_by_id_and_instance_id(
+                    modalities, int(node.modality_id), node.representation_index
+                )
+                key = execution_cache.modality_key(modality)
+            else:
+                key = execution_cache.representation_key(
+                    node.operation,
+                    node.parameters,
+                    [execution_key(input_id) for input_id in node.inputs],
+                )
+            execution_keys[node_id] = key
+            return key
+
         def execute_node(node_id: str, task) -> TransformedModality:
             if external_cache is not None:
                 cached = external_cache.get(node_id)
@@ -350,45 +373,7 @@ def execute_node(node_id: str, task) -> TransformedModality:
             input_mods = [execute_node(input_id, task) for input_id in node.inputs]
             is_unimodal = len(input_mods) == 1
 
-            if external_cache is not None and is_unimodal:
-                cached = external_cache.get(node_id)
-                if cached is not None:
-                    result = cached
-                else:
-                    node_operation = node.operation(params=node.parameters)
-                    if gpu_id is not None and hasattr(node_operation, "gpu_id"):
-                        node_operation.gpu_id = gpu_id
-                    if len(input_mods) == 1:
-                        # It's a unimodal operation
-                        if isinstance(node_operation, Context):
-                            result = input_mods[0].context(node_operation)
-                        elif isinstance(node_operation, DimensionalityReduction):
-                            result = input_mods[0].dimensionality_reduction(
-                                node_operation
-                            )
-                        elif isinstance(node_operation, AggregatedRepresentation):
-                            result = node_operation.transform(input_mods[0])
-                        elif isinstance(node_operation, UnimodalRepresentation):
-                            if rep_cache is not None:
-                                result = rep_cache[node_operation.name]
-                            else:
-                                agg = pushdown_aggregation_for_node(node.parameters)
-                                result = input_mods[0].apply_representation(
-                                    node_operation, aggregation=agg
-                                )
-                    else:
-                        # It's a fusion operation
-                        fusion_op = node_operation
-                        if (
-                            hasattr(fusion_op, "needs_training")
-                            and fusion_op.needs_training
-                        ):
-                            result = input_mods[0].combine_with_training(
-                                input_mods[1:], fusion_op, task
-                            )
-                        else:
-                            result = input_mods[0].combine(input_mods[1:], fusion_op)
-            else:
+            def run_operation():
                 node_operation = node.operation(params=node.parameters)
                 if gpu_id is not None and hasattr(node_operation, "gpu_id"):
                     node_operation.gpu_id = gpu_id
@@ -419,15 +404,23 @@ def execute_node(node_id: str, task) -> TransformedModality:
                             input_mods[1:], fusion_op, task
                         )
                     else:
-                        result = input_mods[0].combine(input_mods[1:], fusion_op)
-
-                if (
-                    enable_cache
-                    and external_cache is not None
-                    and is_unimodal
-                    and consumer_count[node_id] > 1
-                ):
-                    external_cache.put(node_id, result)
+                        return input_mods[0].combine(input_mods[1:], fusion_op)
+                return result
+
+            if execution_cache is not None and is_unimodal:
+                result = execution_cache.get_or_compute_representation(
+                    node.operation, execution_key(node_id), run_operation
+                )
+            else:
+                result = run_operation()
+
+            if (
+                enable_cache
+                and external_cache is not None
+                and is_unimodal
+                and (consumer_count is None or consumer_count.get(node_id, 0) > 1)
+            ):
+                external_cache.put(node_id, result)
             return result
 
         result = execute_node(self.root_node_id, task)
diff --git a/src/main/python/systemds/scuro/representations/bert.py b/src/main/python/systemds/scuro/representations/bert.py
index 394da666ccd..d5b851dd2a1 100644
--- a/src/main/python/systemds/scuro/representations/bert.py
+++ b/src/main/python/systemds/scuro/representations/bert.py
@@ -37,7 +37,10 @@
     transformer_inference_context,
 )
 from systemds.scuro.modality.type import ModalityType
-from systemds.scuro.drsearch.operator_registry import register_representation
+from systemds.scuro.drsearch.operator_registry import (
+    register_representation,
+    register_expensive_representation,
+)
 from systemds.scuro.utils.memory_utility import (
     get_device,
 )
@@ -66,7 +69,7 @@ def __init__(
     ):
         parameters = {
             **(parameters or {}),
-            "batch_size": [1, 2, 4, 8, 16, 32, 64, 128],
+            # "batch_size": [1, 2, 4, 8, 16, 32, 64, 128],
         }
         self.model_name = model_name
         super().__init__(representation_name, ModalityType.EMBEDDING, parameters)
@@ -88,7 +91,7 @@ def __init__(
         self._activation_hook = None
         if params is not None:
             self.layer = params.get("layer", self.layer)
-            self.batch_size = int(params.get("batch_size", self.batch_size))
+            self.batch_size = int((params or {}).get("batch_size", batch_size))
 
     @property
     def gpu_id(self):
@@ -104,7 +107,7 @@ def set_parameters(
     ):
         if params is not None:
             self.max_seq_length = int(params.get("max_seq_length", max_seq_length))
-            self.batch_size = int(params.get("batch_size", batch_size))
+            self.batch_size = int((params or {}).get("batch_size", batch_size))
             self.layer = params.get("layer", layer)
             self.output_file = params.get("output_file", output_file)
         else:
@@ -349,6 +352,7 @@ def collate(samples):
 
 
 @register_representation(ModalityType.TEXT)
+@register_expensive_representation(ModalityType.TEXT)
 class Bert(BertFamily):
     def __init__(
         self,
@@ -393,6 +397,7 @@ def __init__(
 
 
 @register_representation(ModalityType.TEXT)
+@register_expensive_representation(ModalityType.TEXT)
 class RoBERTa(BertFamily):
     def __init__(
         self,
diff --git a/src/main/python/systemds/scuro/representations/clip.py b/src/main/python/systemds/scuro/representations/clip.py
index adffc0f39cc..a64354b1ace 100644
--- a/src/main/python/systemds/scuro/representations/clip.py
+++ b/src/main/python/systemds/scuro/representations/clip.py
@@ -39,10 +39,12 @@
     transformer_inference_context,
 )
 from systemds.scuro.modality.type import ModalityType
-from systemds.scuro.drsearch.operator_registry import register_representation
+from systemds.scuro.drsearch.operator_registry import (
+    register_representation,
+    register_expensive_representation,
+)
 from transformers import CLIPProcessor, CLIPModel
 
-from systemds.scuro.utils.converter import numpy_dtype_to_torch_dtype
 from systemds.scuro.utils.static_variables import get_device
 from systemds.scuro.utils.torch_dataset import (
     CustomDataset,
@@ -59,6 +61,7 @@
 
 
 @register_representation([ModalityType.VIDEO, ModalityType.IMAGE])
+@register_expensive_representation([ModalityType.VIDEO, ModalityType.IMAGE])
 class CLIPVisual(UnimodalRepresentation):
     supports_aggregation_pushdown = True
     cache_in_worker = True
@@ -71,7 +74,7 @@ def __init__(self, output_file=None, batch_size=32, layer_name="", params=None):
         self.model = CLIPModel.from_pretrained("openai/clip-vit-base-patch32")
         self.processor = CLIPProcessor.from_pretrained("openai/clip-vit-base-patch32")
         if params is not None:
-            self.batch_size = int(params.get("batch_size", batch_size))
+            self.batch_size = int((params or {}).get("batch_size", batch_size))
             self.layer_name = params.get("layer_name", layer_name)
         else:
             self.batch_size = batch_size
@@ -93,7 +96,7 @@ def gpu_id(self, gpu_id):
 
     def _get_parameters(self):
         parameters = {
-            "batch_size": [1, 2, 4, 8, 16, 32, 64, 128],
+            # "batch_size": [1, 2, 4, 8, 16, 32, 64, 128],
             "layer_name": [
                 "",
                 "encoder.layers.0.layer_norm2",
@@ -339,13 +342,14 @@ def _pool_visual_output(self, output: torch.Tensor) -> torch.Tensor:
 
 
 @register_representation(ModalityType.TEXT)
+@register_expensive_representation(ModalityType.TEXT)
 class CLIPText(UnimodalRepresentation):
     supports_aggregation_pushdown = True
     cache_in_worker = True
 
     def __init__(self, output_file=None, batch_size=32, layer_name="", params=None):
         if params is not None:
-            self.batch_size = int(params.get("batch_size", batch_size))
+            self.batch_size = int((params or {}).get("batch_size", batch_size))
             self.layer_name = params.get("layer_name", layer_name)
         else:
             self.batch_size = batch_size
diff --git a/src/main/python/systemds/scuro/representations/color_histogram.py b/src/main/python/systemds/scuro/representations/color_histogram.py
index c159b285893..34147452791 100644
--- a/src/main/python/systemds/scuro/representations/color_histogram.py
+++ b/src/main/python/systemds/scuro/representations/color_histogram.py
@@ -66,7 +66,7 @@ def _get_parameters(self):
             "color_space": ["RGB", "HSV", "GRAY"],
             "bins": [8, 16, 32, 64, 128, 256],
             "normalize": [True, False],
-            "aggregation": ["mean", "max", "concat"],
+            "aggregation": ["mean", "max"],
         }
 
     def compute_histogram(self, image):
diff --git a/src/main/python/systemds/scuro/representations/glove.py b/src/main/python/systemds/scuro/representations/glove.py
index acf7aff522b..9513760fb9f 100644
--- a/src/main/python/systemds/scuro/representations/glove.py
+++ b/src/main/python/systemds/scuro/representations/glove.py
@@ -29,7 +29,10 @@
 from systemds.scuro.representations.unimodal import UnimodalRepresentation
 from systemds.scuro.representations.utils import save_embeddings
 from systemds.scuro.modality.type import ModalityType
-from systemds.scuro.drsearch.operator_registry import register_representation
+from systemds.scuro.drsearch.operator_registry import (
+    register_representation,
+    register_expensive_representation,
+)
 
 
 def load_glove_embeddings(file_path):
@@ -44,6 +47,7 @@ def load_glove_embeddings(file_path):
 
 
 @register_representation(ModalityType.TEXT)
+@register_expensive_representation(ModalityType.TEXT)
 class GloVe(UnimodalRepresentation):
     def __init__(self, output_file=None, params=None):
         super().__init__("GloVe", ModalityType.TEXT)
diff --git a/src/main/python/systemds/scuro/representations/image_bind.py b/src/main/python/systemds/scuro/representations/image_bind.py
index 0cd30af1a70..0dc448a46fb 100644
--- a/src/main/python/systemds/scuro/representations/image_bind.py
+++ b/src/main/python/systemds/scuro/representations/image_bind.py
@@ -27,7 +27,10 @@
 import torchaudio
 from torchvision import transforms
 
-from systemds.scuro.drsearch.operator_registry import register_representation
+from systemds.scuro.drsearch.operator_registry import (
+    register_representation,
+    register_expensive_representation,
+)
 from systemds.scuro.modality.transformed import TransformedModality
 from systemds.scuro.modality.type import ModalityType
 from systemds.scuro.representations.representation import RepresentationStats
@@ -43,6 +46,9 @@
 
 
 @register_representation([ModalityType.VIDEO, ModalityType.AUDIO, ModalityType.TEXT])
+@register_expensive_representation(
+    [ModalityType.VIDEO, ModalityType.AUDIO, ModalityType.TEXT]
+)
 class ImageBind(UnimodalRepresentation):
     _EMBEDDING_DIM = 1024
     _MODEL_PARAMETER_COUNT = 1_200_000_000
@@ -54,13 +60,14 @@ class ImageBind(UnimodalRepresentation):
     cache_in_worker = True
 
     def __init__(self, output_file=None, batch_size=8, params=None):
-        parameters = {"batch_size": [1, 2, 4, 8, 16, 32]}
+        # parameters = {"batch_size": [1, 2, 4, 8, 16, 32]}
+        parameters = {}
         super().__init__("ImageBind", ModalityType.EMBEDDING, parameters)
         self.params = params
         self.output_file = output_file
         self.batch_size = batch_size
         if params is not None:
-            self.batch_size = int(params.get("batch_size", batch_size))
+            batch_size = int((params or {}).get("batch_size", batch_size))
             self.output_file = params.get("output_file", output_file)
         self.data_type = torch.float32
         self.model = None
diff --git a/src/main/python/systemds/scuro/representations/lstm.py b/src/main/python/systemds/scuro/representations/lstm.py
index 08d5bc8af78..385eeac4202 100644
--- a/src/main/python/systemds/scuro/representations/lstm.py
+++ b/src/main/python/systemds/scuro/representations/lstm.py
@@ -52,7 +52,7 @@ def __init__(
             "dropout_rate": [0.1, 0.2, 0.3, 0.4, 0.5],
             "learning_rate": [0.001, 0.0001, 0.01, 0.1],
             "epochs": [10, 20, 50, 100, 200],
-            "batch_size": [8, 16, 32, 64, 128],
+            # "batch_size": [8, 16, 32, 64, 128],
         }
 
         super().__init__("LSTM", parameters)
@@ -63,7 +63,7 @@ def __init__(
             dropout_rate = params.get("dropout_rate", dropout_rate)
             learning_rate = params.get("learning_rate", learning_rate)
             epochs = params.get("epochs", epochs)
-            batch_size = params.get("batch_size", batch_size)
+            batch_size = int((params or {}).get("batch_size", batch_size))
 
         self.width = int(width)
         self.depth = int(depth)
diff --git a/src/main/python/systemds/scuro/representations/mlp_averaging.py b/src/main/python/systemds/scuro/representations/mlp_averaging.py
index 1150a149ccd..1e330b0eb25 100644
--- a/src/main/python/systemds/scuro/representations/mlp_averaging.py
+++ b/src/main/python/systemds/scuro/representations/mlp_averaging.py
@@ -51,12 +51,12 @@ class MLPAveraging(DimensionalityReduction):
     def __init__(self, output_dim=512, batch_size=32, params=None):
         parameters = {
             "output_dim": [64, 128, 256, 512, 1024, 2048, 4096],
-            "batch_size": [8, 16, 32, 64, 128],
+            # "batch_size": [8, 16, 32, 64, 128],
         }
         super().__init__("MLPAveraging", parameters)
         if params is not None:
             output_dim = params.get("output_dim", output_dim)
-            batch_size = params.get("batch_size", batch_size)
+            batch_size = int((params or {}).get("batch_size", batch_size))
         self.output_dim = output_dim
         self.batch_size = batch_size
         self.device = None
diff --git a/src/main/python/systemds/scuro/representations/mlp_learned_dim_reduction.py b/src/main/python/systemds/scuro/representations/mlp_learned_dim_reduction.py
index 0cdc2b8fad8..bf7951f6b24 100644
--- a/src/main/python/systemds/scuro/representations/mlp_learned_dim_reduction.py
+++ b/src/main/python/systemds/scuro/representations/mlp_learned_dim_reduction.py
@@ -53,7 +53,7 @@ class MLPLearnedDimReduction(DimensionalityReduction):
     def __init__(self, output_dim=256, batch_size=32, learning_rate=0.001, epochs=5):
         parameters = {
             "output_dim": [64, 128, 256, 512, 1024],
-            "batch_size": [8, 16, 32, 64, 128],
+            # "batch_size": [8, 16, 32, 64, 128],
             "learning_rate": [0.001, 0.0001, 0.01, 0.1],
             "epochs": [5, 10, 20, 50, 100],
         }
diff --git a/src/main/python/systemds/scuro/representations/multimodal_attention_fusion.py b/src/main/python/systemds/scuro/representations/multimodal_attention_fusion.py
index af5a83c7e50..5363a0bdfbc 100644
--- a/src/main/python/systemds/scuro/representations/multimodal_attention_fusion.py
+++ b/src/main/python/systemds/scuro/representations/multimodal_attention_fusion.py
@@ -48,7 +48,7 @@ def __init__(
             "hidden_dim": [32, 128, 256, 384, 512, 768],
             "num_heads": [2, 4, 8, 12],
             "dropout": [0.0, 0.1, 0.2, 0.3, 0.4],
-            "batch_size": [8, 16, 32, 64, 128],
+            # "batch_size": [8, 16, 32, 64, 128],
             "num_epochs": [10, 20, 50, 100, 150, 200],
             "learning_rate": [1e-5, 1e-4, 1e-3, 1e-2],
         }
@@ -64,7 +64,7 @@ def __init__(
             self.hidden_dim = int(params.get("hidden_dim", self.hidden_dim))
             self.num_heads = int(params.get("num_heads", self.num_heads))
             self.dropout = float(params.get("dropout", self.dropout))
-            self.batch_size = int(params.get("batch_size", self.batch_size))
+            self.batch_size = int((params or {}).get("batch_size", batch_size))
             self.num_epochs = int(params.get("num_epochs", self.num_epochs))
             self.learning_rate = float(params.get("learning_rate", self.learning_rate))
 
diff --git a/src/main/python/systemds/scuro/representations/openface.py b/src/main/python/systemds/scuro/representations/openface.py
index 7715e85951f..67c26ba4a97 100644
--- a/src/main/python/systemds/scuro/representations/openface.py
+++ b/src/main/python/systemds/scuro/representations/openface.py
@@ -29,7 +29,10 @@
 import torch
 
 from systemds.scuro.dataloader.video_loader import VideoStats
-from systemds.scuro.drsearch.operator_registry import register_representation
+from systemds.scuro.drsearch.operator_registry import (
+    register_representation,
+    register_expensive_representation,
+)
 from systemds.scuro.modality.transformed import TransformedModality
 from systemds.scuro.modality.type import ModalityType
 from systemds.scuro.representations.representation import (
@@ -82,6 +85,7 @@ def patched_init(self, args):
 
 
 @register_representation([ModalityType.IMAGE, ModalityType.VIDEO])
+@register_expensive_representation([ModalityType.IMAGE, ModalityType.VIDEO])
 class OpenFace(UnimodalRepresentation):
     supports_aggregation_pushdown = True
     cache_in_worker = True
diff --git a/src/main/python/systemds/scuro/representations/resnet.py b/src/main/python/systemds/scuro/representations/resnet.py
index ccefd4e11b4..00b27a22602 100644
--- a/src/main/python/systemds/scuro/representations/resnet.py
+++ b/src/main/python/systemds/scuro/representations/resnet.py
@@ -33,7 +33,10 @@
 from systemds.scuro.modality.transformed import TransformedModality
 from systemds.scuro.representations.unimodal import UnimodalRepresentation
 from typing import Tuple, Any
-from systemds.scuro.drsearch.operator_registry import register_representation
+from systemds.scuro.drsearch.operator_registry import (
+    register_representation,
+    register_expensive_representation,
+)
 import torch.utils.data
 import torch
 import torchvision.models as models
@@ -48,6 +51,7 @@ def forward(self, input_: torch.Tensor) -> torch.Tensor:
 
 
 @register_representation([ModalityType.IMAGE, ModalityType.VIDEO])
+@register_expensive_representation([ModalityType.IMAGE, ModalityType.VIDEO])
 class ResNet(UnimodalRepresentation):
     supports_aggregation_pushdown = True
     cache_in_worker = True
@@ -67,7 +71,7 @@ def __init__(
         self.gpu_id = None
         self.device = get_device()
         if params is not None:
-            self.batch_size = int(params.get("batch_size", batch_size))
+            self.batch_size = int((params or {}).get("batch_size", batch_size))
             self.layer_name = params.get("layer_name", layer_name)
             model_name = params.get("model_name", model_name)
         else:
@@ -192,7 +196,7 @@ def estimate_peak_memory_bytes(self, input_stats: ImageStats) -> dict:
 
     def _get_parameters(self, high_level=True):
         parameters = {
-            "batch_size": [1, 2, 4, 8, 16, 32, 64, 128],
+            # "batch_size": [1, 2, 4, 8, 16, 32, 64, 128],
             "model_name": [],
             "layer_name": [],
         }
diff --git a/src/main/python/systemds/scuro/representations/swin_video_transformer.py b/src/main/python/systemds/scuro/representations/swin_video_transformer.py
index f57ac18f037..818ce94134d 100644
--- a/src/main/python/systemds/scuro/representations/swin_video_transformer.py
+++ b/src/main/python/systemds/scuro/representations/swin_video_transformer.py
@@ -29,7 +29,10 @@
 import torchvision.models as models
 import numpy as np
 from systemds.scuro.modality.type import ModalityType
-from systemds.scuro.drsearch.operator_registry import register_representation
+from systemds.scuro.drsearch.operator_registry import (
+    register_representation,
+    register_expensive_representation,
+)
 from systemds.scuro.dataloader.video_loader import VideoStats
 from systemds.scuro.representations.utils import (
     LengthBucketBatchSampler,
@@ -48,6 +51,7 @@
 
 
 @register_representation([ModalityType.VIDEO])
+@register_expensive_representation([ModalityType.VIDEO])
 class SwinVideoTransformer(UnimodalRepresentation):
     _EMBED_DIM = 768
     cache_in_worker = True
@@ -64,7 +68,7 @@ def __init__(self, layer_name="avgpool", batch_size=8, params=None):
                 "features.6",
                 "avgpool",
             ],
-            "batch_size": [1, 2, 4, 8, 16, 32],
+            # "batch_size": [1, 2, 4, 8, 16, 32],
         }
         self.data_type = torch.float32
         super().__init__("SwinVideoTransformer", ModalityType.EMBEDDING, parameters)
diff --git a/src/main/python/systemds/scuro/representations/vgg.py b/src/main/python/systemds/scuro/representations/vgg.py
index d73753b2288..6f82937952a 100644
--- a/src/main/python/systemds/scuro/representations/vgg.py
+++ b/src/main/python/systemds/scuro/representations/vgg.py
@@ -24,7 +24,10 @@
 from systemds.scuro.dataloader.video_loader import VideoStats
 from systemds.scuro.representations.unimodal import UnimodalRepresentation
 from typing import Tuple, Any
-from systemds.scuro.drsearch.operator_registry import register_representation
+from systemds.scuro.drsearch.operator_registry import (
+    register_representation,
+    register_expensive_representation,
+)
 import torch.utils.data
 import torch
 import torchvision.models as models
@@ -51,6 +54,7 @@ def forward(self, input_: torch.Tensor) -> torch.Tensor:
 
 
 @register_representation([ModalityType.IMAGE, ModalityType.VIDEO])
+@register_expensive_representation([ModalityType.IMAGE, ModalityType.VIDEO])
 class VGG19(UnimodalRepresentation):
     supports_aggregation_pushdown = True
     cache_in_worker = True
@@ -70,7 +74,7 @@ def __init__(
         super().__init__("VGG19", ModalityType.EMBEDDING, parameters)
         self.params = params
         if params is not None:
-            batch_size = int(params.get("batch_size", batch_size))
+            batch_size = int((params or {}).get("batch_size", batch_size))
             layer = params.get("layer_name", layer)
         self.output_file = output_file
         self.layer_name = layer
@@ -95,7 +99,7 @@ def gpu_id(self, gpu_id):
 
     def _get_parameters(self):
         parameters = {
-            "batch_size": [1, 2, 4, 8, 16, 32, 64, 128],
+            # "batch_size": [1, 2, 4, 8, 16, 32, 64, 128],
             "layer_name": [
                 "features.35",
                 "classifier.0",
diff --git a/src/main/python/systemds/scuro/representations/wav2vec.py b/src/main/python/systemds/scuro/representations/wav2vec.py
index d50465ca5c0..4fed27da24a 100644
--- a/src/main/python/systemds/scuro/representations/wav2vec.py
+++ b/src/main/python/systemds/scuro/representations/wav2vec.py
@@ -27,7 +27,10 @@
 
 from systemds.scuro.representations.representation import RepresentationStats
 from systemds.scuro.representations.unimodal import UnimodalRepresentation
-from systemds.scuro.drsearch.operator_registry import register_representation
+from systemds.scuro.drsearch.operator_registry import (
+    register_representation,
+    register_expensive_representation,
+)
 from systemds.scuro.utils.memory_utility import get_device
 from systemds.scuro.representations.utils import (
     LengthBucketBatchSampler,
@@ -46,6 +49,7 @@
 
 
 @register_representation(ModalityType.AUDIO)
+@register_expensive_representation(ModalityType.AUDIO)
 class Wav2Vec(UnimodalRepresentation):
     cache_in_worker = True
     instance_parallel = False
@@ -53,7 +57,8 @@ class Wav2Vec(UnimodalRepresentation):
     MODEL_NAME = "facebook/wav2vec2-base-960h"
 
     def __init__(self, batch_size=8, params=None):
-        parameters = {"batch_size": [1, 2, 4, 8, 16, 32, 64]}
+        # parameters = {"batch_size": [1, 2, 4, 8, 16, 32, 64]}
+        parameters = {}
         super().__init__("Wav2Vec", ModalityType.TIMESERIES, parameters)
         self.batch_size = int((params or {}).get("batch_size", batch_size))
         self._processor = None
diff --git a/src/main/python/systemds/scuro/representations/word2vec.py b/src/main/python/systemds/scuro/representations/word2vec.py
index 0954a17e290..73368fe2bfe 100644
--- a/src/main/python/systemds/scuro/representations/word2vec.py
+++ b/src/main/python/systemds/scuro/representations/word2vec.py
@@ -28,7 +28,10 @@
 from gensim.utils import tokenize
 
 from systemds.scuro.modality.type import ModalityType
-from systemds.scuro.drsearch.operator_registry import register_representation
+from systemds.scuro.drsearch.operator_registry import (
+    register_representation,
+    register_expensive_representation,
+)
 import nltk
 
 
@@ -42,6 +45,7 @@ def get_embedding(sentence, model):
 
 
 @register_representation(ModalityType.TEXT)
+@register_expensive_representation(ModalityType.TEXT)
 class W2V(UnimodalRepresentation):
     def __init__(self, vector_size=128, min_count=1, output_file=None, params=None):
         parameters = {
diff --git a/src/main/python/systemds/scuro/representations/x3d.py b/src/main/python/systemds/scuro/representations/x3d.py
index fae60d1f1cd..ec41a59ea46 100644
--- a/src/main/python/systemds/scuro/representations/x3d.py
+++ b/src/main/python/systemds/scuro/representations/x3d.py
@@ -28,7 +28,10 @@
 from torchvision.models.video import r3d_18, s3d
 
 from systemds.scuro.dataloader.video_loader import VideoStats
-from systemds.scuro.drsearch.operator_registry import register_representation
+from systemds.scuro.drsearch.operator_registry import (
+    register_representation,
+    register_expensive_representation,
+)
 from systemds.scuro.modality.transformed import TransformedModality
 from systemds.scuro.modality.type import ModalityType
 from systemds.scuro.representations.representation import RepresentationStats
@@ -54,6 +57,7 @@ def forward(self, input_: torch.Tensor) -> torch.Tensor:
 
 
 @register_representation([ModalityType.VIDEO])
+@register_expensive_representation([ModalityType.VIDEO])
 class X3D(UnimodalRepresentation):
     cache_in_worker = True
 
@@ -167,7 +171,7 @@ def model_name(self, model_name):
 
     def _get_parameters(self, high_level=True):
         parameters = {
-            "batch_size": [1, 2, 4, 8, 16, 32],
+            "batch_size": [8],
             "model_name": [],
             "layer_name": [],
         }
@@ -338,7 +342,7 @@ def estimate_output_memory_bytes(self, input_stats: VideoStats) -> int:
 
     def _get_parameters(self, high_level=True):
         parameters = {
-            "batch_size": [1, 2, 4, 8, 16, 32],
+            "batch_size": [8],
             "layer_name": [],
         }
 
diff --git a/src/main/python/tests/scuro/test_hp_tuner.py b/src/main/python/tests/scuro/test_hp_tuner.py
index 03ffc1c2dad..8f7aa0b1284 100644
--- a/src/main/python/tests/scuro/test_hp_tuner.py
+++ b/src/main/python/tests/scuro/test_hp_tuner.py
@@ -134,9 +134,7 @@ def run_hp_for_modality(
                 ModalityType.IMAGE: [ResNet, ColorHistogram],
                 ModalityType.EMBEDDING: [],
             },
-        ):
-            registry = Registry()
-            registry._fusion_operators = [LSTM]
+        ), patch.object(Registry, "_fusion_operators", [LSTM]):
             unimodal_optimizer = UnimodalOptimizer(
                 modalities, self.tasks, False, k=2, max_num_workers=1
             )
diff --git a/src/main/python/tests/scuro/test_hpo_cache.py b/src/main/python/tests/scuro/test_hpo_cache.py
new file mode 100644
index 00000000000..67240adf4f0
--- /dev/null
+++ b/src/main/python/tests/scuro/test_hpo_cache.py
@@ -0,0 +1,154 @@
+# -------------------------------------------------------------
+#
+# Licensed to the Apache Software Foundation (ASF) under one
+# or more contributor license agreements.  See the NOTICE file
+# distributed with this work for additional information
+# regarding copyright ownership.  The ASF licenses this file
+# to you under the Apache License, Version 2.0 (the
+# "License"); you may not use this file except in compliance
+# with the License.  You may obtain a copy of the License at
+#
+#   http://www.apache.org/licenses/LICENSE-2.0
+#
+# Unless required by applicable law or agreed to in writing,
+# software distributed under the License is distributed on an
+# "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+# KIND, either express or implied.  See the License for the
+# specific language governing permissions and limitations
+# under the License.
+#
+# -------------------------------------------------------------
+import unittest
+from types import SimpleNamespace
+
+import numpy as np
+from systemds.scuro.drsearch.hyperparameter_tuner import (
+    HyperparameterTuner,
+)
+from systemds.scuro.drsearch.operator_registry import (
+    is_expensive_representation,
+    register_expensive_representation,
+)
+from systemds.scuro.drsearch.process_cache import BoundedProcessCache
+from systemds.scuro.drsearch.representation_dag import RepresentationDAGBuilder
+from systemds.scuro.modality.transformed import TransformedModality
+from systemds.scuro.modality.type import ModalityType
+from systemds.scuro.modality.unimodal_modality import UnimodalModality
+from systemds.scuro.representations.bert import Bert
+from systemds.scuro.representations.representation import RepresentationStats
+from systemds.scuro.representations.unimodal import UnimodalRepresentation
+from tests.scuro.data_generator import TestDataLoader
+
+
+@register_expensive_representation(ModalityType.TEXT)
+class CountingEncoder(UnimodalRepresentation):
+    calls = 0
+
+    def __init__(self, params=None):
+        super().__init__("CountingEncoder", ModalityType.EMBEDDING)
+
+    def transform(self, modality, aggregation=None):
+        type(self).calls += 1
+        result = TransformedModality(modality, self, ModalityType.EMBEDDING)
+        result.data = np.asarray(
+            [[len(text)] for text in modality.data], dtype=np.float32
+        )
+        return result
+
+    def get_output_stats(self, input_stats):
+        return RepresentationStats(input_stats.num_instances, (1,))
+
+    def estimate_peak_memory_bytes(self, input_stats):
+        return {"cpu_peak_bytes": 1024, "gpu_peak_bytes": 0}
+
+
+class ScalingRepresentation(UnimodalRepresentation):
+    calls = 0
+
+    def __init__(self, scale=1, params=None):
+        super().__init__(
+            "ScalingRepresentation", ModalityType.EMBEDDING, {"scale": [1, 2, 3]}
+        )
+        if params is not None:
+            scale = params.get("scale", scale)
+        self.scale = scale
+
+    def transform(self, modality, aggregation=None):
+        type(self).calls += 1
+        result = TransformedModality(modality, self, ModalityType.EMBEDDING)
+        result.data = np.asarray(modality.data) * self.scale
+        return result
+
+    def get_output_stats(self, input_stats):
+        return RepresentationStats(input_stats.num_instances, input_stats.output_shape)
+
+    def estimate_peak_memory_bytes(self, input_stats):
+        return {"cpu_peak_bytes": 1024, "gpu_peak_bytes": 0}
+
+
+class _OptimizationResults:
+    def get_k_best_results(self, modality, task, scoring_metric, cache_needed=False):
+        return [], []
+
+
+class TestHPOCache(unittest.TestCase):
+    def setUp(self):
+        CountingEncoder.calls = 0
+        ScalingRepresentation.calls = 0
+
+    def test_bounded_cache_evicts_least_recently_used_entry(self):
+        cache = BoundedProcessCache(2)
+        cache.put("first", 1)
+        cache.put("second", 2)
+        self.assertEqual(cache.get("first"), 1)
+        cache.put("third", 3)
+
+        self.assertIsNone(cache.get("second"))
+        self.assertEqual(cache.get("first"), 1)
+        self.assertEqual(cache.get("third"), 3)
+
+    def test_default_policy_only_selects_expensive_representations(self):
+        self.assertTrue(is_expensive_representation(Bert))
+        self.assertFalse(is_expensive_representation(ScalingRepresentation))
+
+    def test_hpo_reuses_unchanged_dag_prefix(self):
+        data = ["one", "three"]
+        metadata = [ModalityType.TEXT.create_metadata(len(text), text) for text in data]
+        modality = UnimodalModality(
+            TestDataLoader(
+                np.arange(len(data)), None, ModalityType.TEXT, data, str, metadata
+            )
+        )
+        builder = RepresentationDAGBuilder()
+        leaf_id = builder.create_leaf_node(modality.modality_id)
+        encoder_id = builder.create_operation_node(CountingEncoder, [leaf_id])
+        scaling_id = builder.create_operation_node(
+            ScalingRepresentation, [encoder_id], {"scale": 1}
+        )
+        dag = builder.build(scaling_id)
+        task = SimpleNamespace(
+            model=SimpleNamespace(name="cache_task"),
+            run=lambda data: [float(np.mean(data))] * 3,
+        )
+        tuner = HyperparameterTuner(
+            [modality],
+            [task],
+            _OptimizationResults(),
+            n_jobs=1,
+        )
+
+        for scale in (1, 2, 3, 1):
+            tuner.evaluate_dag_config(
+                dag,
+                {f"{scaling_id}-scale": scale},
+                [encoder_id, scaling_id],
+                [modality.modality_id],
+                task,
+            )
+
+        self.assertEqual(CountingEncoder.calls, 1)
+        self.assertEqual(ScalingRepresentation.calls, 4)
+
+
+if __name__ == "__main__":
+    unittest.main()