wehub-resource-sync
192 lines
8.1 KiB
Python
192 lines
8.1 KiB
Python
import time
|
|
|
|
import numpy as np
|
|
from tqdm.auto import tqdm
|
|
|
|
from shap import Explanation, links
|
|
from shap.maskers import FixedComposite, Image, Text
|
|
from shap.utils import MaskedModel, partition_tree_shuffle
|
|
from shap.utils._exceptions import DimensionError
|
|
|
|
from ._result import BenchmarkResult
|
|
|
|
|
|
class ExplanationError:
|
|
"""A measure of the explanation error relative to a model's actual output.
|
|
|
|
This benchmark metric measures the discrepancy between the output of the model predicted by an
|
|
attribution explanation vs. the actual output of the model. This discrepancy is measured over
|
|
many masking patterns drawn from permutations of the input features.
|
|
|
|
For explanations (like Shapley values) that explain the difference between one alternative and another
|
|
(for example a current sample and typical background feature values) there is possible explanation error
|
|
for every pattern of mixing foreground and background, or other words every possible masking pattern.
|
|
In this class we compute the standard deviation over these explanation errors where masking patterns
|
|
are drawn from prefixes of random feature permutations. This seems natural, and aligns with Shapley value
|
|
computations, but of course you could choose to summarize explanation errors in others ways as well.
|
|
"""
|
|
|
|
def __init__(
|
|
self,
|
|
masker,
|
|
model,
|
|
*model_args,
|
|
batch_size=500,
|
|
num_permutations=10,
|
|
link=links.identity,
|
|
linearize_link=True,
|
|
seed=38923,
|
|
):
|
|
"""Build a new explanation error benchmarker with the given masker, model, and model args.
|
|
|
|
Parameters
|
|
----------
|
|
masker : function or shap.Masker
|
|
The masker defines how we hide features during the perturbation process.
|
|
|
|
model : function or shap.Model
|
|
The model we want to evaluate explanations against.
|
|
|
|
model_args : ...
|
|
The list of arguments we will give to the model that we will have explained. When we later call this benchmark
|
|
object we should pass explanations that have been computed on this same data.
|
|
|
|
batch_size : int
|
|
The maximum batch size we should use when calling the model. For some large NLP models this needs to be set
|
|
lower (at say 1) to avoid running out of GPU memory.
|
|
|
|
num_permutations : int
|
|
How many permutations we will use to estimate the average explanation error for each sample. If you are running
|
|
this benchmark on a large dataset with many samples then you can reduce this value since the final result is
|
|
averaged over samples as well and the averages of both directly combine to reduce variance. So for 10k samples
|
|
num_permutations=1 is appropreiate.
|
|
|
|
link : function
|
|
Allows for a non-linear link function to be used to bringe between the model output space and the explanation
|
|
space.
|
|
|
|
linearize_link : bool
|
|
Non-linear links can destroy additive separation in generalized linear models, so by linearizing the link we can
|
|
retain additive separation. See upcoming paper/doc for details.
|
|
|
|
"""
|
|
self.masker = masker
|
|
self.model = model
|
|
self.model_args = model_args
|
|
self.num_permutations = num_permutations
|
|
self.link = link
|
|
self.linearize_link = linearize_link
|
|
self.model_args = model_args
|
|
self.batch_size = batch_size
|
|
self.seed = seed
|
|
|
|
# user must give valid masker
|
|
underlying_masker = masker.masker if isinstance(masker, FixedComposite) else masker
|
|
if isinstance(underlying_masker, Text):
|
|
self.data_type = "text"
|
|
elif isinstance(underlying_masker, Image):
|
|
self.data_type = "image"
|
|
else:
|
|
self.data_type = "tabular"
|
|
|
|
def __call__(self, explanation, name, step_fraction=0.01, indices=[], silent=False):
|
|
"""Run this benchmark on the given explanation."""
|
|
if isinstance(explanation, np.ndarray):
|
|
attributions = explanation
|
|
elif isinstance(explanation, Explanation):
|
|
attributions = explanation.values
|
|
else:
|
|
raise ValueError("The passed explanation must be either of type numpy.ndarray or shap.Explanation!")
|
|
|
|
if len(attributions) != len(self.model_args[0]):
|
|
emsg = "The explanation passed must have the same number of rows as the self.model_args that were passed!"
|
|
raise DimensionError(emsg)
|
|
|
|
# it is important that we choose the same permutations for the different explanations we are comparing
|
|
# so as to avoid needless noise
|
|
old_seed = np.random.seed()
|
|
np.random.seed(self.seed)
|
|
|
|
pbar = None
|
|
start_time = time.time()
|
|
svals = []
|
|
mask_vals = []
|
|
|
|
for i, args in enumerate(zip(*self.model_args)):
|
|
if len(args[0].shape) != len(attributions[i].shape):
|
|
raise ValueError(
|
|
"The passed explanation must have the same dim as the model_args and must not have a vector output!"
|
|
)
|
|
|
|
feature_size = np.prod(attributions[i].shape)
|
|
sample_attributions = attributions[i].flatten()
|
|
|
|
# compute any custom clustering for this row
|
|
row_clustering = None
|
|
if getattr(self.masker, "clustering", None) is not None:
|
|
if isinstance(self.masker.clustering, np.ndarray):
|
|
row_clustering = self.masker.clustering
|
|
elif callable(self.masker.clustering):
|
|
row_clustering = self.masker.clustering(*args)
|
|
else:
|
|
raise NotImplementedError(
|
|
"The masker passed has a .clustering attribute that is not yet supported by the ExplanationError benchmark!"
|
|
)
|
|
|
|
masked_model = MaskedModel(self.model, self.masker, self.link, self.linearize_link, *args)
|
|
|
|
total_values = None
|
|
for _ in range(self.num_permutations):
|
|
masks = []
|
|
mask = np.zeros(feature_size, dtype=bool)
|
|
masks.append(mask.copy())
|
|
ordered_inds = np.arange(feature_size)
|
|
|
|
# shuffle the indexes so we get a random permutation ordering
|
|
if row_clustering is not None:
|
|
inds_mask = np.ones(feature_size, dtype=bool)
|
|
partition_tree_shuffle(ordered_inds, inds_mask, row_clustering)
|
|
else:
|
|
np.random.shuffle(ordered_inds)
|
|
|
|
increment = max(1, int(feature_size * step_fraction))
|
|
for j in range(0, feature_size, increment):
|
|
mask[ordered_inds[np.arange(j, min(feature_size, j + increment))]] = True
|
|
masks.append(mask.copy())
|
|
mask_vals.append(masks)
|
|
|
|
values = []
|
|
masks_arr = np.array(masks)
|
|
for j in range(0, len(masks_arr), self.batch_size):
|
|
values.append(masked_model(masks_arr[j : j + self.batch_size]))
|
|
values = np.concatenate(values)
|
|
base_value = values[0]
|
|
for j, v in enumerate(values):
|
|
values[j] = (v - (base_value + np.sum(sample_attributions[masks_arr[j]]))) ** 2
|
|
|
|
if total_values is None:
|
|
total_values = values
|
|
else:
|
|
total_values += values
|
|
total_values /= self.num_permutations
|
|
|
|
svals.append(total_values)
|
|
|
|
if pbar is None and time.time() - start_time > 5:
|
|
pbar = tqdm(
|
|
total=len(self.model_args[0]), disable=silent, leave=False, desc=f"ExplanationError for {name}"
|
|
)
|
|
pbar.update(i + 1)
|
|
if pbar is not None:
|
|
pbar.update(1)
|
|
|
|
if pbar is not None:
|
|
pbar.close()
|
|
|
|
svals = np.array(svals)
|
|
|
|
# reset the random seed so we don't mess up the caller
|
|
np.random.seed(old_seed)
|
|
|
|
return BenchmarkResult("explanation error", name, value=np.sqrt(np.sum(total_values) / len(total_values)))
|