Attributors

This module implement the influence function.

class dattri.algorithm.influence_function.IFAttributorArnoldi(task: AttributionTask, layer_name: str | List[str] | None = None, device: str | None = 'cpu', precompute_data_ratio: float = 1.0, proj_dim: int = 100, max_iter: int = 100, norm_constant: float = 1.0, tol: float = 1e-07, regularization: float = 0.0, seed: int = 0)

Bases: BaseInnerProductAttributor

The inner product attributor with Arnoldi projection transformation.

__init__(task: AttributionTask, layer_name: str | List[str] | None = None, device: str | None = 'cpu', precompute_data_ratio: float = 1.0, proj_dim: int = 100, max_iter: int = 100, norm_constant: float = 1.0, tol: float = 1e-07, regularization: float = 0.0, seed: int = 0) → None

Initialize the Arnoldi projection attributor.

Parameters:

• task (AttributionTask) – The task to be attributed. Must be an instance of AttributionTask.

• layer_name (Optional[Union[str, List[str]]]) – The name of the layer to be used to calculate the train/test representations. If None, full parameters are used. This should be a string or a list of strings if multiple layers are needed. The name of layer should follow the key of model.named_parameters(). Default: None.

• device (str) – Device to run the attributor on. Default is “cpu”.

• precompute_data_ratio (float) – Ratio of full training data used to precompute the Arnoldi projector. Default is 1.0.

• proj_dim (int) – Dimension after projection. Corresponds to number of top eigenvalues to keep for Hessian approximation.

• max_iter (int) – Maximum iterations for Arnoldi Iteration. Default is 100.

• norm_constant (float) – Constant for the norm of the projected vector. May need to be > 1 for large number of parameters to avoid dividing the projected vector by a very large normalization constant. Default is 1.0.

• tol (float) – Convergence tolerance. Algorithm stops if the norm of the current basis vector < tol. Default is 1e-7.

• regularization (float) – Regularization term for Hessian vector product. Adding regularization * I to the Hessian matrix, where I is the identity matrix. Useful for singular or ill-conditioned matrices. Default is 0.0.

• seed (int) – Random seed for projector. Default is 0.

attribute(train_dataloader: DataLoader, test_dataloader: DataLoader, relatif_method: str | None = None) → torch.Tensor

Calculate the influence of the training set on the test set.

Parameters:

• train_dataloader (DataLoader) – Dataloader for training samples to calculate the influence. It can be a subset of the full training set if cache is called before. A subset means that only a part of the training set’s influence is calculated. The dataloader should not be shuffled.

• test_dataloader (DataLoader) – Dataloader for test samples to calculate the influence. The dataloader should not be shuffled.

• relatif_method (Optional[str]) – Method for normalizing the influence values. Supported options: - “l”: Normalizes by sqrt(g_i^T (H^-1 g_i)). - “theta”: Normalizes by ||H^-1 g_i||. - None: No normalization applied.

Returns:

The influence of the training set on the test set, with

the shape of (num_train_samples, num_test_samples).

Return type:

torch.Tensor

cache(full_train_dataloader: DataLoader) → None

Cache the dataset and pre-calculate the Arnoldi projector.

Parameters:

full_train_dataloader (DataLoader) – Dataloader with full training data.

generate_test_rep(ckpt_idx: int, data: Tuple[torch.Tensor, ...]) → torch.Tensor

Generate initial representations of test data.

Inner product attributors calculate the inner product between the (transformed) train representations and test representations. This function generates the initial test representations.

The default implementation calculates the gradient of the test loss with respect to the parameter. Subclasses may override this function to calculate something else.

Parameters:

• ckpt_idx (int) – The index of the model checkpoints. This index is used for ensembling different trained model checkpoints.

• data (Tuple[Tensor]) – The test data. Typically, this is a tuple of input data and target data but the number of items in this tuple should align with the corresponding argument in the target function. The tensors’ shape follows (1, batch_size, …).

Returns:

The initial representations of the test data. Typically,

it is a 2-d dimensional tensor with the shape of (batch_size, num_parameters).

Return type:

torch.Tensor

generate_train_rep(ckpt_idx: int, data: Tuple[torch.Tensor, ...]) → torch.Tensor

Generate initial representations of train data.

Inner product attributors calculate the inner product between the (transformed) train representations and test representations. This function generates the initial train representations.

The default implementation calculates the gradient of the train loss with respect to the parameter. Subclasses may override this function to calculate something else.

Parameters:

• ckpt_idx (int) – The index of the model checkpoints. This index is used for ensembling different trained model checkpoints.

• data (Tuple[Tensor]) – The train data. Typically, this is a tuple of input data and target data but the number of items in this tuple should align with the corresponding argument in the target function. The tensors’ shape follows (1, batch_size, …).

Returns:

The initial representations of the train data. Typically,

it is a 2-d dimensional tensor with the shape of (batch_size, num_parameters).

Return type:

torch.Tensor

self_attribute(train_dataloader: DataLoader, relatif_method: str | None = None) → torch.Tensor

Calculate the influence of the training set on itself.

Parameters:

• train_dataloader (DataLoader) – Dataloader for training samples to calculate the influence. It can be a subset of the full training set if cache is called before. A subset means that only a part of the training set’s influence is calculated. The dataloader should not be shuffled.

• relatif_method (Optional[str]) – Method for normalizing the influence values. Supported options: - “l”: Normalizes by sqrt(g_i^T (H^-1 g_i)). - “theta”: Normalizes by ||H^-1 g_i||. - None: No normalization applied.

Returns:

The influence of the training set on itself, with

the shape of (num_train_samples,).

Return type:

torch.Tensor

transform_test_rep(ckpt_idx: int, test_rep: Tensor) → Tensor

Transform the test representations via Arnoldi projection.

Parameters:

• ckpt_idx (int) – Index of the model checkpoints. Used for ensembling different trained model checkpoints.

• test_rep (torch.Tensor) – Test representations to be transformed. A 2-d tensor with shape (batch_size, num_params).

Returns:

Transformed test representations. A 2-d tensor with

shape (batch_size, proj_dim).

Return type:

torch.Tensor

Raises:

ValueError – If the Arnoldi projector has not been cached.

transform_train_rep(ckpt_idx: int, train_rep: Tensor) → Tensor

Transform the train representations via Arnoldi projection.

Parameters:

• ckpt_idx (int) – Index of the model checkpoints. Used for ensembling different trained model checkpoints.

• train_rep (torch.Tensor) – Train representations to be transformed. A 2-d tensor with shape (batch_size, num_params).

Returns:

Transformed train representations. A 2-d tensor with

shape (batch_size, proj_dim).

Return type:

torch.Tensor

Raises:

ValueError – If the Arnoldi projector has not been cached.

class dattri.algorithm.influence_function.IFAttributorCG(task: AttributionTask, layer_name: str | List[str] | None = None, device: str | None = 'cpu', max_iter: int = 10, tol: float = 1e-07, mode: str = 'rev-rev', regularization: float = 0.0)

Bases: BaseInnerProductAttributor

The inner product attributor with CG inverse hessian transformation.

__init__(task: AttributionTask, layer_name: str | List[str] | None = None, device: str | None = 'cpu', max_iter: int = 10, tol: float = 1e-07, mode: str = 'rev-rev', regularization: float = 0.0) → None

Initialize the CG inverse Hessian attributor.

Parameters:

• task (AttributionTask) – The task to be attributed. Must be an instance of AttributionTask.

• device (str) – Device to run the attributor on. Default is “cpu”.

• layer_name (Optional[Union[str, List[str]]]) – The name of the layer to be used to calculate the train/test representations. If None, full parameters are used. This should be a string or a list of strings if multiple layers are needed. The name of layer should follow the key of model.named_parameters(). Default: None.

• max_iter (int) – Maximum iterations for Conjugate Gradient Descent. Default is 10.

• tol (float) – Convergence tolerance. Algorithm stops if residual norm < tol. Default is 1e-7.

• mode (str) – Auto-diff mode. Options: - “rev-rev”: Two reverse-mode auto-diffs. Better compatibility, more memory cost. - “rev-fwd”: Reverse-mode + forward-mode. Memory-efficient, less compatible.

• regularization (float) – Regularization term for Hessian vector product. Adding regularization * I to the Hessian matrix, where I is the identity matrix. Useful for singular or ill-conditioned matrices. Default is 0.0.

attribute(train_dataloader: DataLoader, test_dataloader: DataLoader, relatif_method: str | None = None) → torch.Tensor

Calculate the influence of the training set on the test set.

Parameters:

• train_dataloader (DataLoader) – Dataloader for training samples to calculate the influence. It can be a subset of the full training set if cache is called before. A subset means that only a part of the training set’s influence is calculated. The dataloader should not be shuffled.

• test_dataloader (DataLoader) – Dataloader for test samples to calculate the influence. The dataloader should not be shuffled.

• relatif_method (Optional[str]) – Method for normalizing the influence values. Supported options: - “l”: Normalizes by sqrt(g_i^T (H^-1 g_i)). - “theta”: Normalizes by ||H^-1 g_i||. - None: No normalization applied.

Returns:

The influence of the training set on the test set, with

the shape of (num_train_samples, num_test_samples).

Return type:

torch.Tensor

cache(full_train_dataloader: DataLoader) → None

Cache the full training dataloader or precompute and cache more information.

By default, the cache function only caches the full training dataloader. Subclasses may override this function to precompute and cache more information.

Parameters:

full_train_dataloader (torch.utils.data.DataLoader) – Dataloader for the full training data. Ideally, the batch size of the dataloader should be the same as the number of training samples to get the best accuracy for some attributors. Smaller batch size may lead to a less accurate result but lower memory consumption.

generate_test_rep(ckpt_idx: int, data: Tuple[torch.Tensor, ...]) → torch.Tensor

Generate initial representations of test data.

Inner product attributors calculate the inner product between the (transformed) train representations and test representations. This function generates the initial test representations.

The default implementation calculates the gradient of the test loss with respect to the parameter. Subclasses may override this function to calculate something else.

Parameters:

• ckpt_idx (int) – The index of the model checkpoints. This index is used for ensembling different trained model checkpoints.

• data (Tuple[Tensor]) – The test data. Typically, this is a tuple of input data and target data but the number of items in this tuple should align with the corresponding argument in the target function. The tensors’ shape follows (1, batch_size, …).

Returns:

The initial representations of the test data. Typically,

it is a 2-d dimensional tensor with the shape of (batch_size, num_parameters).

Return type:

torch.Tensor

generate_train_rep(ckpt_idx: int, data: Tuple[torch.Tensor, ...]) → torch.Tensor

Generate initial representations of train data.

Inner product attributors calculate the inner product between the (transformed) train representations and test representations. This function generates the initial train representations.

The default implementation calculates the gradient of the train loss with respect to the parameter. Subclasses may override this function to calculate something else.

Parameters:

• ckpt_idx (int) – The index of the model checkpoints. This index is used for ensembling different trained model checkpoints.

• data (Tuple[Tensor]) – The train data. Typically, this is a tuple of input data and target data but the number of items in this tuple should align with the corresponding argument in the target function. The tensors’ shape follows (1, batch_size, …).

Returns:

The initial representations of the train data. Typically,

it is a 2-d dimensional tensor with the shape of (batch_size, num_parameters).

Return type:

torch.Tensor

self_attribute(train_dataloader: DataLoader, relatif_method: str | None = None) → torch.Tensor

Calculate the influence of the training set on itself.

Parameters:

• train_dataloader (DataLoader) – Dataloader for training samples to calculate the influence. It can be a subset of the full training set if cache is called before. A subset means that only a part of the training set’s influence is calculated. The dataloader should not be shuffled.

• relatif_method (Optional[str]) – Method for normalizing the influence values. Supported options: - “l”: Normalizes by sqrt(g_i^T (H^-1 g_i)). - “theta”: Normalizes by ||H^-1 g_i||. - None: No normalization applied.

Returns:

The influence of the training set on itself, with

the shape of (num_train_samples,).

Return type:

torch.Tensor

transform_test_rep(ckpt_idx: int, test_rep: Tensor) → Tensor

Calculate the transformation on the test rep through ihvp_cg.

Parameters:

• ckpt_idx (int) – Index of the model checkpoints. Used for ensembling different trained model checkpoints.

• test_rep (torch.Tensor) – Test representations to be transformed. Typically a 2-d tensor with shape (batch_size, num_parameters).

Returns:

Transformed test representations. Typically a 2-d

tensor with shape (batch_size, transformed_dimension).

Return type:

torch.Tensor

transform_train_rep(ckpt_idx: int, train_rep: Tensor) → Tensor

Transform the train representations.

Inner product attributor calculates the inner product between the (transformed) train representations and test representations. This function calculates the transformation of the train representations. For example, the transformation could be a dimension reduction of the train representations.

Parameters:

• ckpt_idx (int) – The index of the model checkpoints. This index is used for ensembling different trained model checkpoints.

• train_rep (torch.Tensor) – The train representations to be transformed. Typically, it is a 2-d dimensional tensor with the shape of (batch_size, num_parameters).

Returns:

The transformed train representations. Typically,

it is a 2-d dimensional tensor with the shape of (batch_size, transformed_dimension).

Return type:

torch.Tensor

class dattri.algorithm.influence_function.IFAttributorDataInf(task: AttributionTask, layer_name: str | List[str] | None = None, device: str | None = 'cpu', regularization: float = 0.0, fim_estimate_data_ratio: float = 1.0)

Bases: BaseInnerProductAttributor

The inner product attributor with DataInf inverse hessian transformation.

__init__(task: AttributionTask, layer_name: str | List[str] | None = None, device: str | None = 'cpu', regularization: float = 0.0, fim_estimate_data_ratio: float = 1.0) → None

Initialize the DataInf inverse Hessian attributor.

Parameters:

• task (AttributionTask) – The task to be attributed. Must be an instance of AttributionTask.

• layer_name (Optional[Union[str, List[str]]]) – The name of the layer to be used to calculate the train/test representations. If None, full parameters are used. This should be a string or a list of strings if multiple layers are needed. The name of layer should follow the key of model.named_parameters(). Default: None.

• device (str) – Device to run the attributor on. Default is “cpu”.

• regularization (float) – Regularization term for Hessian vector product. Adding regularization * I to the Hessian matrix, where I is the identity matrix. Useful for singular or ill-conditioned matrices. Default is 0.0.

• fim_estimate_data_ratio (float) – Ratio of full training data used to approximate the empirical Fisher information matrix. Default is 1.0.

attribute(train_dataloader: DataLoader, test_dataloader: DataLoader, relatif_method: str | None = None) → torch.Tensor

Calculate the influence of the training set on the test set.

Parameters:

• train_dataloader (DataLoader) – Dataloader for training samples to calculate the influence. It can be a subset of the full training set if cache is called before. A subset means that only a part of the training set’s influence is calculated. The dataloader should not be shuffled.

• test_dataloader (DataLoader) – Dataloader for test samples to calculate the influence. The dataloader should not be shuffled.

• relatif_method (Optional[str]) – Method for normalizing the influence values. Supported options: - “l”: Normalizes by sqrt(g_i^T (H^-1 g_i)). - “theta”: Normalizes by ||H^-1 g_i||. - None: No normalization applied.

Returns:

The influence of the training set on the test set, with

the shape of (num_train_samples, num_test_samples).

Return type:

torch.Tensor

cache(full_train_dataloader: DataLoader) → None

Cache the dataset and pre-calculate the Arnoldi projector.

Parameters:

full_train_dataloader (DataLoader) – Dataloader with full training data.

generate_test_rep(ckpt_idx: int, data: Tuple[torch.Tensor, ...]) → torch.Tensor

Generate initial representations of test data.

Inner product attributors calculate the inner product between the (transformed) train representations and test representations. This function generates the initial test representations.

The default implementation calculates the gradient of the test loss with respect to the parameter. Subclasses may override this function to calculate something else.

Parameters:

• ckpt_idx (int) – The index of the model checkpoints. This index is used for ensembling different trained model checkpoints.

• data (Tuple[Tensor]) – The test data. Typically, this is a tuple of input data and target data but the number of items in this tuple should align with the corresponding argument in the target function. The tensors’ shape follows (1, batch_size, …).

Returns:

The initial representations of the test data. Typically,

it is a 2-d dimensional tensor with the shape of (batch_size, num_parameters).

Return type:

torch.Tensor

generate_train_rep(ckpt_idx: int, data: Tuple[torch.Tensor, ...]) → torch.Tensor

Generate initial representations of train data.

Inner product attributors calculate the inner product between the (transformed) train representations and test representations. This function generates the initial train representations.

The default implementation calculates the gradient of the train loss with respect to the parameter. Subclasses may override this function to calculate something else.

Parameters:

• ckpt_idx (int) – The index of the model checkpoints. This index is used for ensembling different trained model checkpoints.

• data (Tuple[Tensor]) – The train data. Typically, this is a tuple of input data and target data but the number of items in this tuple should align with the corresponding argument in the target function. The tensors’ shape follows (1, batch_size, …).

Returns:

The initial representations of the train data. Typically,

it is a 2-d dimensional tensor with the shape of (batch_size, num_parameters).

Return type:

torch.Tensor

self_attribute(train_dataloader: DataLoader, relatif_method: str | None = None) → torch.Tensor

Calculate the influence of the training set on itself.

Parameters:

• train_dataloader (DataLoader) – Dataloader for training samples to calculate the influence. It can be a subset of the full training set if cache is called before. A subset means that only a part of the training set’s influence is calculated. The dataloader should not be shuffled.

• relatif_method (Optional[str]) – Method for normalizing the influence values. Supported options: - “l”: Normalizes by sqrt(g_i^T (H^-1 g_i)). - “theta”: Normalizes by ||H^-1 g_i||. - None: No normalization applied.

Returns:

The influence of the training set on itself, with

the shape of (num_train_samples,).

Return type:

torch.Tensor

transform_test_rep(ckpt_idx: int, test_rep: Tensor) → Tensor

Calculate the transformation on the test representations.

Parameters:

• ckpt_idx (int) – Index of the model checkpoints. Used for ensembling different trained model checkpoints.

• test_rep (torch.Tensor) – Test representations to be transformed. Typically a 2-d tensor with shape (batch_size, num_parameters).

Returns:

Transformed test representations. Typically a 2-d

tensor with shape (batch_size, transformed_dimension).

Return type:

torch.Tensor

transform_train_rep(ckpt_idx: int, train_rep: Tensor) → Tensor

Transform the train representations.

Inner product attributor calculates the inner product between the (transformed) train representations and test representations. This function calculates the transformation of the train representations. For example, the transformation could be a dimension reduction of the train representations.

Parameters:

• ckpt_idx (int) – The index of the model checkpoints. This index is used for ensembling different trained model checkpoints.

• train_rep (torch.Tensor) – The train representations to be transformed. Typically, it is a 2-d dimensional tensor with the shape of (batch_size, num_parameters).

Returns:

The transformed train representations. Typically,

it is a 2-d dimensional tensor with the shape of (batch_size, transformed_dimension).

Return type:

torch.Tensor

class dattri.algorithm.influence_function.IFAttributorEKFAC(task: AttributionTask, module_name: str | List[str] | None = None, device: str | None = 'cpu', damping: float = 0.0)

Bases: BaseInnerProductAttributor

The inner product attributor with EK-FAC inverse FIM transformation.

__init__(task: AttributionTask, module_name: str | List[str] | None = None, device: str | None = 'cpu', damping: float = 0.0) → None

Initialize the EK-FAC inverse FIM attributor.

Parameters:

• task (AttributionTask) –

  The task to be attributed. Must be an instance of AttributionTask. The loss function for EK-FAC attributor should return the following, - loss: a single tensor of loss. Should be the mean loss by the

  batch size.

  • mask (optional): a tensor of shape (batch_size, t), where 1’s

    indicate that the IFVP will be estimated on these input positions and 0’s indicate that these positions are irrelevant (e.g. padding tokens).

  t is the number of steps, or sequence length of the input data. If the input data are non-sequential, t should be set to 1. The FIM will be estimated on this function.

• module_name (Optional[Union[str, List[str]]]) – The name of the module to be used to calculate the train/test representations. If None, all linear modules are used. This should be a string or a list of strings if multiple modules are needed. The name of module should follow the key of model.named_modules(). Default: None.

• device (str) – Device to run the attributor on. Default is “cpu”.

• damping (float) – Damping factor used for non-convexity in EK-FAC IFVP calculation. Default is 0.0.

Raises:

ValueError – If there are multiple checkpoints in task.

attribute(train_dataloader: DataLoader, test_dataloader: DataLoader, relatif_method: str | None = None) → torch.Tensor

Calculate the influence of the training set on the test set.

Parameters:

• train_dataloader (DataLoader) – Dataloader for training samples to calculate the influence. It can be a subset of the full training set if cache is called before. A subset means that only a part of the training set’s influence is calculated. The dataloader should not be shuffled.

• test_dataloader (DataLoader) – Dataloader for test samples to calculate the influence. The dataloader should not be shuffled.

• relatif_method (Optional[str]) – Method for normalizing the influence values. Supported options: - “l”: Normalizes by sqrt(g_i^T (H^-1 g_i)). - “theta”: Normalizes by ||H^-1 g_i||. - None: No normalization applied.

Returns:

The influence of the training set on the test set, with

the shape of (num_train_samples, num_test_samples).

Return type:

torch.Tensor

cache(full_train_dataloader: DataLoader, max_iter: int | None = None) → None

Cache the dataset and statistics for inverse FIM calculation.

Cache the full training dataset as other attributors. Estimate and cache the covariance matrices, eigenvector matrices and corrected eigenvalues based on the samples of training data.

Parameters:

• full_train_dataloader (DataLoader) – The dataloader with full training samples for inverse FIM calculation.

• max_iter (int, optional) – An integer indicating the maximum number of batches that will be used for estimating the the covariance matrices and lambdas. Default to length of full_train_dataloader.

generate_test_rep(ckpt_idx: int, data: Tuple[torch.Tensor, ...]) → torch.Tensor

Generate initial representations of test data.

Inner product attributors calculate the inner product between the (transformed) train representations and test representations. This function generates the initial test representations.

The default implementation calculates the gradient of the test loss with respect to the parameter. Subclasses may override this function to calculate something else.

Parameters:

• ckpt_idx (int) – The index of the model checkpoints. This index is used for ensembling different trained model checkpoints.

• data (Tuple[Tensor]) – The test data. Typically, this is a tuple of input data and target data but the number of items in this tuple should align with the corresponding argument in the target function. The tensors’ shape follows (1, batch_size, …).

Returns:

The initial representations of the test data. Typically,

it is a 2-d dimensional tensor with the shape of (batch_size, num_parameters).

Return type:

torch.Tensor

generate_train_rep(ckpt_idx: int, data: Tuple[torch.Tensor, ...]) → torch.Tensor

Generate initial representations of train data.

Inner product attributors calculate the inner product between the (transformed) train representations and test representations. This function generates the initial train representations.

The default implementation calculates the gradient of the train loss with respect to the parameter. Subclasses may override this function to calculate something else.

Parameters:

• ckpt_idx (int) – The index of the model checkpoints. This index is used for ensembling different trained model checkpoints.

• data (Tuple[Tensor]) – The train data. Typically, this is a tuple of input data and target data but the number of items in this tuple should align with the corresponding argument in the target function. The tensors’ shape follows (1, batch_size, …).

Returns:

The initial representations of the train data. Typically,

it is a 2-d dimensional tensor with the shape of (batch_size, num_parameters).

Return type:

torch.Tensor

self_attribute(train_dataloader: DataLoader, relatif_method: str | None = None) → torch.Tensor

Calculate the influence of the training set on itself.

Parameters:

• train_dataloader (DataLoader) – Dataloader for training samples to calculate the influence. It can be a subset of the full training set if cache is called before. A subset means that only a part of the training set’s influence is calculated. The dataloader should not be shuffled.

• relatif_method (Optional[str]) – Method for normalizing the influence values. Supported options: - “l”: Normalizes by sqrt(g_i^T (H^-1 g_i)). - “theta”: Normalizes by ||H^-1 g_i||. - None: No normalization applied.

Returns:

The influence of the training set on itself, with

the shape of (num_train_samples,).

Return type:

torch.Tensor

transform_test_rep(ckpt_idx: int, test_rep: Tensor) → Tensor

Calculate the transformation on the test representations.

Parameters:

• ckpt_idx (int) – Index of the model checkpoints. Used for ensembling different trained model checkpoints.

• test_rep (torch.Tensor) – Test representations to be transformed. Typically a 2-d tensor with shape (batch_size, num_parameters).

Returns:

Transformed test representations. Typically a 2-d

tensor with shape (batch_size, transformed_dimension).

Return type:

torch.Tensor

Raises:

ValueError – If specifies a non-zero ckpt_idx.

transform_train_rep(ckpt_idx: int, train_rep: Tensor) → Tensor

Transform the train representations.

Inner product attributor calculates the inner product between the (transformed) train representations and test representations. This function calculates the transformation of the train representations. For example, the transformation could be a dimension reduction of the train representations.

Parameters:

• ckpt_idx (int) – The index of the model checkpoints. This index is used for ensembling different trained model checkpoints.

• train_rep (torch.Tensor) – The train representations to be transformed. Typically, it is a 2-d dimensional tensor with the shape of (batch_size, num_parameters).

Returns:

The transformed train representations. Typically,

it is a 2-d dimensional tensor with the shape of (batch_size, transformed_dimension).

Return type:

torch.Tensor

class dattri.algorithm.influence_function.IFAttributorExplicit(task: AttributionTask, layer_name: str | List[str] | None = None, device: str | None = 'cpu', regularization: float = 0.0)

Bases: BaseInnerProductAttributor

The inner product attributor with explicit inverse hessian transformation.

__init__(task: AttributionTask, layer_name: str | List[str] | None = None, device: str | None = 'cpu', regularization: float = 0.0) → None

Initialize the explicit inverse Hessian attributor.

Parameters:

• task (AttributionTask) – Task to attribute. Must be an instance of AttributionTask.

• layer_name (Optional[Union[str, List[str]]]) – The name of the layer to be used to calculate the train/test representations. If None, full parameters are used. This should be a string or a list of strings if multiple layers are needed. The name of layer should follow the key of model.named_parameters(). Default: None.

• device (str) – Device to run the attributor on. Default is “cpu”.

• regularization (float) – Regularization term added to Hessian matrix. Useful for singular or ill-conditioned Hessian matrices. Added as regularization * I, where I is the identity matrix. Default is 0.0.

attribute(train_dataloader: DataLoader, test_dataloader: DataLoader, relatif_method: str | None = None) → torch.Tensor

Calculate the influence of the training set on the test set.

Parameters:

• train_dataloader (DataLoader) – Dataloader for training samples to calculate the influence. It can be a subset of the full training set if cache is called before. A subset means that only a part of the training set’s influence is calculated. The dataloader should not be shuffled.

• test_dataloader (DataLoader) – Dataloader for test samples to calculate the influence. The dataloader should not be shuffled.

• relatif_method (Optional[str]) – Method for normalizing the influence values. Supported options: - “l”: Normalizes by sqrt(g_i^T (H^-1 g_i)). - “theta”: Normalizes by ||H^-1 g_i||. - None: No normalization applied.

Returns:

The influence of the training set on the test set, with

the shape of (num_train_samples, num_test_samples).

Return type:

torch.Tensor

cache(full_train_dataloader: DataLoader) → None

Cache the full training dataloader or precompute and cache more information.

By default, the cache function only caches the full training dataloader. Subclasses may override this function to precompute and cache more information.

Parameters:

full_train_dataloader (torch.utils.data.DataLoader) – Dataloader for the full training data. Ideally, the batch size of the dataloader should be the same as the number of training samples to get the best accuracy for some attributors. Smaller batch size may lead to a less accurate result but lower memory consumption.

generate_test_rep(ckpt_idx: int, data: Tuple[torch.Tensor, ...]) → torch.Tensor

Generate initial representations of test data.

Inner product attributors calculate the inner product between the (transformed) train representations and test representations. This function generates the initial test representations.

The default implementation calculates the gradient of the test loss with respect to the parameter. Subclasses may override this function to calculate something else.

Parameters:

• ckpt_idx (int) – The index of the model checkpoints. This index is used for ensembling different trained model checkpoints.

• data (Tuple[Tensor]) – The test data. Typically, this is a tuple of input data and target data but the number of items in this tuple should align with the corresponding argument in the target function. The tensors’ shape follows (1, batch_size, …).

Returns:

The initial representations of the test data. Typically,

it is a 2-d dimensional tensor with the shape of (batch_size, num_parameters).

Return type:

torch.Tensor

generate_train_rep(ckpt_idx: int, data: Tuple[torch.Tensor, ...]) → torch.Tensor

Generate initial representations of train data.

Inner product attributors calculate the inner product between the (transformed) train representations and test representations. This function generates the initial train representations.

The default implementation calculates the gradient of the train loss with respect to the parameter. Subclasses may override this function to calculate something else.

Parameters:

• ckpt_idx (int) – The index of the model checkpoints. This index is used for ensembling different trained model checkpoints.

• data (Tuple[Tensor]) – The train data. Typically, this is a tuple of input data and target data but the number of items in this tuple should align with the corresponding argument in the target function. The tensors’ shape follows (1, batch_size, …).

Returns:

The initial representations of the train data. Typically,

it is a 2-d dimensional tensor with the shape of (batch_size, num_parameters).

Return type:

torch.Tensor

self_attribute(train_dataloader: DataLoader, relatif_method: str | None = None) → torch.Tensor

Calculate the influence of the training set on itself.

Parameters:

• train_dataloader (DataLoader) – Dataloader for training samples to calculate the influence. It can be a subset of the full training set if cache is called before. A subset means that only a part of the training set’s influence is calculated. The dataloader should not be shuffled.

• relatif_method (Optional[str]) – Method for normalizing the influence values. Supported options: - “l”: Normalizes by sqrt(g_i^T (H^-1 g_i)). - “theta”: Normalizes by ||H^-1 g_i||. - None: No normalization applied.

Returns:

The influence of the training set on itself, with

the shape of (num_train_samples,).

Return type:

torch.Tensor

transform_test_rep(ckpt_idx: int, test_rep: Tensor) → Tensor

Calculate the transformation on the test rep through ihvp_explicit.

Parameters:

• ckpt_idx (int) – Index of model parameters. Used for ensembling.

• test_rep (torch.Tensor) – Test representations to be transformed. Typically a 2-d tensor with shape (batch_size, num_parameters).

Returns:

Transformed test representations. Typically a 2-d

tensor with shape (batch_size, transformed_dimension).

Return type:

torch.Tensor

transform_train_rep(ckpt_idx: int, train_rep: Tensor) → Tensor

Transform the train representations.

Inner product attributor calculates the inner product between the (transformed) train representations and test representations. This function calculates the transformation of the train representations. For example, the transformation could be a dimension reduction of the train representations.

Parameters:

• ckpt_idx (int) – The index of the model checkpoints. This index is used for ensembling different trained model checkpoints.

• train_rep (torch.Tensor) – The train representations to be transformed. Typically, it is a 2-d dimensional tensor with the shape of (batch_size, num_parameters).

Returns:

The transformed train representations. Typically,

it is a 2-d dimensional tensor with the shape of (batch_size, transformed_dimension).

Return type:

torch.Tensor

class dattri.algorithm.influence_function.IFAttributorLiSSA(task: AttributionTask, layer_name: str | List[str] | None = None, device: str | None = 'cpu', batch_size: int = 1, num_repeat: int = 1, recursion_depth: int = 5000, damping: float = 0.0, scaling: float = 50.0, mode: str = 'rev-rev')

Bases: BaseInnerProductAttributor

The inner product attributor with LiSSA inverse hessian transformation.

__init__(task: AttributionTask, layer_name: str | List[str] | None = None, device: str | None = 'cpu', batch_size: int = 1, num_repeat: int = 1, recursion_depth: int = 5000, damping: float = 0.0, scaling: float = 50.0, mode: str = 'rev-rev') → None

Initialize the LiSSA inverse Hessian attributor.

Parameters:

• task (AttributionTask) – The task to be attributed. Must be an instance of AttributionTask.

• layer_name (Optional[Union[str, List[str]]]) – The name of the layer to be used to calculate the train/test representations. If None, full parameters are used. This should be a string or a list of strings if multiple layers are needed. The name of layer should follow the key of model.named_parameters(). Default: None.

• device (str) – Device to run the attributor on. Default is “cpu”.

• batch_size (int) – Batch size for LiSSA inner loop update. Default is 1.

• num_repeat (int) – Number of samples of the HVP approximation to average. Default is 1.

• recursion_depth (int) – Number of recursions to estimate each IHVP sample. Default is 5000.

• damping (float) – Damping factor for non-convexity in LiSSA IHVP calculation.

• scaling (float) – Scaling factor for convergence in LiSSA IHVP calculation.

• mode (str) – Auto-diff mode. Options: - “rev-rev”: Two reverse-mode auto-diffs. Better compatibility, more memory cost. - “rev-fwd”: Reverse-mode + forward-mode. Memory-efficient, less compatible.

attribute(train_dataloader: DataLoader, test_dataloader: DataLoader, relatif_method: str | None = None) → torch.Tensor

Calculate the influence of the training set on the test set.

Parameters:

• train_dataloader (DataLoader) – Dataloader for training samples to calculate the influence. It can be a subset of the full training set if cache is called before. A subset means that only a part of the training set’s influence is calculated. The dataloader should not be shuffled.

• test_dataloader (DataLoader) – Dataloader for test samples to calculate the influence. The dataloader should not be shuffled.

• relatif_method (Optional[str]) – Method for normalizing the influence values. Supported options: - “l”: Normalizes by sqrt(g_i^T (H^-1 g_i)). - “theta”: Normalizes by ||H^-1 g_i||. - None: No normalization applied.

Returns:

The influence of the training set on the test set, with

the shape of (num_train_samples, num_test_samples).

Return type:

torch.Tensor

cache(full_train_dataloader: DataLoader) → None

Cache the full training dataloader or precompute and cache more information.

By default, the cache function only caches the full training dataloader. Subclasses may override this function to precompute and cache more information.

Parameters:

full_train_dataloader (torch.utils.data.DataLoader) – Dataloader for the full training data. Ideally, the batch size of the dataloader should be the same as the number of training samples to get the best accuracy for some attributors. Smaller batch size may lead to a less accurate result but lower memory consumption.

generate_test_rep(ckpt_idx: int, data: Tuple[torch.Tensor, ...]) → torch.Tensor

Generate initial representations of test data.

Inner product attributors calculate the inner product between the (transformed) train representations and test representations. This function generates the initial test representations.

The default implementation calculates the gradient of the test loss with respect to the parameter. Subclasses may override this function to calculate something else.

Parameters:

• ckpt_idx (int) – The index of the model checkpoints. This index is used for ensembling different trained model checkpoints.

• data (Tuple[Tensor]) – The test data. Typically, this is a tuple of input data and target data but the number of items in this tuple should align with the corresponding argument in the target function. The tensors’ shape follows (1, batch_size, …).

Returns:

The initial representations of the test data. Typically,

it is a 2-d dimensional tensor with the shape of (batch_size, num_parameters).

Return type:

torch.Tensor

generate_train_rep(ckpt_idx: int, data: Tuple[torch.Tensor, ...]) → torch.Tensor

Generate initial representations of train data.

Inner product attributors calculate the inner product between the (transformed) train representations and test representations. This function generates the initial train representations.

The default implementation calculates the gradient of the train loss with respect to the parameter. Subclasses may override this function to calculate something else.

Parameters:

• ckpt_idx (int) – The index of the model checkpoints. This index is used for ensembling different trained model checkpoints.

• data (Tuple[Tensor]) – The train data. Typically, this is a tuple of input data and target data but the number of items in this tuple should align with the corresponding argument in the target function. The tensors’ shape follows (1, batch_size, …).

Returns:

The initial representations of the train data. Typically,

it is a 2-d dimensional tensor with the shape of (batch_size, num_parameters).

Return type:

torch.Tensor

static lissa_collate_fn(sampled_input: List[Tensor]) → Tuple[Tensor, List[Tuple[Tensor, ...]]]

Collate function for LISSA.

Parameters:

sampled_input (List[Tensor]) – The sampled input from the dataloader.

Returns:

The collated input for the LISSA.

Return type:

Tuple[Tensor, List[Tuple[Tensor, …]]]

self_attribute(train_dataloader: DataLoader, relatif_method: str | None = None) → torch.Tensor

Calculate the influence of the training set on itself.

Parameters:

• train_dataloader (DataLoader) – Dataloader for training samples to calculate the influence. It can be a subset of the full training set if cache is called before. A subset means that only a part of the training set’s influence is calculated. The dataloader should not be shuffled.

• relatif_method (Optional[str]) – Method for normalizing the influence values. Supported options: - “l”: Normalizes by sqrt(g_i^T (H^-1 g_i)). - “theta”: Normalizes by ||H^-1 g_i||. - None: No normalization applied.

Returns:

The influence of the training set on itself, with

the shape of (num_train_samples,).

Return type:

torch.Tensor

transform_test_rep(ckpt_idx: int, test_rep: Tensor) → Tensor

Calculate the transformation on the test rep through ihvp_lissa.

Parameters:

• ckpt_idx (int) – Index of the model checkpoints. Used for ensembling different trained model checkpoints.

• test_rep (torch.Tensor) – Test representations to be transformed. Typically a 2-d tensor with shape (batch_size, num_parameters).

Returns:

Transformed test representations. Typically a 2-d

tensor with shape (batch_size, transformed_dimension).

Return type:

torch.Tensor

transform_train_rep(ckpt_idx: int, train_rep: Tensor) → Tensor

Transform the train representations.

Inner product attributor calculates the inner product between the (transformed) train representations and test representations. This function calculates the transformation of the train representations. For example, the transformation could be a dimension reduction of the train representations.

Parameters:

• ckpt_idx (int) – The index of the model checkpoints. This index is used for ensembling different trained model checkpoints.

• train_rep (torch.Tensor) – The train representations to be transformed. Typically, it is a 2-d dimensional tensor with the shape of (batch_size, num_parameters).

Returns:

The transformed train representations. Typically,

it is a 2-d dimensional tensor with the shape of (batch_size, transformed_dimension).

Return type:

torch.Tensor

class dattri.algorithm.tracin.TracInAttributor(task: AttributionTask, weight_list: Tensor, normalized_grad: bool, projector_kwargs: Dict[str, Any] | None = None, layer_name: str | List[str] | None = None, device: str = 'cpu')

Bases: BaseAttributor

TracIn attributor.

__init__(task: AttributionTask, weight_list: Tensor, normalized_grad: bool, projector_kwargs: Dict[str, Any] | None = None, layer_name: str | List[str] | None = None, device: str = 'cpu') → None

Initialize the TracIn attributor.

Parameters:

• task (AttributionTask) – The task to be attributed. Please refer to the AttributionTask for more details.

• weight_list (Tensor) – The weight used for the “weighted sum”. For TracIn/CosIn, this will contain a list of learning rates at each ckpt; for Grad-Dot/Grad-Cos, this will be a list of ones.

• normalized_grad (bool) – Whether to apply normalization to gradients.

• projector_kwargs (Optional[Dict[str, Any]]) – The keyword arguments for the projector.

• layer_name (Optional[Union[str, List[str]]]) – The name of the layer to be used to calculate the train/test representations. If None, full parameters are used. This should be a string or a list of strings if multiple layers are needed. The name of layer should follow the key of model.named_parameters(). Default: None.

• device (str) – The device to run the attributor. Default is cpu.

attribute(train_dataloader: DataLoader, test_dataloader: DataLoader) → Tensor

Calculate the influence of the training set on the test set.

Parameters:

• train_dataloader (torch.utils.data.DataLoader) – The dataloader for training samples to calculate the influence. It can be a subset of the full training set if cache is called before. A subset means that only a part of the training set’s influence is calculated. The dataloader should not be shuffled.

• test_dataloader (torch.utils.data.DataLoader) – The dataloader for test samples to calculate the influence. The dataloader should not be shuffled.

Raises:

ValueError – The length of params_list and weight_list don’t match.

Returns:

The influence of the training set on the test set, with

the shape of (num_train_samples, num_test_samples).

Return type:

Tensor

cache() → None

Precompute and cache some values for efficiency.

self_attribute(train_dataloader: DataLoader) → Tensor

Calculate the influence of the training set on itself.

Parameters:

train_dataloader (torch.utils.data.DataLoader) – The dataloader for training samples to calculate the influence. It can be a subset of the full training set if cache is called before. A subset means that only a part of the training set’s influence is calculated. The dataloader should not be shuffled.

Raises:

ValueError – The length of params_list and weight_list don’t match.

Returns:

The influence of the training set on itself, with

the shape of (num_train_samples,).

Return type:

Tensor

class dattri.algorithm.trak.TRAKAttributor(task: AttributionTask, correct_probability_func: Callable, projector_kwargs: Dict[str, Any] | None = None, layer_name: str | List[str] | None = None, device: str = 'cpu', regularization: float = 0.0)

Bases: BaseAttributor

TRAK attributor.

__init__(task: AttributionTask, correct_probability_func: Callable, projector_kwargs: Dict[str, Any] | None = None, layer_name: str | List[str] | None = None, device: str = 'cpu', regularization: float = 0.0) → None

Initialize the TRAK attributor.

Parameters:

• task (AttributionTask) – The task to be attributed. Please refer to the AttributionTask for more details.

• correct_probability_func (Callable) –

  The function to calculate the probability to correctly predict the label of the input data. A typical example is as follows: ```python @flatten_func(model) def m(params, image_label_pair):

  image, label = image_label_pair image_t = image.unsqueeze(0) label_t = label.unsqueeze(0) loss = nn.CrossEntropyLoss() yhat = torch.func.functional_call(model, params, image_t) p = torch.exp(-loss(yhat, label_t)) return p

  ```

• projector_kwargs (Optional[Dict[str, Any]], optional) – The kwargs for the random projection. Defaults to None.

• layer_name (Optional[Union[str, List[str]]]) – The name of the layer to be used to calculate the train/test representations. If None, full parameters are used. This should be a string or a list of strings if multiple layers are needed. The name of layer should follow the key of model.named_parameters(). Default: None.

• device (str) – The device to run the attributor. Default is “cpu”.

• regularization (float) – Regularization term add before matrix inversion. Useful for singular or ill-conditioned matrices. Added as regularization * I, where I is the identity matrix. Default is 0.0.

attribute(test_dataloader: torch.utils.data.DataLoader, train_dataloader: torch.utils.data.DataLoader | None = None) → torch.Tensor

Calculate the influence of the training set on the test set.

Parameters:

• train_dataloader (torch.utils.data.DataLoader) – The dataloader for training samples to calculate the influence. If cache is called before attribute, this dataloader can consists of a subset of the full training dataset cached in cache. In this case, only a part of the training set’s influence will be calculated. The dataloader should not be shuffled.

• test_dataloader (torch.utils.data.DataLoader) – The dataloader for test samples to calculate the influence. The dataloader should not be shuffled.

Returns:

The influence of the training set on the test set, with

the shape of (num_train_samples, num_test_samples).

Return type:

torch.Tensor

Raises:

ValueError – If the train_dataloader is not None and the full training dataloader is cached or no train_loader is provided in both cases.

cache(full_train_dataloader: DataLoader) → None

Cache the dataset for gradient calculation.

Parameters:

full_train_dataloader (torch.utils.data.DataLoader) – The dataloader with full training samples for gradient calculation.

self_attribute(train_dataloader: torch.utils.data.DataLoader | None = None) → torch.Tensor

Calculate the influence of the training set on itself.

Parameters:

train_dataloader (torch.utils.data.DataLoader) – The dataloader for training samples to calculate the influence. If cache is called before attribute, this dataloader can consists of a subset of the full training dataset cached in cache. In this case, only a part of the training set’s influence will be calculated. The dataloader should not be shuffled.

Returns:

The influence of the training set on itself, with

the shape of (num_train_samples,).

Return type:

torch.Tensor

Raises:

ValueError – If the train_dataloader is not None and the full training dataloader is cached or no train_loader is provided in both cases.

class dattri.algorithm.rps.RPSAttributor(task: AttributionTask, final_linear_layer_name: str, normalize_preactivate: bool = False, l2_strength: float = 0.003, epoch: int = 3000, device: str = 'cpu')

Bases: BaseAttributor

Representer point selection attributor.

__init__(task: AttributionTask, final_linear_layer_name: str, normalize_preactivate: bool = False, l2_strength: float = 0.003, epoch: int = 3000, device: str = 'cpu') → None

Representer point selection attributor.

Parameters:

• task (AttributionTask) – The task to be attributed. Please refer to the AttributionTask for more details. Notably, the target_func is required to have inputs are list of pre-activation values (f_i in the paper) and list of labels. Typical examples are loss functions such as BCELoss and CELoss. We also assume the model has a final linear layer. RPS will extract the final linear layer’s input and its parameter. The parameters will be used for the initialization of the l2-finetuning. That is, model_output = linear(second-to-last feature).

• final_linear_layer_name (str) – The name of the final linear layer’s name in the model.

• normalize_preactivate (bool) – If set to true, then the intermediate layer output will be normalized. The value of the output inner-product will not be affected by the value of individual output magnitude.

• l2_strength (float) – The l2 regularization to fine-tune the last layer.

• epoch (int) – The number of epoch used to fine-tune the last layer.

• device (str) – The device to run the attributor. Default is cpu.

attribute(train_dataloader: DataLoader, test_dataloader: DataLoader) → Tensor

Calculate the influence of the training set on the test set.

Parameters:

• train_dataloader (DataLoader) – The dataloader for training samples to calculate the influence. It can be a subset of the full training set if cache is called before. A subset means that only a part of the training set’s influence is calculated. The dataloader should not be shuffled.

• test_dataloader (DataLoader) – The dataloader for test samples to calculate the influence. The dataloader should not be shuffled.

Returns:

The influence of the training set on the test set, with the shape

of (num_train_samples, num_test_samples).

Return type:

Tensor

cache(full_train_dataloader: DataLoader) → None

Cache the full dataset for fine-tuning.

Parameters:

full_train_dataloader (DataLoader) – The dataloader with full training samples for the last linear layer fine-tuning.

class dattri.algorithm.data_shapley.KNNShapleyAttributor(k_neighbors: int, task: AttributionTask = None, distance_func: Callable | None = None)

Bases: BaseAttributor

KNN Data Shapley Attributor.

__init__(k_neighbors: int, task: AttributionTask = None, distance_func: Callable | None = None) → None

Initialize the AttributionTask.

KNN Data Shapley Valuation is generally dataset-specific. Passing a model is optional and currently can be done in the customizable distance function.

Parameters:

• k_neighbors (int) – The number of neighbors in KNN model.

• task (AttributionTask) – The task to be attributed. Used to pass the model and hook information in this attributor. Please refer to the AttributionTask for more details.

• distance_func (Callable, optional) –

  Customizable function used for distance calculation in KNN. The function can be quite flexible in terms of what is calculated, but it should take two batches of data as input. A typical example is as follows: ```python def f(batch_x, batch_y):

  coord1 = batch_x[0] coord2 = batch_y[0] return torch.cdist(coord1, coord2)

  ```. If not provided, a default Euclidean distance function will be used.

Raises:

NotImplementedError – If task is not None.

attribute(train_dataloader: torch.utils.data.DataLoader, test_dataloader: torch.utils.data.DataLoader, train_labels: List[int] | None = None, test_labels: List[int] | None = None) → None

Calculate the KNN shapley values of the training set on each test sample.

Parameters:

• train_dataloader (torch.utils.data.DataLoader) – The dataloader for training samples to calculate the shapley values. The dataloader should not be shuffled.

• test_dataloader (torch.utils.data.DataLoader) – The dataloader for test samples to calculate the shapley values. The dataloader should not be shuffled.

• train_labels – (List[int], optional): The list of training labels, with the same size and order of the training dataloader. If not provided, the last element in each batch from the loader will be used as label.

• test_labels – (List[int], optional): The list of test labels, with the same size and order of the test dataloader. If not provided, the last element in each batch from the loader will be used as label.

Returns:

The KNN shapley values of the training set on the test set, with

the shape of (num_train_samples, num_test_samples).

Return type:

Tensor

Raises:

ValueError – If the length of provided labels and dataset mismatch.

cache() → None

Precompute and cache some values for efficiency.

class dattri.algorithm.LoGraAttributor(task: AttributionTask, layer_names: str | List[str] | None = None, hessian: Literal['Identity', 'eFIM'] = 'eFIM', damping: float | None = None, device: str = 'cpu', proj_dim: int = 4096, offload: Literal['none', 'cpu', 'disk'] = 'cpu', cache_dir: str | None = None, chunk_size: int = 16)

Bases: BlockProjectedIFAttributor

LoGra Attributor.

Low-Rank Gradient Projection (LoGra) attributor that uses normal projection for the first stage and identity projection (no compression) for the second stage.

This is equivalent to the original LoGra method from the paper.

The projection is factorized: if you specify proj_dim=4096, each component will have dimension sqrt(4096)=64, and the Kronecker product will have dimension 4096.

__init__(task: AttributionTask, layer_names: str | List[str] | None = None, hessian: Literal['Identity', 'eFIM'] = 'eFIM', damping: float | None = None, device: str = 'cpu', proj_dim: int = 4096, offload: Literal['none', 'cpu', 'disk'] = 'cpu', cache_dir: str | None = None, chunk_size: int = 16) → None

Initialize LoGra attributor.

Parameters:

• task – Attribution task containing model, loss function, and checkpoints

• layer_names – Names of layers where gradients will be collected. If None, uses all Linear layers.

• hessian – Type of Hessian approximation (“Identity”, “eFIM”).

• damping – Damping factor for Hessian inverse (when hessian=”eFIM”)

• device – Device to run computations on

• proj_dim – Projection dimension after Kronecker product (default: 4096). Must be a perfect square. The per-component dimension will be √proj_dim. For example, proj_dim=4096 gives per-component dim of 64.

• offload – Memory management strategy (“none”, “cpu”, “disk”)

• cache_dir – Directory for caching (required when offload=”disk”)

• chunk_size – Chunk size for processing in disk offload

Raises:

ValueError – If proj_dim is not a perfect square.

class dattri.algorithm.block_projected_if.BlockProjectedIFAttributor(task: AttributionTask, layer_names: str | List[str] | None = None, hessian: Literal['Identity', 'eFIM'] = 'eFIM', damping: float | None = None, device: str = 'cpu', sparsifier_kwargs: Dict[str, Any] | None = None, projector_kwargs: Dict[str, Any] | None = None, offload: Literal['none', 'cpu', 'disk'] = 'cpu', cache_dir: str | None = None, chunk_size: int = 16)

Bases: BaseAttributor

Block-Projected Influence Function Attributor.

A general attributor that computes influence scores using a two-stage compression pipeline with projected gradients for efficiency. Uses hooks to capture per-sample gradients and applies random projections.

This is a general implementation that supports arbitrary compressor configurations. For specific methods like LoGra or FactGraSS, use the wrapper classes.

__init__(task: AttributionTask, layer_names: str | List[str] | None = None, hessian: Literal['Identity', 'eFIM'] = 'eFIM', damping: float | None = None, device: str = 'cpu', sparsifier_kwargs: Dict[str, Any] | None = None, projector_kwargs: Dict[str, Any] | None = None, offload: Literal['none', 'cpu', 'disk'] = 'cpu', cache_dir: str | None = None, chunk_size: int = 16) → None

Initialize Block-Projected IF attributor.

Parameters:

• task – Attribution task containing model, loss function, and checkpoints

• layer_names – Names of layers where gradients will be collected. If None, uses all Linear layers. You can check the names using model.named_modules(). HookManager will register hooks to named layers.

• hessian – Type of Hessian approximation (“Identity”, “eFIM”). For “Identity”, the hessian will be taken as the identity matrix. For “eFIM”, the hessian will be computed as the empirical fisher information matrix.

• damping – Damping factor for Hessian inverse (when hessian=”eFIM”)

• device – Device to run computations on

• sparsifier_kwargs – Arguments for sparsifier stage (first stage compression)

• projector_kwargs – Arguments for projector stage (second stage compression). If None, defaults to identity projection (no further compression).

• offload – Memory management strategy (“none”, “cpu”, “disk”), stating the place to offload the gradients. “cpu”: stores gradients on CPU and moves to device when needed. “disk”: stores gradients on disk and moves to device when needed.

• cache_dir – Directory for caching (required when offload=”disk”).

• chunk_size – Chunk size for processing in disk offload.

Raises:

ValueError – If cache_dir is None when offload=”disk”.

attribute(train_dataloader: DataLoader, test_dataloader: DataLoader) → torch.Tensor

Compute influence scores between training and test samples.

Parameters:

• train_dataloader – Training data (can be subset if cache was called)

• test_dataloader – Test data to compute influence for

Returns:

Influence score tensor of shape (num_train, num_test)

cache(full_train_dataloader: DataLoader) → None

Cache gradients and IFVP for the full training dataset.

Parameters:

full_train_dataloader – DataLoader for full training data

compute_ifvp() → None

Compute inverse-Hessian-vector products (IFVP).

Here we use empirical fisher information matrix.

Raises:

ValueError – If layer dimensions are not found.

compute_preconditioners(damping: float | None = None) → None

Compute preconditioners (inverse Hessian) from gradients.

Parameters:

damping – Damping factor for numerical stability

Raises:

ValueError – If layer dimensions are not found.

compute_self_attribution() → Tensor

Compute self-influence scores.

Returns:

Self-influence scores

class dattri.algorithm.factgrass.FactGraSSAttributor(task: AttributionTask, layer_names: str | List[str] | None = None, hessian: Literal['Identity', 'eFIM'] = 'eFIM', damping: float | None = None, device: str = 'cpu', proj_dim: int = 4096, blowup_factor: int = 4, offload: Literal['none', 'cpu', 'disk'] = 'cpu', cache_dir: str | None = None, chunk_size: int = 16)

Bases: BlockProjectedIFAttributor

FactGraSS Attributor.

Factorized Gradient Sketching with Structured Sparsity (FactGraSS) attributor that uses random mask projection for the first stage and SJLT (Sparse Johnson-Lindenstrauss Transform) for the second stage.

This is a follow-up work to LoGra that provides better compression by using a two-stage compression pipeline.

The first stage is factorized with a blowup factor: - If proj_dim=4096 and blowup_factor=4:

• Intermediate dimension = 4096 * 4 = 16384

• Per-component dimension = sqrt(16384) = 128

• After second stage: final dimension = 4096

__init__(task: AttributionTask, layer_names: str | List[str] | None = None, hessian: Literal['Identity', 'eFIM'] = 'eFIM', damping: float | None = None, device: str = 'cpu', proj_dim: int = 4096, blowup_factor: int = 4, offload: Literal['none', 'cpu', 'disk'] = 'cpu', cache_dir: str | None = None, chunk_size: int = 16) → None

Initialize FactGraSS attributor.

Parameters:

• task – Attribution task containing model, loss function, and checkpoints

• layer_names – Names of layers where gradients will be collected. If None, uses all Linear layers.

• hessian – Type of Hessian approximation (“Identity”, “eFIM”).

• damping – Damping factor for Hessian inverse (when hessian=”eFIM”)

• device – Device to run computations on

• proj_dim – Projection dimension after second stage (default: 4096).

• blowup_factor – Multiplier for intermediate dimension after sparsification (default: 4). The intermediate dimension will be proj_dim * blowup_factor, which must be a perfect square for factorized projection. For example, proj_dim=4096 and blowup_factor=4 gives intermediate_dim=16384, per-component dim=128.

• offload – Memory management strategy (“none”, “cpu”, “disk”)

• cache_dir – Directory for caching (required when offload=”disk”)

• chunk_size – Chunk size for processing in disk offload

Raises:

ValueError – If intermediate_dim (proj_dim * blowup_factor) is not a perfect square.

class dattri.algorithm.dvemb.DVEmbAttributor(task: AttributionTask, proj_dim: int | None = None, proj_seed: int = 0, factorization_type: str | None = 'none', layer_names: str | List[str] | None = None)

Bases: object

Data Value Embedding (DVEmb) attributor.

DVEmb captures temporal dependence in training by computing data value embeddings that approximate trajectory-specific leave-one-out influence. This implementation stores embeddings for each epoch separately, allowing for epoch-specific analysis.

__init__(task: AttributionTask, proj_dim: int | None = None, proj_seed: int = 0, factorization_type: str | None = 'none', layer_names: str | List[str] | None = None) → None

Initializes the DVEmb attributor.

Parameters:

• task –

  Task to attribute. Must be an instance of AttributionTask. Note: The checkpoint functionality of the task is not used by DVEmb. The loss function of the task must follow specific formats:

  • If factorization_type is “none”, the loss function should follow the signature of the following example:

    ```python def f(params, data):

    image, label = data loss = nn.CrossEntropyLoss() yhat = torch.func.functional_call(model, params, image) return loss(yhat, label)

    ```.

  • If factorization_type is not “none”, the loss function should follow the signature of the following example:

    ```python def f(model, data, device):

    image, label = data loss = nn.CrossEntropyLoss() yhat = model(image.to(device)) return loss(yhat, label.to(device))

    ```.

• proj_dim – The dimension for projection (if used). Defaults to None, meaning no projection.

• proj_seed – Random seed for projection. Defaults to 0.

• factorization_type – Type of gradient factorization to use. Options are “none” (default), “kronecker” (same as in the paper), or “elementwise” (efficiently projects Kronecker products via factorized elementwise products).

• layer_names – Names of layers where gradients will be collected. If None, uses all Linear layers. You can check the names using model.named_modules(). Hooks will be registered on these layers to collect gradients. Will only be used when factorization_type is not “none”.

Raises:

ValueError – If an unknown factorization type is provided or if no Linear layers are found for factorization or if the loss function format is incorrect.

attribute(test_dataloader: DataLoader, epoch: int | None = None, train_data_indices: List[int] | None = None) → torch.Tensor

Calculates influence scores for a test set for one or all epochs.

Parameters:

• test_dataloader – A dataloader for the test set.

• epoch – Optional. If specified, returns scores using embeddings from that epoch. If None, returns scores based on the sum of embeddings across all epochs.

• train_data_indices – Optional. A list of training sample indices for which to compute influence. If None, computes for all.

Returns:

A tensor of influence scores.

Raises:

• RuntimeError – If embeddings have not been computed by calling cache first, or if a projection dimension was specified but the projector is not initialized.

• ValueError – If embeddings for the specified epoch are not found.

cache(gradients: Dict[int, List[Tensor]] | None = None, learning_rates: Dict[int, List[float]] | None = None, memory_saving: bool | None = True) → None

Computes data value embeddings for each epoch separately.

Parameters:

• gradients – Optional external gradients instead of cached ones (e.g., (epoch -> list of per-sample gradients)).

• learning_rates – Optional external learning rates instead of cached ones (e.g., (epoch -> list of learning rates)).

• memory_saving – If True, cached gradients will be cleared from memory after computation to save space.

Raises:

ValueError – If no gradients are cached before computation, or if NaN values are detected during computation, or if external gradients are provided when using gradient factorization.

cache_gradients(epoch: int, batch_data: tuple[Tensor, ...], indices: Tensor, learning_rate: float) → None

Cache per-sample gradients for a specific epoch and training step.

Parameters:

• epoch – The current epoch number.

• batch_data – A tuple containing the batch of inputs and targets, e.g., (inputs, labels).

• indices – A tensor containing the original indices for the samples in batch_data.

• learning_rate – The learning rate for this step.

clear_cache() → None

Clears cached gradients and factors to free memory.