+
+ | Model |
+ Mode |
+ Avg. Activation Sparsity |
+ Word Perplexity (↓) |
+
+
+ | meta-llama/Llama-2-7b-hf |
+ fp32 |
+ - |
+ 9.242 |
+
+
+ |
+ sparse_activation |
+ up/gate30% + down50% |
+ 9.508 |
+
+
+ |
+ int8_asym + sparse_activation |
+ up/gate30% + down50% |
+ 9.511 |
+
+
+ | meta-llama/Meta-Llama-3-8B-Instruct |
+ fp32 |
+ - |
+ 10.802 |
+
+
+ |
+ sparse_activation |
+ up/gate30% + down50% |
+ 11.294 |
+
+
+ |
+ int8_asym + sparse_activation |
+ up/gate30% + down50% |
+ 11.302 |
+
+
+ | mistralai/Mixtral-8x7B-Instruct-v0.1 |
+ fp32 |
+ - |
+ 6.224 |
+
+
+ |
+ sparse_activation |
+ up/gate40% + down50% |
+ 6.561 |
+
+
+ |
+ int8_asym + sparse_activation |
+ up/gate40% + down50% |
+ 6.579 |
+
+
+
+#### Known Limitations
+
+1. Currently activation sparsity only supports Torch backend. Consequently, this restricts the available compression modes to 8-bit integer modes when using `nncf.compress_weights()` before activation sparsification. More information on supported modes can be found at [Weights Compression](../../../../docs/usage/post_training_compression/weights_compression/Usage.md#limitations).
+2. Actual activation sparsity during inference is dynamic and per input basis, deviation from the target should be expected. In our local experiments, the statistical mean of actual activation sparsity aligned to the target when thresholds are calibrated on datasets similar to the final task.
+3. Similar to other compression methods, model accuracy and activation sparsity are trade-off at play. For LLMs like [Llama](https://llama.meta.com), it is recommended to start with 30%~50% sparsity for the Linear layers in feed-forward networks.
diff --git a/nncf/experimental/torch/sparsify_activations/__init__.py b/nncf/experimental/torch/sparsify_activations/__init__.py
new file mode 100644
index 00000000000..ecfaa78cc4f
--- /dev/null
+++ b/nncf/experimental/torch/sparsify_activations/__init__.py
@@ -0,0 +1,13 @@
+# Copyright (c) 2024 Intel Corporation
+# Licensed 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.
+
+from nncf.experimental.torch.sparsify_activations.sparsify_activations_impl import sparsify_activations # noqa: F401
+from nncf.experimental.torch.sparsify_activations.target_scope import TargetScope # noqa: F401
diff --git a/nncf/experimental/torch/sparsify_activations/sparsify_activations_impl.py b/nncf/experimental/torch/sparsify_activations/sparsify_activations_impl.py
new file mode 100644
index 00000000000..83a7a418911
--- /dev/null
+++ b/nncf/experimental/torch/sparsify_activations/sparsify_activations_impl.py
@@ -0,0 +1,264 @@
+# Copyright (c) 2024 Intel Corporation
+# Licensed 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.
+
+from abc import ABC
+from abc import abstractmethod
+from typing import Dict, List, Optional, Type, TypeVar
+
+import nncf
+from nncf.common import factory
+from nncf.common.factory import NNCFGraphFactory
+from nncf.common.graph.graph import NNCFGraph
+from nncf.common.graph.graph import NNCFNode
+from nncf.common.graph.operator_metatypes import OperatorMetatype
+from nncf.common.logging.track_progress import track
+from nncf.common.scopes import should_consider_scope
+from nncf.common.utils.backend import BackendType
+from nncf.common.utils.backend import get_backend
+from nncf.data import Dataset
+from nncf.experimental.torch.sparsify_activations.target_scope import TargetScope
+from nncf.experimental.torch.sparsify_activations.target_scope import get_target_node_names_from_target_scope
+from nncf.scopes import IgnoredScope
+from nncf.scopes import get_ignored_node_names_from_ignored_scope
+from nncf.torch.model_creation import is_wrapped_model
+from nncf.torch.model_creation import wrap_model
+
+TModel = TypeVar("TModel")
+
+
+class SparsifyActivationsAlgoBackend(ABC):
+ """
+ Abstract class for activation sparsification algorithm backend.
+ """
+
+ CALIBRATION_TRACKING_DESC = "Conducting Activations Sparsifier Calibration"
+
+ @staticmethod
+ def do_inference(model: TModel, dataset: Dataset):
+ """
+ Conducts model inference on given dataset to calibrate the activation sparsifiers.
+
+ :param model: The model with activation sparsifiers.
+ :param dataset: The calibration dataset to update the sparsifiers.
+ """
+ engine = factory.EngineFactory.create(model)
+ for input_data in track(
+ dataset.get_inference_data(),
+ total=dataset.get_length(),
+ description=SparsifyActivationsAlgoBackend.CALIBRATION_TRACKING_DESC,
+ ):
+ engine.infer(input_data)
+
+ @property
+ @abstractmethod
+ def supported_metatypes(self) -> List[Type[OperatorMetatype]]:
+ """
+ Property for the backend-specific metatypes for supported layers.
+ """
+
+ @abstractmethod
+ def insert_sparsifiers(
+ self,
+ model: TModel,
+ graph: NNCFGraph,
+ target_sparsity_by_node: Dict[NNCFNode, float],
+ ) -> TModel:
+ """
+ Inserts the activation sparsifiers to the model.
+
+ :param model: The model to conduct activation sparsification.
+ :param graph: The model's NNCF graph.
+ :param target_sparsity_by_node: The target sparsity level for the input activation in each given node layer.
+ :return: The model with inserted activation sparsifiers.
+ """
+
+ @abstractmethod
+ def calibrate_sparsifiers(self, model: TModel, graph: NNCFGraph, dataset: Dataset) -> TModel:
+ """
+ Calibrates the thresholds in the activation sparsifiers.
+
+ :param model: The model with inserted activation sparsifiers.
+ :param graph: The model's NNCF graph.
+ :param dataset: The calibration dataset to update the thresholds in the sparsifiers.
+ :return: The model with calibrated activation sparsifiers.
+ """
+
+
+class SparsifyActivationsAlgorithm:
+ """
+ Implementation of activation sparsification algorithm.
+ """
+
+ def __init__(
+ self,
+ target_sparsity_by_scope: Dict[TargetScope, float],
+ ignored_scope: IgnoredScope,
+ ):
+ """
+ :param target_sparsity_by_scope: A dictionary that defines the target sparsity level for specified layers.
+ :param ignored_scope: An ignored scope that defines the list of model control flow
+ graph nodes to be ignored during activation sparsification.
+ """
+ self._target_sparsity_by_scope = target_sparsity_by_scope
+ self._ignored_scope = ignored_scope
+ self._backend_entity: SparsifyActivationsAlgoBackend = None
+
+ @property
+ def available_backends(self) -> List[BackendType]:
+ """
+ Supported backends for this algorithm.
+ """
+ return [BackendType.TORCH]
+
+ def apply(
+ self,
+ model: TModel,
+ graph: NNCFGraph,
+ dataset: Dataset,
+ ) -> TModel:
+ """
+ Applies the algorithm to the given model.
+
+ :param model: The model to be sparsified.
+ :param graph: The model's NNCF graph.
+ :param dataset: The dataset to calibrate the activation sparsifiers.
+ :return: The sparsified model.
+ """
+ self._set_backend_entity(model)
+ target_sparsity_by_node = self._get_target_sparsity_by_node(graph)
+ sparse_model = self.do_sparsification(model, graph, target_sparsity_by_node, dataset)
+ return sparse_model
+
+ def do_sparsification(
+ self,
+ model: TModel,
+ graph: NNCFGraph,
+ target_sparsity_by_node: Dict[NNCFNode, float],
+ dataset: Dataset,
+ ):
+ """
+ Transforms the model into a sparsified one with node-specific target activation sparsity levels.
+
+ :param model: The model to be sparsified.
+ :param graph: The model's NNCF graph.
+ :param target_sparsity_by_node: A dictionary that defines the target sparsity level
+ for specified node layers.
+ :param dataset: The dataset to calibrate the activation sparsifiers.
+ :return: The sparsified model.
+ """
+ model = self._backend_entity.insert_sparsifiers(model, graph, target_sparsity_by_node)
+ model = self._backend_entity.calibrate_sparsifiers(model, graph, dataset)
+ return model
+
+ def _set_backend_entity(self, model: TModel) -> None:
+ """
+ Creates a helper class with a backend-specific logic of the algorithm.
+
+ :param model: Backend-specific input model.
+ """
+ model_backend = get_backend(model)
+ if model_backend == BackendType.TORCH:
+ from nncf.experimental.torch.sparsify_activations.torch_backend import PTSparsifyActivationsAlgoBackend
+
+ self._backend_entity = PTSparsifyActivationsAlgoBackend()
+ else:
+ raise nncf.UnsupportedBackendError(
+ f"{model_backend.value} backend is not supported for `sparsify_activations`."
+ )
+
+ def _get_target_sparsity_by_node(self, graph: NNCFGraph) -> Dict[NNCFNode, float]:
+ """
+ Collects nodes in the model's graph corresponding to the layers for sparsification.
+
+ :param graph: NNCFGraph instance.
+ :return: A dictionary with nodes and the corresponding target sparsity level.
+ """
+ supported_metatypes = self._backend_entity.supported_metatypes
+ ignored_names = get_ignored_node_names_from_ignored_scope(
+ self._ignored_scope, graph, strict=self._ignored_scope.validate
+ )
+ target_sparsity_by_node = {}
+ for scope, target_sparsity in self._target_sparsity_by_scope.items():
+ target_names = get_target_node_names_from_target_scope(scope, graph, strict=scope.validate)
+ for node_name in target_names:
+ node = graph.get_node_by_name(node_name)
+ if node.metatype not in supported_metatypes or not should_consider_scope(
+ node.node_name, ignored_scopes=ignored_names
+ ):
+ continue
+ if node in target_sparsity_by_node:
+ raise nncf.ValidationError(
+ f'"{node.node_name}" is matched by multiple items in `target_sparsity_by_scope`.'
+ )
+ target_sparsity_by_node[node] = target_sparsity
+ if not target_sparsity_by_node:
+ raise nncf.ValidationError("No layers to conduct activation sparsification.")
+ return target_sparsity_by_node
+
+
+def sparsify_activations(
+ model: TModel,
+ dataset: Dataset,
+ target_sparsity_by_scope: Dict[TargetScope, float],
+ ignored_scope: Optional[IgnoredScope] = None,
+) -> TModel:
+ """
+ Post-training activation sparsification on the given model.
+
+ This algorithm sparsifies the input activations in supported layers based on a calibration
+ dataset. The goal is to zero out neurons with small activation values around 0, thereby
+ roughly achieving the target sparsity at a statistical level.
+
+ Note that currently only linear layers are supported.
+
+ :param model: The model to be sparsified.
+ :param dataset: The dataset to calibrate the activation sparsifiers.
+ :param target_sparsity_by_scope: Defines the target activation sparsity level
+ for specified layers. For each item, the key is an instance of `TargetScope` class
+ representing the layers to match in the model's NNCF graph; the corresponding value
+ is a float number in the range [0, 1] representing the target sparsity level.
+
+ Example:
+ .. code-block:: python
+ {
+ # Target sparsity is 60% for node "Dummy/Linear[layer]/linear_0" in the model graph
+ TargetScope(names=["Dummy/Linear[layer]/linear_0"]): 0.6,
+ # Target sparsity is 30% for the layers whose name contains "up_proj" or "down_proj".
+ TargetScope(patterns=[".*up_proj.*", ".*down_proj.*"]): 0.3,
+ }
+
+ :param ignored_scope: Optional. It defines the nodes in the model graph that should be
+ ignored during activation sparsification. Note that unsupported layer types are already
+ filtered out internally, so there is no need to mention them in `ignored_scope`.
+ :return: The sparsified model.
+ """
+
+ for scope, target_sparsity in target_sparsity_by_scope.items():
+ if target_sparsity < 0.0 or target_sparsity > 1.0:
+ raise ValueError(f'Target sparsity for scope "{scope}" should be in range [0, 1].')
+
+ if ignored_scope is None:
+ ignored_scope = IgnoredScope()
+
+ backend = get_backend(model)
+ if backend == BackendType.TORCH and not is_wrapped_model(model):
+ example_input = next(iter(dataset.get_inference_data()))
+ model = wrap_model(
+ model,
+ example_input=example_input,
+ trace_parameters=True,
+ )
+
+ algorithm = SparsifyActivationsAlgorithm(target_sparsity_by_scope, ignored_scope)
+
+ graph = NNCFGraphFactory.create(model)
+ sparse_model = algorithm.apply(model, graph, dataset)
+ return sparse_model
diff --git a/nncf/experimental/torch/sparsify_activations/target_scope.py b/nncf/experimental/torch/sparsify_activations/target_scope.py
new file mode 100644
index 00000000000..eb09718224a
--- /dev/null
+++ b/nncf/experimental/torch/sparsify_activations/target_scope.py
@@ -0,0 +1,108 @@
+# Copyright (c) 2024 Intel Corporation
+# Licensed 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.
+
+from dataclasses import dataclass
+from typing import Set
+
+import nncf
+from nncf.common.graph.graph import NNCFGraph
+from nncf.scopes import IgnoredScope
+from nncf.scopes import get_difference_ignored_scope
+from nncf.scopes import get_matched_ignored_scope_info
+
+
+@dataclass
+class TargetScope(IgnoredScope):
+ """
+ Specifies the target portions in a model graph.
+
+ Example:
+
+ .. code-block:: python
+ # Specified by node names:
+ node_names = ['node_1', 'node_2', 'node_3']
+ target_scope = TargetScope(names=node_names)
+
+ # Specified using regular expressions:
+ patterns = ['.*node_\\d']
+ target_scope = TargetScope(patterns=patterns)
+
+ # Specified by operation types, e.g.,
+
+ # OpenVINO opset https://docs.openvino.ai/latest/openvino_docs_ops_opset.html
+ operation_types = ['Multiply', 'GroupConvolution', 'Interpolate']
+ target_scope = TargetScope(types=operation_types)
+
+ # ONNX opset https://github.com/onnx/onnx/blob/main/docs/Operators.md
+ operation_types = ['Mul', 'Conv', 'Resize']
+ target_scope = TargetScope(types=operation_types)
+
+ # Specifies by subgraphs:
+ from nncf import Subgraph
+ target_scope = TargetScope(subgraphs=[
+ Subgraph(inputs=["node_1"], outputs=["node_3"])
+ ])
+
+ **Note:** Operation types must be specified according to the model framework.
+
+ :param names: List of target node names.
+ :param patterns: List of regular expressions that define patterns for names of target nodes.
+ :param types: List of target operation types.
+ :param subgraphs: List of target subgraphs.
+ :param validate: If set to True, then a RuntimeError will be raised if any target scope does not match
+ in the model graph.
+ """
+
+ def __hash__(self) -> int:
+ return hash(
+ (
+ frozenset(self.names),
+ frozenset(self.patterns),
+ frozenset(self.types),
+ frozenset((frozenset(subgraph.inputs), frozenset(subgraph.outputs)) for subgraph in self.subgraphs),
+ self.validate,
+ )
+ )
+
+
+def get_target_node_names_from_target_scope(
+ target_scope: TargetScope, nncf_graph: NNCFGraph, strict: bool = True
+) -> Set[str]:
+ """
+ Returns NNCF node names from the graph that are matched by target scope.
+ If strict is True, raises nncf.ValidationError if no rule is matched.
+
+ :param target_scope: Target scope specifying the matching rules.
+ :param nncf_graph: The graph.
+ :param strict: Whether target_scope must match at least one node or not.
+ :return: NNCF node names from the given graph matched by target scope.
+ """
+ matched_target_scope, matches = get_matched_ignored_scope_info(target_scope, [nncf_graph])
+ if strict:
+ _check_target_scope_strictly_matched(target_scope, matched_target_scope)
+ return set().union(*matches.values())
+
+
+def _check_target_scope_strictly_matched(target_scope: TargetScope, matched_target_scope: TargetScope):
+ """
+ Passes when target_scope and matched_target_scope are equal, otherwise raises ValidationError.
+
+ :param target_scope: The given target scope.
+ :param matched_target_scope: The actual target scope matched in a graph.
+ """
+ unmatched_scope = get_difference_ignored_scope(target_scope, matched_target_scope)
+ error_messages = []
+ for match_type in ("names", "types", "patterns", "subgraphs"):
+ unmatched_rules = getattr(unmatched_scope, match_type)
+ if unmatched_rules:
+ error_messages.append(f"The following {match_type} are not found in the graph: {unmatched_rules}.")
+ if error_messages:
+ raise nncf.ValidationError("\n".join(error_messages))
diff --git a/nncf/experimental/torch/sparsify_activations/torch_backend.py b/nncf/experimental/torch/sparsify_activations/torch_backend.py
new file mode 100644
index 00000000000..a10f12c6518
--- /dev/null
+++ b/nncf/experimental/torch/sparsify_activations/torch_backend.py
@@ -0,0 +1,199 @@
+# Copyright (c) 2024 Intel Corporation
+# Licensed 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.
+
+from typing import Dict, List, Type, TypeVar
+
+import torch
+import torch.nn as nn
+
+import nncf
+from nncf.common.graph.graph import NNCFGraph
+from nncf.common.graph.graph import NNCFNode
+from nncf.common.graph.operator_metatypes import CONST_NOOP_METATYPES
+from nncf.common.graph.operator_metatypes import OperatorMetatype
+from nncf.common.graph.transformations.commands import TargetType
+from nncf.data import Dataset
+from nncf.experimental.torch.sparsify_activations.sparsify_activations_impl import SparsifyActivationsAlgoBackend
+from nncf.tensor.functions.torch_numeric import quantile
+from nncf.torch.graph import operator_metatypes as om
+from nncf.torch.graph.transformations.commands import PTSharedFnInsertionCommand
+from nncf.torch.graph.transformations.commands import PTTargetPoint
+from nncf.torch.graph.transformations.layout import PTTransformationLayout
+from nncf.torch.model_transformer import PTModelTransformer
+from nncf.torch.nncf_network import NNCFNetwork
+from nncf.torch.utils import training_mode_switcher
+
+ACTIVATIONS_SPARSIFIER_PREFIX = "activations_sparsifier"
+TModel = TypeVar("TModel")
+
+
+class ActivationsSparsifier(nn.Module):
+ """
+ Sparsifies input activations by masking out values around zero.
+ """
+
+ def __init__(self, target_sparsity: float, alpha: float = 0.2):
+ """
+ :param target_sparsity: The target activation sparsity level.
+ :param alpha: The exponential moving average decay factor in range (0, 1) for calibrating
+ the threshold. A larger alpha will give more weight to the most recent batches.
+ """
+ super().__init__()
+ self.target_sparsity = target_sparsity
+ if alpha <= 0.0 or alpha >= 1.0:
+ raise ValueError("The decay factor `alpha` should be in range (0, 1).")
+ self.alpha = alpha
+ self.register_buffer("running_threshold", torch.tensor(float("-inf")))
+ self.register_buffer("num_batches_tracked", torch.tensor(0))
+ self.running_threshold: torch.Tensor
+ self.num_batches_tracked: torch.Tensor
+ self._freeze = True
+
+ @staticmethod
+ def calculate_threshold(x: torch.Tensor, target_sparsity: float) -> torch.Tensor:
+ """
+ Calculates the threshold to sparsify the input tensor with target sparsity if locations of
+ `x.abs() <= threshold` are zeroed out.
+
+ :param x: The input tensor.
+ :param target_sparsity: The target sparsity level on the input tensor.
+ :return: The threshold value.
+ """
+ return quantile(x.detach().abs().view(-1), q=target_sparsity, axis=0)
+
+ @property
+ def freeze(self):
+ return self._freeze
+
+ @freeze.setter
+ def freeze(self, value: bool):
+ self._freeze = value
+
+ def forward(self, x: torch.Tensor) -> torch.Tensor:
+ if not self.freeze:
+ threshold = self.calculate_threshold(x, self.target_sparsity)
+ self._update(threshold, dtype=x.dtype)
+ mask = torch.le(x.abs(), self.running_threshold)
+ x = torch.masked_fill(x, mask, 0.0)
+ return x
+
+ def reset_running_stats(self):
+ """
+ Resets the running threshold and the number of tracked batches to the initial stage.
+ """
+ self.running_threshold.fill_(float("-inf"))
+ self.num_batches_tracked.zero_()
+
+ def extra_repr(self) -> str:
+ return f"target_sparsity={self.target_sparsity}"
+
+ def _update(self, threshold: torch.Tensor, dtype: torch.dtype) -> torch.Tensor:
+ """
+ Updates the running threshold by exponential moving average with decaying adjustment.
+ The updating logic is similar to `pandas.DataFrame.ewm(adjust=True)`.
+
+ :param threshold: The threshold value derived from this batch to update the running threshold.
+ :param dtype: Data type of the updated running threshold.
+ :return: The updated running threshold.
+ """
+ if self.num_batches_tracked == 0:
+ running_threshold = threshold
+ else:
+ beta = 1.0 - self.alpha
+ old_running_threshold = self.running_threshold.to(device=threshold.device, dtype=torch.float64)
+ running_threshold = (
+ threshold.to(torch.float64) * self.alpha
+ + old_running_threshold * beta * (1 - beta**self.num_batches_tracked)
+ ) / (1 - beta ** (self.num_batches_tracked + 1))
+ self.running_threshold = running_threshold.type(dtype)
+ self.num_batches_tracked += 1
+ return self.running_threshold
+
+
+class PTSparsifyActivationsAlgoBackend(SparsifyActivationsAlgoBackend):
+ """
+ Torch backend for the activation sparsification algorithm.
+ """
+
+ SUPPORTED_METATYPES = [om.PTLinearMetatype]
+
+ @staticmethod
+ def get_sparsifiers(model: NNCFNetwork) -> List[ActivationsSparsifier]:
+ """
+ Finds all the activation sparsifiers in the model.
+
+ :param model: The model with activation sparsifiers.
+ :return: List of activation sparsifiers.
+ """
+ return [m for m in model.nncf.modules() if isinstance(m, ActivationsSparsifier)]
+
+ @property
+ def supported_metatypes(self) -> List[Type[OperatorMetatype]]:
+ return PTSparsifyActivationsAlgoBackend.SUPPORTED_METATYPES
+
+ def insert_sparsifiers(
+ self,
+ model: NNCFNetwork,
+ graph: NNCFGraph,
+ target_sparsity_by_node: Dict[NNCFNode, float],
+ ) -> NNCFNetwork:
+ transformation_layout = PTTransformationLayout()
+ for node, target_sparsity in target_sparsity_by_node.items():
+ activation_port_id = self._get_activation_port_id(node, graph)
+ sparsifier = ActivationsSparsifier(target_sparsity=target_sparsity)
+ sparsifier_name = f"{ACTIVATIONS_SPARSIFIER_PREFIX}_{node.node_name.replace('.', '_')}"
+ transformation_layout.register(
+ PTSharedFnInsertionCommand(
+ [
+ PTTargetPoint(
+ target_type=TargetType.PRE_LAYER_OPERATION,
+ target_node_name=node.node_name,
+ input_port_id=activation_port_id,
+ )
+ ],
+ sparsifier,
+ sparsifier_name,
+ )
+ )
+
+ transformed_model = PTModelTransformer(model).transform(transformation_layout)
+ return transformed_model
+
+ def calibrate_sparsifiers(self, model: NNCFNetwork, graph: NNCFGraph, dataset: Dataset) -> NNCFNetwork:
+ sparsifiers = self.get_sparsifiers(model)
+ for sparsifier in sparsifiers:
+ sparsifier.reset_running_stats()
+ sparsifier.freeze = False
+ with training_mode_switcher(model, is_training=False):
+ with torch.no_grad():
+ self.do_inference(model, dataset)
+ for sparsifier in sparsifiers:
+ sparsifier.freeze = True
+ return model
+
+ @staticmethod
+ def _get_activation_port_id(node: NNCFNode, graph: NNCFGraph) -> int:
+ """
+ Finds the input activation port id for the node.
+
+ :param node: The node to find its activation port id.
+ :param graph: The NNCF graph containing the node.
+ :return: The activation port id.
+ """
+ activation_ports = []
+ for prev_node in graph.get_previous_nodes(node):
+ edge = graph.get_edge(prev_node, node)
+ if prev_node.metatype in CONST_NOOP_METATYPES or edge.input_port_id in node.metatype.weight_port_ids:
+ continue
+ activation_ports.append(edge.input_port_id)
+ if len(activation_ports) != 1:
+ raise nncf.InternalError(f'Cannot find activation port for node "{node}".')
+ return activation_ports[0]
diff --git a/tests/post_training/experimental/sparsify_activations/model_scope.py b/tests/post_training/experimental/sparsify_activations/model_scope.py
new file mode 100644
index 00000000000..5a89847829d
--- /dev/null
+++ b/tests/post_training/experimental/sparsify_activations/model_scope.py
@@ -0,0 +1,116 @@
+# Copyright (c) 2024 Intel Corporation
+# Licensed 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 copy
+from typing import Dict, List
+
+import nncf
+from nncf.experimental.torch.sparsify_activations import TargetScope
+from nncf.parameters import CompressWeightsMode
+from tests.post_training.experimental.sparsify_activations.pipelines import ImageClassificationTimmSparsifyActivations
+from tests.post_training.experimental.sparsify_activations.pipelines import LMSparsifyActivations
+from tests.post_training.pipelines.base import BackendType
+
+SPARSIFY_ACTIVATIONS_MODELS = [
+ {
+ "reported_name": "tinyllama",
+ "model_id": "tinyllama/tinyllama-1.1b-step-50k-105b",
+ "pipeline_cls": LMSparsifyActivations,
+ "compression_params": {},
+ "backends": [BackendType.FP32],
+ },
+ {
+ "reported_name": "tinyllama_ffn_sparse20",
+ "model_id": "tinyllama/tinyllama-1.1b-step-50k-105b",
+ "pipeline_cls": LMSparsifyActivations,
+ "compression_params": {
+ "compress_weights": None,
+ "sparsify_activations": {
+ "target_sparsity_by_scope": {
+ TargetScope(patterns=[".*up_proj.*", ".*gate_proj.*", ".*down_proj.*"]): 0.2,
+ }
+ },
+ },
+ "backends": [BackendType.TORCH, BackendType.CUDA_TORCH],
+ "batch_size": 8,
+ },
+ {
+ "reported_name": "tinyllama_int8_asym_data_free_ffn_sparse20",
+ "model_id": "tinyllama/tinyllama-1.1b-step-50k-105b",
+ "pipeline_cls": LMSparsifyActivations,
+ "compression_params": {
+ "compress_weights": {
+ "mode": CompressWeightsMode.INT8_ASYM,
+ },
+ "sparsify_activations": {
+ "target_sparsity_by_scope": {
+ TargetScope(patterns=[".*up_proj.*", ".*gate_proj.*", ".*down_proj.*"]): 0.2,
+ }
+ },
+ },
+ "backends": [BackendType.TORCH, BackendType.CUDA_TORCH],
+ "batch_size": 8,
+ },
+ {
+ "reported_name": "timm/deit3_small_patch16_224",
+ "model_id": "deit3_small_patch16_224",
+ "pipeline_cls": ImageClassificationTimmSparsifyActivations,
+ "compression_params": {},
+ "backends": [BackendType.FP32],
+ "batch_size": 128,
+ },
+ {
+ "reported_name": "timm/deit3_small_patch16_224_qkv_sparse20_fc1_sparse20_fc2_sparse30",
+ "model_id": "deit3_small_patch16_224",
+ "pipeline_cls": ImageClassificationTimmSparsifyActivations,
+ "compression_params": {
+ "sparsify_activations": {
+ "target_sparsity_by_scope": {
+ TargetScope(patterns=[".*qkv.*", ".*fc1.*"]): 0.2,
+ TargetScope(patterns=[".*fc2.*"]): 0.3,
+ }
+ },
+ },
+ "backends": [BackendType.TORCH, BackendType.CUDA_TORCH],
+ "batch_size": 128,
+ },
+]
+
+
+def generate_tests_scope(models_list: List[Dict]) -> Dict[str, Dict]:
+ """
+ Generate tests by names "{reported_name}_backend_{backend}"
+ """
+ tests_scope = {}
+ fp32_models = set()
+ for test_model_param in models_list:
+ model_id = test_model_param["model_id"]
+ reported_name = test_model_param["reported_name"]
+
+ for backend in test_model_param["backends"]:
+ model_param = copy.deepcopy(test_model_param)
+ if "is_batch_size_supported" not in model_param: # Set default value of is_batch_size_supported.
+ model_param["is_batch_size_supported"] = True
+ test_case_name = f"{reported_name}_backend_{backend.value}"
+ model_param["backend"] = backend
+ model_param.pop("backends")
+ if backend == BackendType.FP32:
+ if model_id in fp32_models:
+ raise nncf.ValidationError(f"Duplicate test case for {model_id} with FP32 backend")
+ fp32_models.add(model_id)
+ if test_case_name in tests_scope:
+ raise nncf.ValidationError(f"{test_case_name} already in tests_scope")
+ tests_scope[test_case_name] = model_param
+
+ return tests_scope
+
+
+SPARSIFY_ACTIVATIONS_TEST_CASES = generate_tests_scope(SPARSIFY_ACTIVATIONS_MODELS)
diff --git a/tests/post_training/experimental/sparsify_activations/pipelines.py b/tests/post_training/experimental/sparsify_activations/pipelines.py
new file mode 100644
index 00000000000..82da57caa86
--- /dev/null
+++ b/tests/post_training/experimental/sparsify_activations/pipelines.py
@@ -0,0 +1,323 @@
+# Copyright (c) 2024 Intel Corporation
+# Licensed 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.
+
+
+from dataclasses import dataclass
+from dataclasses import field
+from pathlib import Path
+from typing import Dict, List, Optional
+
+import numpy as np
+import openvino as ov
+import torch
+import torch.utils
+import torch.utils.data
+import torchvision
+from datasets import load_dataset
+from optimum.exporters.openvino.convert import export_from_model
+from optimum.intel.openvino import OVModelForCausalLM
+from transformers import AutoModelForCausalLM
+
+import nncf
+from nncf.experimental.torch.sparsify_activations import sparsify_activations
+from nncf.experimental.torch.sparsify_activations.sparsify_activations_impl import SparsifyActivationsAlgoBackend
+from nncf.experimental.torch.sparsify_activations.torch_backend import PTSparsifyActivationsAlgoBackend
+from nncf.torch.quantization.layers import AsymmetricWeightsDecompressor
+from nncf.torch.quantization.layers import SymmetricWeightsDecompressor
+from tests.post_training.pipelines.base import LIMIT_LENGTH_OF_STATUS
+from tests.post_training.pipelines.base import PT_BACKENDS
+from tests.post_training.pipelines.base import BackendType
+from tests.post_training.pipelines.base import NumCompressNodes
+from tests.post_training.pipelines.base import RunInfo
+from tests.post_training.pipelines.image_classification_timm import ImageClassificationTimm
+from tests.post_training.pipelines.lm_weight_compression import LMWeightCompression
+from tests.post_training.pipelines.lm_weight_compression import WCTimeStats
+from tests.torch.experimental.sparsify_activations.helpers import count_sparsifier_patterns_in_ov
+from tests.torch.helpers import set_torch_seed
+
+
+@dataclass
+class SATimeStats(WCTimeStats):
+ """
+ Contains statistics that are parsed from the stdout of Sparsify Activations tests.
+ """
+
+ time_sparsifier_calibration: Optional[str] = None
+ STAT_NAMES = [*WCTimeStats.STAT_NAMES, "Activations Sparsifier calibration time"]
+ VAR_NAMES = [*WCTimeStats.VAR_NAMES, "time_sparsifier_calibration"]
+ REGEX_PREFIX = [*WCTimeStats.REGEX_PREFIX, SparsifyActivationsAlgoBackend.CALIBRATION_TRACKING_DESC]
+
+
+@dataclass
+class SANumCompressNodes(NumCompressNodes):
+ num_sparse_activations: Optional[int] = None
+
+
+@dataclass
+class SARunInfo(RunInfo):
+ num_compress_nodes: SANumCompressNodes = field(default_factory=SANumCompressNodes)
+
+ def get_result_dict(self):
+ return {
+ "Model": self.model,
+ "Backend": self.backend.value if self.backend else None,
+ "Metric name": self.metric_name,
+ "Metric value": self.metric_value,
+ "Metric diff": self.metric_diff,
+ "Num FQ": self.num_compress_nodes.num_fq_nodes,
+ "Num int4": self.num_compress_nodes.num_int4,
+ "Num int8": self.num_compress_nodes.num_int8,
+ "Num sparse activations": self.num_compress_nodes.num_sparse_activations,
+ "RAM MiB": self.format_memory_usage(self.compression_memory_usage),
+ "Compr. time": self.format_time(self.time_compression),
+ **self.stats_from_output.get_stats(),
+ "Total time": self.format_time(self.time_total),
+ "FPS": self.fps,
+ "Status": self.status[:LIMIT_LENGTH_OF_STATUS] if self.status is not None else None,
+ }
+
+
+class SAPipelineMixin:
+ """
+ Common methods in the test pipeline for Sparsify Activations.
+ """
+
+ def __init__(
+ self,
+ reported_name: str,
+ model_id: str,
+ backend: BackendType,
+ compression_params: dict,
+ output_dir: Path,
+ data_dir: Path,
+ reference_data: dict,
+ no_eval: bool,
+ run_benchmark_app: bool,
+ params: dict = None,
+ batch_size: int = 1,
+ ):
+ super().__init__(
+ reported_name=reported_name,
+ model_id=model_id,
+ backend=backend,
+ compression_params=compression_params,
+ output_dir=output_dir,
+ data_dir=data_dir,
+ reference_data=reference_data,
+ no_eval=no_eval,
+ run_benchmark_app=run_benchmark_app,
+ params=params,
+ batch_size=batch_size,
+ )
+ self.run_info = SARunInfo(model=reported_name, backend=backend)
+
+ @staticmethod
+ def count_compressed_nodes_from_ir(model: ov.Model) -> SANumCompressNodes:
+ """
+ Get number of compressed nodes in the compressed IR.
+ """
+ num_fq_nodes = 0
+ num_int8 = 0
+ num_int4 = 0
+ for node in model.get_ops():
+ if node.type_info.name == "FakeQuantize":
+ num_fq_nodes += 1
+ for i in range(node.get_output_size()):
+ if node.get_output_element_type(i).get_type_name() in ["i8", "u8"]:
+ num_int8 += 1
+ if node.get_output_element_type(i).get_type_name() in ["i4", "u4"]:
+ num_int4 += 1
+
+ num_sparse_activations = count_sparsifier_patterns_in_ov(model)
+ return SANumCompressNodes(
+ num_fq_nodes=num_fq_nodes,
+ num_int8=num_int8,
+ num_int4=num_int4,
+ num_sparse_activations=num_sparse_activations,
+ )
+
+ def collect_data_from_stdout(self, stdout: str):
+ stats = SATimeStats()
+ stats.fill(stdout)
+ self.run_info.stats_from_output = stats
+
+ @set_torch_seed(seed=42)
+ @torch.no_grad()
+ def _compress(self):
+ """
+ Actual call of weight compression and/or activation sparsification.
+ """
+ self.compressed_model = self.model
+ if self.compression_params.get("compress_weights", None) is not None:
+ self.compressed_model = nncf.compress_weights(
+ self.compressed_model,
+ dataset=self.calibration_dataset,
+ **self.compression_params["compress_weights"],
+ )
+ if self.compression_params.get("sparsify_activations", None) is not None:
+ self.compressed_model = sparsify_activations(
+ self.compressed_model,
+ dataset=self.calibration_dataset,
+ **self.compression_params["sparsify_activations"],
+ )
+
+ def _validate(self):
+ super()._validate()
+ ref_num_sparse_activations = self.reference_data.get("num_sparse_activations", 0)
+ num_sparse_activations = self.run_info.num_compress_nodes.num_sparse_activations
+ if num_sparse_activations != ref_num_sparse_activations:
+ status_msg = f"Regression: The number of sparse activations is {num_sparse_activations}, \
+ which differs from reference {ref_num_sparse_activations}."
+ raise ValueError(status_msg)
+
+
+class LMSparsifyActivations(SAPipelineMixin, LMWeightCompression):
+ DEFAULT_SUBSET_SIZE = 32
+
+ def prepare_model(self):
+ is_stateful = self.params.get("is_stateful", False)
+
+ if self.backend in PT_BACKENDS:
+ if is_stateful:
+ raise RuntimeError(f"is_stateful={is_stateful} is not supported for PyTorch backend.")
+
+ self.model_hf = AutoModelForCausalLM.from_pretrained(
+ self.model_id,
+ torch_dtype=torch.float32,
+ device_map="cuda" if self.backend == BackendType.CUDA_TORCH else "cpu",
+ attn_implementation="eager",
+ )
+ self.model = self.model_hf
+ elif self.backend in [BackendType.OV, BackendType.FP32]:
+ if is_stateful:
+ self.fp32_model_dir = self.fp32_model_dir.parent / (self.fp32_model_dir.name + "_sf")
+ if not (self.fp32_model_dir / self.OV_MODEL_NAME).exists():
+ # export by model_id
+ self.model_hf = OVModelForCausalLM.from_pretrained(
+ self.model_id,
+ trust_remote_code=True,
+ export=True,
+ load_in_8bit=False,
+ compile=False,
+ stateful=is_stateful,
+ )
+ else:
+ # no export, load from IR. Applicable for sequential run of test cases in local environment.
+ self.model_hf = OVModelForCausalLM.from_pretrained(
+ self.fp32_model_dir, load_in_8bit=False, compile=False, stateful=is_stateful
+ )
+ self.model = self.model_hf.model
+ else:
+ raise RuntimeError(f"backend={self.backend.value} is not supported.")
+
+ if not (self.fp32_model_dir / self.OV_MODEL_NAME).exists():
+ self._dump_model_fp32()
+
+ # Use FP16 for CUDA_TORCH backend as it is more common when running LLM on CUDA.
+ if self.backend == BackendType.CUDA_TORCH:
+ self.model_hf.half()
+
+ def get_transform_calibration_fn(self):
+ process_one = super().get_transform_calibration_fn()
+
+ def transform_fn(chunk: List[Dict]):
+ samples = [process_one(data, max_tokens=128, filter_bad_tokens=False) for data in chunk]
+ inputs = {}
+ for input_name, sample_value in samples[0].items():
+ if isinstance(sample_value, torch.Tensor):
+ inputs[input_name] = torch.cat([sample[input_name] for sample in samples], dim=0)
+ elif isinstance(sample_value, np.ndarray):
+ inputs[input_name] = np.concatenate([sample[input_name] for sample in samples], axis=0)
+ elif isinstance(sample_value, ov.Tensor):
+ shape = sample_value.get_shape()
+ shape[0] = len(samples)
+ inputs[input_name] = ov.Tensor(sample_value.get_element_type(), shape)
+ else:
+ raise RuntimeError(
+ f"Failed to generate calibration set for {input_name} in type {type(sample_value)}"
+ )
+ if self.backend == BackendType.CUDA_TORCH:
+ for input_name in inputs:
+ inputs[input_name] = torch.from_numpy(inputs[input_name]).cuda()
+ return inputs
+
+ return transform_fn
+
+ def prepare_calibration_dataset(self):
+ subset_size = self.compression_params.get("subset_size") or self.DEFAULT_SUBSET_SIZE
+ dataset = (
+ load_dataset("wikitext", "wikitext-2-v1", split="train", revision="b08601e")
+ .filter(lambda example: len(example["text"].split()) > 256)
+ .shuffle(seed=42)
+ .select(range(subset_size))
+ .to_list()
+ )
+ chunks = [dataset[i : i + self.batch_size] for i in range(0, subset_size, self.batch_size)]
+ self.calibration_dataset = nncf.Dataset(chunks, self.get_transform_calibration_fn())
+
+ def save_compressed_model(self):
+ if self.backend == BackendType.CUDA_TORCH:
+ self.model_hf.float()
+ for module in self.model_hf.nncf.modules():
+ if isinstance(module, (AsymmetricWeightsDecompressor, SymmetricWeightsDecompressor)):
+ module.result_dtype = torch.float32
+ export_from_model(
+ self.model_hf, self.output_model_dir, stateful=False, compression_option="fp32", device="cuda"
+ )
+ else:
+ super().save_compressed_model()
+
+ def get_num_compressed(self):
+ """
+ Get number of quantization ops and sparsifier ops in the compressed IR.
+ """
+ if self.backend in PT_BACKENDS:
+ model = ov.Core().read_model(self.output_model_dir / self.OV_MODEL_NAME)
+ else:
+ model = self.model
+ self.run_info.num_compress_nodes = self.count_compressed_nodes_from_ir(model)
+
+ def _dump_model_fp32(self):
+ if self.backend == BackendType.CUDA_TORCH:
+ export_from_model(
+ self.model_hf, self.fp32_model_dir, stateful=False, compression_option="fp32", device="cuda"
+ )
+ else:
+ super()._dump_model_fp32()
+
+ def _compress(self):
+ super()._compress()
+ if self.backend in PT_BACKENDS:
+ # This helps reproducibility but is not needed in actual use.
+ for sparsifier in PTSparsifyActivationsAlgoBackend.get_sparsifiers(self.compressed_model):
+ original_dtype = sparsifier.running_threshold.dtype
+ sparsifier.running_threshold = sparsifier.running_threshold.half().to(original_dtype)
+
+
+class ImageClassificationTimmSparsifyActivations(SAPipelineMixin, ImageClassificationTimm):
+ DEFAULT_SUBSET_SIZE = 256
+
+ def prepare_calibration_dataset(self):
+ subset_size = self.compression_params.get("subset_size") or self.DEFAULT_SUBSET_SIZE
+ val_dataset = torchvision.datasets.ImageFolder(
+ root=self.data_dir / "imagenet" / "val", transform=self.transform
+ )
+ indices = np.random.default_rng(42).choice(len(val_dataset), size=subset_size, replace=False)
+ subset = torch.utils.data.Subset(val_dataset, indices=indices)
+ loader = torch.utils.data.DataLoader(subset, batch_size=self.batch_size, num_workers=2, shuffle=False)
+ self.calibration_dataset = nncf.Dataset(loader, self.get_transform_calibration_fn())
+
+ def get_num_compressed(self):
+ """
+ Get number of quantization ops and sparsifier ops in the compressed IR.
+ """
+ model = ov.Core().read_model(model=self.path_compressed_ir)
+ self.run_info.num_compress_nodes = self.count_compressed_nodes_from_ir(model)
diff --git a/tests/post_training/experimental/sparsify_activations/reference_data.yaml b/tests/post_training/experimental/sparsify_activations/reference_data.yaml
new file mode 100644
index 00000000000..3e368a9c185
--- /dev/null
+++ b/tests/post_training/experimental/sparsify_activations/reference_data.yaml
@@ -0,0 +1,47 @@
+tinyllama_backend_FP32:
+ metric_value: 1.0
+ num_int4: 0
+ num_int8: 0
+ num_sparse_activations: 0
+tinyllama_ffn_sparse20_backend_CUDA_TORCH:
+ metric_value: 0.7818
+ atol: 0.025
+ num_int4: 0
+ num_int8: 0
+ num_sparse_activations: 44
+tinyllama_ffn_sparse20_backend_TORCH:
+ metric_value: 0.7879
+ atol: 0.025
+ num_int4: 0
+ num_int8: 0
+ num_sparse_activations: 44
+tinyllama_int8_asym_data_free_ffn_sparse20_backend_CUDA_TORCH:
+ metric_value: 0.8044
+ atol: 0.025
+ num_int4: 0
+ num_int8: 312
+ num_sparse_activations: 44
+tinyllama_int8_asym_data_free_ffn_sparse20_backend_TORCH:
+ metric_value: 0.7846
+ atol: 0.030
+ num_int4: 0
+ num_int8: 312
+ num_sparse_activations: 44
+timm/deit3_small_patch16_224_backend_FP32:
+ metric_value: 0.8135
+ atol: 0.001
+ num_int4: 0
+ num_int8: 0
+ num_sparse_activations: 0
+timm/deit3_small_patch16_224_qkv_sparse20_fc1_sparse20_fc2_sparse30_backend_CUDA_TORCH:
+ metric_value: 0.8102
+ atol: 0.001
+ num_int4: 0
+ num_int8: 0
+ num_sparse_activations: 36
+timm/deit3_small_patch16_224_qkv_sparse20_fc1_sparse20_fc2_sparse30_backend_TORCH:
+ metric_value: 0.8102
+ atol: 0.001
+ num_int4: 0
+ num_int8: 0
+ num_sparse_activations: 36
\ No newline at end of file
diff --git a/tests/post_training/experimental/sparsify_activations/test_sparsify_activations_conformance.py b/tests/post_training/experimental/sparsify_activations/test_sparsify_activations_conformance.py
new file mode 100644
index 00000000000..ebcb1921981
--- /dev/null
+++ b/tests/post_training/experimental/sparsify_activations/test_sparsify_activations_conformance.py
@@ -0,0 +1,161 @@
+# Copyright (c) 2024 Intel Corporation
+# Licensed 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 time
+import traceback
+from collections import OrderedDict
+from pathlib import Path
+from typing import Dict, Optional
+
+import pandas as pd
+import pytest
+import yaml
+
+from tests.post_training.experimental.sparsify_activations.model_scope import SPARSIFY_ACTIVATIONS_TEST_CASES
+from tests.post_training.experimental.sparsify_activations.pipelines import SARunInfo
+from tests.post_training.pipelines.base import BackendType
+from tests.post_training.pipelines.base import BaseTestPipeline
+from tests.post_training.test_quantize_conformance import create_short_run_info
+from tests.post_training.test_quantize_conformance import fixture_batch_size # noqa: F401
+from tests.post_training.test_quantize_conformance import fixture_data # noqa: F401
+from tests.post_training.test_quantize_conformance import fixture_extra_columns # noqa: F401
+from tests.post_training.test_quantize_conformance import fixture_no_eval # noqa: F401
+from tests.post_training.test_quantize_conformance import fixture_output # noqa: F401
+from tests.post_training.test_quantize_conformance import fixture_run_benchmark_app # noqa: F401
+from tests.post_training.test_quantize_conformance import fixture_run_fp32_backend # noqa: F401
+from tests.post_training.test_quantize_conformance import fixture_run_torch_cuda_backend # noqa: F401
+from tests.post_training.test_quantize_conformance import fixture_subset_size # noqa: F401
+from tests.post_training.test_quantize_conformance import maybe_skip_test_case
+from tests.post_training.test_quantize_conformance import write_logs
+
+
+@pytest.fixture(scope="session", name="sparsify_activations_reference_data")
+def fixture_sparsify_activations_reference_data():
+ path_reference = Path(__file__).parent / "reference_data.yaml"
+ with path_reference.open() as f:
+ data = yaml.safe_load(f)
+ for test_case in data.values():
+ test_case["atol"] = test_case.get("atol", 1e-3)
+ return data
+
+
+@pytest.fixture(scope="session", name="sparsify_activations_result_data")
+def fixture_sparsify_activations_report_data(output_dir):
+ data: Dict[str, SARunInfo] = {}
+ yield data
+ if data:
+ test_results = OrderedDict(sorted(data.items()))
+ df = pd.DataFrame(v.get_result_dict() for v in test_results.values())
+ output_dir.mkdir(parents=True, exist_ok=True)
+ df.to_csv(output_dir / "results.csv", index=False)
+
+
+def create_pipeline_kwargs(
+ test_model_param: Dict,
+ subset_size,
+ test_case_name: str,
+ reference_data: Dict[str, Dict],
+ fp32_model_params: Dict[str, Dict],
+):
+ if subset_size:
+ if "compression_params" not in test_model_param:
+ test_model_param["compression_params"] = {}
+ test_model_param["compression_params"]["subset_size"] = subset_size
+
+ print("\n")
+ print(f"Model: {test_model_param['reported_name']}")
+ print(f"Backend: {test_model_param['backend']}")
+ print(f"Comprssion params: {test_model_param['compression_params']}")
+
+ # Get target fp32 metric value
+ model_id = test_model_param["model_id"]
+ fp32_test_case_name = fp32_model_params[model_id]["reported_name"] + f"_backend_{BackendType.FP32.value}"
+ test_reference = reference_data[test_case_name]
+ test_reference["metric_value_fp32"] = reference_data[fp32_test_case_name]["metric_value"]
+
+ return {
+ "reported_name": test_model_param["reported_name"],
+ "model_id": test_model_param["model_id"],
+ "backend": test_model_param["backend"],
+ "compression_params": test_model_param["compression_params"],
+ "params": test_model_param.get("params"),
+ "reference_data": test_reference,
+ }
+
+
+@pytest.mark.parametrize("test_case_name", SPARSIFY_ACTIVATIONS_TEST_CASES.keys())
+def test_sparsify_activations(
+ sparsify_activations_reference_data: dict,
+ test_case_name: str,
+ data_dir: Path,
+ output_dir: Path,
+ sparsify_activations_result_data: Dict[str, SARunInfo],
+ no_eval: bool,
+ batch_size: int,
+ run_fp32_backend: bool,
+ run_torch_cuda_backend: bool,
+ subset_size: Optional[int],
+ run_benchmark_app: bool,
+ capsys: pytest.CaptureFixture,
+ extra_columns: bool,
+):
+ pipeline = None
+ err_msg = None
+ test_model_param = None
+ start_time = time.perf_counter()
+ try:
+ if test_case_name not in sparsify_activations_reference_data:
+ raise RuntimeError(f"{test_case_name} is not defined in `sparsify_activations_reference_data` fixture")
+ test_model_param = SPARSIFY_ACTIVATIONS_TEST_CASES[test_case_name]
+ maybe_skip_test_case(test_model_param, run_fp32_backend, run_torch_cuda_backend, batch_size)
+ fp32_model_params = {
+ tc["model_id"]: tc for tc in SPARSIFY_ACTIVATIONS_TEST_CASES.values() if tc["backend"] == BackendType.FP32
+ }
+ pipeline_cls = test_model_param["pipeline_cls"]
+ pipeline_kwargs = create_pipeline_kwargs(
+ test_model_param, subset_size, test_case_name, sparsify_activations_reference_data, fp32_model_params
+ )
+ calibration_batch_size = batch_size or test_model_param.get("batch_size", 1)
+ pipeline_kwargs.update(
+ {
+ "output_dir": output_dir,
+ "data_dir": data_dir,
+ "no_eval": no_eval,
+ "run_benchmark_app": run_benchmark_app,
+ "batch_size": calibration_batch_size,
+ }
+ )
+ pipeline: BaseTestPipeline = pipeline_cls(**pipeline_kwargs)
+ pipeline.run()
+ except Exception as e:
+ err_msg = str(e)
+ traceback.print_exc()
+
+ if pipeline is not None:
+ pipeline.cleanup_cache()
+ run_info = pipeline.run_info
+ if err_msg:
+ run_info.status = f"{run_info.status} | {err_msg}" if run_info.status else err_msg
+
+ captured = capsys.readouterr()
+ write_logs(captured, pipeline)
+
+ if extra_columns:
+ pipeline.collect_data_from_stdout(captured.out)
+ else:
+ run_info = create_short_run_info(test_model_param, err_msg, test_case_name)
+
+ run_info.time_total = time.perf_counter() - start_time
+ sparsify_activations_result_data[test_case_name] = run_info
+
+ if err_msg:
+ pytest.fail(err_msg)
diff --git a/tests/post_training/pipelines/lm_weight_compression.py b/tests/post_training/pipelines/lm_weight_compression.py
index 31266d172f9..27479fe6a50 100644
--- a/tests/post_training/pipelines/lm_weight_compression.py
+++ b/tests/post_training/pipelines/lm_weight_compression.py
@@ -110,12 +110,14 @@ def prepare_preprocessor(self) -> None:
self.preprocessor = AutoTokenizer.from_pretrained(self.model_id)
def get_transform_calibration_fn(self):
- def transform_fn(data, max_tokens=128):
+ def transform_fn(data, max_tokens=128, filter_bad_tokens=True):
tokenized_text = self.preprocessor(data["text"], return_tensors="np")
-
- bad_tokens = self.preprocessor("