deel.lipdp.layers module

AddBias

Bases: Layer

Adds a bias to the input.

Remark: the euclidean norm of the bias must be bounded in advance. Note that this is the euclidean norm of the whole bias vector, not the norm of each element of the bias vector.

Warning: beware zero gradients outside the ball of norm norm_max. In the future, we might choose a smoother projection on the ball to ensure that the gradient remains non zero outside the ball.

Source code in deel/lipdp/layers.py

class AddBias(tf.keras.layers.Layer):
    """Adds a bias to the input.

    Remark: the euclidean norm of the bias must be bounded in advance.
    Note that this is the euclidean norm of the whole bias vector, not
    the norm of each element of the bias vector.

    Warning: beware zero gradients outside the ball of norm norm_max.
    In the future, we might choose a smoother projection on the ball to ensure
    that the gradient remains non zero outside the ball.
    """

    def __init__(self, norm_max, **kwargs):
        super().__init__(**kwargs)
        self.norm_max = tf.convert_to_tensor(norm_max)

    def build(self, input_shape):
        self.bias = self.add_weight(
            name="bias",
            shape=(input_shape[-1],),
            initializer="zeros",
            trainable=True,
        )

    def call(self, inputs, **kwargs):
        # parametrize the bias so it belongs to a ball of norm norm_max.
        bias = tf.convert_to_tensor(
            tf.clip_by_norm(self.bias, self.norm_max)
        )  # 1-Lipschitz operation.
        return inputs + bias

DPLayer

Wrapper for created differentially private layers, instanciates abstract methods use for computing the bounds of the gradient relatively to the parameters and to the input.

Source code in deel/lipdp/layers.py

class DPLayer:
    """
    Wrapper for created differentially private layers, instanciates abstract methods
    use for computing the bounds of the gradient relatively to the parameters and to the
    input.
    """

    @abstractmethod
    def backpropagate_params(self, input_bound, gradient_bound):
        """Corresponds to the Lipschitz constant of the output wrt the parameters,
            i.e. the norm of the Jacobian of the output wrt the parameters times the norm of the cotangeant vector.

        Args:
            input_bound: Maximum norm of input.
            gradient_bound: Maximum norm of gradients (co-tangent vector)

        Returns:
            Maximum norm of tangent vector."""
        pass

    @abstractmethod
    def backpropagate_inputs(self, input_bound, gradient_bound):
        """Applies the dilatation of the cotangeant vector norm (upstream gradient) by the Jacobian,
            i.e. multiply by the Lipschitz constant of the output wrt input.

        Args:
            input_bound: Maximum norm of input.
            gradient_bound: Maximum norm of gradients (co-tangent vector)

        Returns:
            Maximum norm of tangent vector.
        """
        pass

    @abstractmethod
    def propagate_inputs(self, input_bound):
        """Maximum norm of output of element.

        Remark: when the layer is linear, this coincides with its Lipschitz constant * input_bound.
        """
        pass

    @abstractmethod
    def has_parameters(self):
        pass

backpropagate_inputs(input_bound, gradient_bound) abstractmethod

Applies the dilatation of the cotangeant vector norm (upstream gradient) by the Jacobian, i.e. multiply by the Lipschitz constant of the output wrt input.

Parameters:

Name	Type	Description	Default
input_bound		Maximum norm of input.	required
gradient_bound		Maximum norm of gradients (co-tangent vector)	required

Returns:

Type	Description
	Maximum norm of tangent vector.

Source code in deel/lipdp/layers.py

@abstractmethod
def backpropagate_inputs(self, input_bound, gradient_bound):
    """Applies the dilatation of the cotangeant vector norm (upstream gradient) by the Jacobian,
        i.e. multiply by the Lipschitz constant of the output wrt input.

    Args:
        input_bound: Maximum norm of input.
        gradient_bound: Maximum norm of gradients (co-tangent vector)

    Returns:
        Maximum norm of tangent vector.
    """
    pass

backpropagate_params(input_bound, gradient_bound) abstractmethod

Corresponds to the Lipschitz constant of the output wrt the parameters, i.e. the norm of the Jacobian of the output wrt the parameters times the norm of the cotangeant vector.

Parameters:

Name	Type	Description	Default
input_bound		Maximum norm of input.	required
gradient_bound		Maximum norm of gradients (co-tangent vector)	required

Returns:

Type	Description
	Maximum norm of tangent vector.

Source code in deel/lipdp/layers.py

@abstractmethod
def backpropagate_params(self, input_bound, gradient_bound):
    """Corresponds to the Lipschitz constant of the output wrt the parameters,
        i.e. the norm of the Jacobian of the output wrt the parameters times the norm of the cotangeant vector.

    Args:
        input_bound: Maximum norm of input.
        gradient_bound: Maximum norm of gradients (co-tangent vector)

    Returns:
        Maximum norm of tangent vector."""
    pass

propagate_inputs(input_bound) abstractmethod

Maximum norm of output of element.

Remark: when the layer is linear, this coincides with its Lipschitz constant * input_bound.

Source code in deel/lipdp/layers.py

@abstractmethod
def propagate_inputs(self, input_bound):
    """Maximum norm of output of element.

    Remark: when the layer is linear, this coincides with its Lipschitz constant * input_bound.
    """
    pass

DP_AddBias

Bases: AddBias, DPLayer

Adds a bias to the input.

The bias is projected on the ball of norm norm_max during training. The projection on the ball is a 1-Lipschitz function, since the ball is convex.

Source code in deel/lipdp/layers.py

class DP_AddBias(AddBias, DPLayer):
    """Adds a bias to the input.

    The bias is projected on the ball of norm `norm_max` during training.
    The projection on the ball is a 1-Lipschitz function, since the ball
    is convex.
    """

    def __init__(self, *args, nm_coef=1, **kwargs):
        super().__init__(*args, **kwargs)
        self.nm_coef = nm_coef

    def backpropagate_params(self, input_bound, gradient_bound):
        return gradient_bound  # clipping is a 1-Lipschitz operation.

    def backpropagate_inputs(self, input_bound, gradient_bound):
        return 1 * gradient_bound  # adding is a 1-Lipschitz operation.

    def propagate_inputs(self, input_bound):
        return input_bound + self.norm_max

    def has_parameters(self):
        return True

DP_BoundedInput

Bases: Layer, DPLayer

Input layer that clips the input to a given norm.

Remark: every pipeline should start with this layer.

Attributes:

Name	Type	Description
upper_bound		Maximum norm of the input.
enforce_clipping		If True (default), the input is clipped to the given norm.

Source code in deel/lipdp/layers.py

class DP_BoundedInput(tf.keras.layers.Layer, DPLayer):
    """Input layer that clips the input to a given norm.

    Remark: every pipeline should start with this layer.

    Attributes:
        upper_bound: Maximum norm of the input.
        enforce_clipping: If True (default), the input is clipped to the given norm.
    """

    def __init__(self, *args, upper_bound, enforce_clipping=True, **kwargs):
        super().__init__(*args, **kwargs)
        self.upper_bound = tf.convert_to_tensor(upper_bound)
        self.enforce_clipping = enforce_clipping

    def call(self, x, *args, **kwargs):
        if self.enforce_clipping:
            axes = list(range(1, len(x.shape)))
            x = tf.clip_by_norm(x, self.upper_bound, axes=axes)
        return x

    def backpropagate_params(self, input_bound, gradient_bound):
        raise ValueError("InputLayer doesn't have parameters")

    def backpropagate_inputs(self, input_bound, gradient_bound):
        return 1 * gradient_bound

    def propagate_inputs(self, input_bound):
        if input_bound is None:
            return self.upper_bound
        return tf.math.minimum(self.upper_bound, input_bound)

    def has_parameters(self):
        return False

DP_ClipGradient

Bases: Layer, DPLayer

Clips the gradient during the backward pass.

Behaves like identity function during the forward pass. The clipping is done automatically during the backward pass.

Attributes:

Name	Type	Description
clip_value	float	The maximum norm of the gradient allowed. Only declare this variable if you plan on using the "fixed" clipping mode. Otherwise it will be updated automatically.
mode	str	The mode of clipping. Either "fixed" or "dynamic". Default is "fixed".

Warning : The mode "dynamic" needs to be used along a callback that updates the clipping value.

Source code in deel/lipdp/layers.py

class DP_ClipGradient(tf.keras.layers.Layer, DPLayer):
    """Clips the gradient during the backward pass.

    Behaves like identity function during the forward pass.
    The clipping is done automatically during the backward pass.

    Attributes:
        clip_value (float, optional):
                            The maximum norm of the gradient allowed. Only
                            declare this variable if you plan on using the "fixed" clipping mode.
                            Otherwise it will be updated automatically.
        mode (str): The mode of clipping. Either "fixed" or "dynamic". Default is "fixed".

    Warning : The mode "dynamic" needs to be used along a callback that updates the clipping value.
    """

    def __init__(self, clip_value, mode="fixed", *args, **kwargs):
        super().__init__(*args, **kwargs)

        self._dynamic_dp_dict = {}  # to be filled by the callback

        assert mode in ["fixed", "dynamic"]
        self.mode = mode

        assert clip_value is None or clip_value >= 0, "clip_value must be positive"
        if mode == "fixed":
            assert (
                clip_value is not None
            ), "clip_value must be declared when using the fixed mode"

        if clip_value is None:
            clip_value = (
                0.0  # Change type back to float in case clip_value needs to be updated
            )

        self.clip_value = tf.Variable(clip_value, trainable=False, dtype=tf.float32)

    def update_clipping_value(self, new_clip_value):
        print("Update clipping value to : ", float(new_clip_value.numpy()))
        self.clip_value.assign(new_clip_value)

    def call(self, inputs, *args, **kwargs):
        batch_size = tf.convert_to_tensor(tf.cast(tf.shape(inputs)[0], tf.float32))
        # the clipping is done elementwise
        # since REDUCTION=SUM_OVER_BATCH_SIZE, we need to divide by batch_size
        # to get the correct norm.
        # this makes the clipping independent of the batch size.
        elementwise_clip_value = self.clip_value.value() / batch_size
        return clip_gradient(inputs, elementwise_clip_value)

    def backpropagate_params(self, input_bound, gradient_bound):
        raise ValueError("ClipGradient doesn't have parameters")

    def backpropagate_inputs(self, input_bound, gradient_bound):
        return tf.math.minimum(gradient_bound, self.clip_value)

    def propagate_inputs(self, input_bound):
        return input_bound

    def has_parameters(self):
        return False

DP_MaxPool2D

Bases: MaxPool2D, DPLayer

Max pooling layer that preserves the gradient norm.

Parameters:

Name	Type	Description	Default
layer_cls		Class of the layer to wrap.	required

Returns:

Type	Description
	A differentially private layer that doesn't have parameters.

Source code in deel/lipdp/layers.py

class DP_MaxPool2D(tf.keras.layers.MaxPool2D, DPLayer):
    """Max pooling layer that preserves the gradient norm.

    Args:
        layer_cls: Class of the layer to wrap.

    Returns:
        A differentially private layer that doesn't have parameters.
    """

    def __init__(self, *args, **kwargs):
        super().__init__(*args, **kwargs)
        assert (
            self.strides is None or self.strides == self.pool_size
        ), "Ensure that strides == pool_size, otherwise it is not 1-Lipschitz."

    def backpropagate_params(self, input_bound, gradient_bound):
        raise ValueError("Layer doesn't have parameters")

    def backpropagate_inputs(self, input_bound, gradient_bound):
        return 1 * gradient_bound

    def propagate_inputs(self, input_bound):
        return input_bound

    def has_parameters(self):
        return False

DP_ScaledL2NormPooling2D

Bases: ScaledL2NormPooling2D, DPLayer

Max pooling layer that preserves the gradient norm.

Parameters:

Name	Type	Description	Default
layer_cls		Class of the layer to wrap.	required

Returns:

Type	Description
	A differentially private layer that doesn't have parameters.

Source code in deel/lipdp/layers.py

class DP_ScaledL2NormPooling2D(deel.lip.layers.ScaledL2NormPooling2D, DPLayer):
    """Max pooling layer that preserves the gradient norm.

    Args:
        layer_cls: Class of the layer to wrap.

    Returns:
        A differentially private layer that doesn't have parameters.
    """

    def __init__(self, *args, **kwargs):
        super().__init__(*args, **kwargs)
        assert (
            self.strides is None or self.strides == self.pool_size
        ), "Ensure that strides == pool_size, otherwise it is not 1-Lipschitz."

    def backpropagate_params(self, input_bound, gradient_bound):
        raise ValueError("Layer doesn't have parameters")

    def backpropagate_inputs(self, input_bound, gradient_bound):
        return 1 * gradient_bound

    def propagate_inputs(self, input_bound):
        return input_bound

    def has_parameters(self):
        return False

DP_WrappedResidual

Bases: Layer, DPLayer

Source code in deel/lipdp/layers.py

class DP_WrappedResidual(tf.keras.layers.Layer, DPLayer):
    def __init__(self, block, *args, **kwargs):
        super().__init__(*args, **kwargs)
        self.block = block

    def call(self, inputs, *args, **kwargs):
        assert len(inputs) == 2
        i1, i2 = inputs
        i2 = self.block(i2, *args, **kwargs)
        return i1, i2

    def backpropagate_params(self, input_bound, gradient_bound):
        assert len(input_bound) == 2
        assert len(gradient_bound) == 2
        _, i2 = input_bound
        _, g2 = gradient_bound
        g2 = self.block.backpropagate_params(i2, g2)
        return g2

    def backpropagate_inputs(self, input_bound, gradient_bound):
        assert len(input_bound) == 2
        assert len(gradient_bound) == 2
        _, i2 = input_bound
        g1, g2 = gradient_bound
        g2 = self.block.backpropagate_inputs(i2, g2)
        return g1, g2

    def propagate_inputs(self, input_bound):
        assert len(input_bound) == 2
        i1, i2 = input_bound
        i2 = self.block.propagate_inputs(i2)
        return i1, i2

    def has_parameters(self):
        return self.block.has_parameters()

    @property
    def nm_coef(self):
        """Returns the norm multiplier coefficient of the layer.

        Remark: this is a property to mimic the behavior of an attribute.
        """
        return self.block.nm_coef

nm_coef property

Returns the norm multiplier coefficient of the layer.

Remark: this is a property to mimic the behavior of an attribute.

DP_GNP_Factory(layer_cls)

Factory for creating differentially private gradient norm preserving layers that don't have parameters.

Remark: the layer is assumed to be GNP. This means that the gradient norm is preserved by the layer (i.e its Jacobian norm is 1). Please ensure that the layer is GNP before using this factory.

Parameters:

Name	Type	Description	Default
layer_cls		Class of the layer to wrap.	required

Returns:

Type	Description
	A differentially private layer that doesn't have parameters.

Source code in deel/lipdp/layers.py

def DP_GNP_Factory(layer_cls):
    """Factory for creating differentially private gradient norm preserving layers that don't have parameters.

    Remark: the layer is assumed to be GNP.
    This means that the gradient norm is preserved by the layer (i.e its Jacobian norm is 1).
    Please ensure that the layer is GNP before using this factory.

    Args:
        layer_cls: Class of the layer to wrap.

    Returns:
        A differentially private layer that doesn't have parameters.
    """

    class DP_GNP(layer_cls, DPLayer):
        def __init__(self, *args, **kwargs):
            super().__init__(*args, **kwargs)

        def backpropagate_params(self, input_bound, gradient_bound):
            raise ValueError("Layer doesn't have parameters")

        def backpropagate_inputs(self, input_bound, gradient_bound):
            return 1 * gradient_bound

        def propagate_inputs(self, input_bound):
            return input_bound

        def has_parameters(self):
            return False

    DP_GNP.__name__ = f"DP_{layer_cls.__name__}"
    return DP_GNP

clip_gradient(x, clip_value)

Clips the gradient during the backward pass.

Behave like identity function during the forward pass.

Source code in deel/lipdp/layers.py

@tf.custom_gradient
def clip_gradient(x, clip_value):
    """Clips the gradient during the backward pass.

    Behave like identity function during the forward pass.
    """

    def grad_fn(dy):
        # clip by norm each row
        axes = list(range(1, len(dy.shape)))
        clipped_dy = tf.clip_by_norm(dy, clip_value, axes=axes)
        return clipped_dy, None  # No gradient for clip_value

    return x, grad_fn

make_residuals(merge_policy, wrapped_layers)

Returns a list of layers that implement a residual block.

Parameters:

Name	Type	Description	Default
merge_policy		either "add" or "1-lip-add".	required
wrapped_layers		a list of layers that will be wrapped in residual blocks.	required

Returns:

Type	Description
	A list of layers that implement a residual block.

Source code in deel/lipdp/layers.py

def make_residuals(merge_policy, wrapped_layers):
    """Returns a list of layers that implement a residual block.

    Args:
        merge_policy: either "add" or "1-lip-add".
        wrapped_layers: a list of layers that will be wrapped in residual blocks.

    Returns:
        A list of layers that implement a residual block.
    """
    layers = [DP_SplitResidual()]

    for layer in wrapped_layers:
        residual_block = DP_WrappedResidual(layer)
        layers.append(residual_block)

    layers.append(DP_MergeResidual(merge_policy))

    return layers