federated/utils/rfa.py
def create_robust_measured_process(
    model: tff.learning.Model, iterations: int, v: float, compression: bool = False
) -> tff.templates.MeasuredProcess:
    """Function that creates robust measured process used in federated aggregation.

    Args:
        model (tf.keras.Model): Model to train.\n
        iterations (int): Number of iterations.\n
        v (float): L2 threshold.\n
        compression (bool, optional): If the model should be compressed. Defaults to False.\n

    Returns:
        tff.templates.MeasuredProcess: Returns a `tff.templates.MeasuredProcess` which defines how to aggregate client updates.
    """

    @tff.federated_computation
    def initialize_measured_process() -> tff.FederatedType:
        return tff.federated_value((), tff.SERVER)

    @tff.tf_computation(tf.float32, model, model)
    def calculate_beta(
        alpha: float, server_weights: tf.Tensor, client_weights: tf.Tensor
    ) -> float:

        func = lambda x, y: tf.norm(x - y) ** 2
        return alpha / tf.math.maximum(
            v,
            tf.math.sqrt(
                tf.reduce_sum(
                    tf.nest.map_structure(func, server_weights, client_weights)
                )
            ),
        )

    @tff.federated_computation(
        tff.type_at_server(()),
        tff.type_at_clients(model),
        tff.type_at_clients(tf.float32),
    )
    def weiszfeld_algorithm(
        state: tff.Computation, weights: tf.Tensor, alpha: float
    ) -> tff.templates.MeasuredProcessOutput:
        mean = tff.federated_mean(weights, weight=alpha)
        for _ in range(iterations - 1):
            broadcast_mean = tff.federated_broadcast(mean)
            beta = tff.federated_map(calculate_beta, (alpha, broadcast_mean, weights))
            mean = tff.federated_mean(weights, weight=beta)
        return tff.templates.MeasuredProcessOutput(
            state, mean, tff.federated_value((), tff.SERVER)
        )

    return tff.templates.MeasuredProcess(
        initialize_fn=initialize_measured_process, next_fn=weiszfeld_algorithm
    )
Function that creates robust measured process used in federated aggregation.

    Args:
        model (tf.keras.Model): Model to train.\n
        iterations (int): Number of iterations.\n
        v (float): L2 threshold.\n
        compression (bool, optional): If the model should be compressed. Defaults to False.\n

    Returns:
        tff.templates.MeasuredProcess: Returns a `tff.templates.MeasuredProcess` which defines how to aggregate client updates.
    
Function that creates robust measured process used in federated aggregation.
Function that creates robust measured process used in federated aggregation .
def create_robust_measured_process(
    model: tff.learning.Model, iterations: int, v: float, compression: bool = False
) -> tff.templates.MeasuredProcess:
    @tff.federated_computation
    def initialize_measured_process() -> tff.FederatedType:
        return tff.federated_value((), tff.SERVER)
    @tff.tf_computation(tf.float32, model, model)
    def calculate_beta(
        alpha: float, server_weights: tf.Tensor, client_weights: tf.Tensor
    ) -> float:
        func = lambda x, y: tf.norm(x - y) ** 2
        return alpha / tf.math.maximum(
            v,
            tf.math.sqrt(
                tf.reduce_sum(
                    tf.nest.map_structure(func, server_weights, client_weights)
                )
            ),
        )
    @tff.federated_computation(
        tff.type_at_server(()),
        tff.type_at_clients(model),
        tff.type_at_clients(tf.float32),
    )
    def weiszfeld_algorithm(
        state: tff.Computation, weights: tf.Tensor, alpha: float
    ) -> tff.templates.MeasuredProcessOutput:
        mean = tff.federated_mean(weights, weight=alpha)
        for _ in range(iterations - 1):
            broadcast_mean = tff.federated_broadcast(mean)
            beta = tff.federated_map(calculate_beta, (alpha, broadcast_mean, weights))
            mean = tff.federated_mean(weights, weight=beta)
        return tff.templates.MeasuredProcessOutput(
            state, mean, tff.federated_value((), tff.SERVER)
        )
    return tff.templates.MeasuredProcess(
        initialize_fn=initialize_measured_process, next_fn=weiszfeld_algorithm
    )
def create_robust_measured_process ( model : tff . learning . Model , iterations : int , v : float , compression : bool = False ) -> tff . templates . MeasuredProcess : @ tff . federated_computation def initialize_measured_process ( ) -> tff . FederatedType : return tff . federated_value ( ( ) , tff . SERVER ) @ tff . tf_computation ( tf . float32 , model , model ) def calculate_beta ( alpha : float , server_weights : tf . Tensor , client_weights : tf . Tensor ) -> float : func = lambda x , y : tf . norm ( x - y ) ** 2 return alpha / tf . math . maximum ( v , tf . math . sqrt ( tf . reduce_sum ( tf . nest . map_structure ( func , server_weights , client_weights ) ) ) , ) @ tff . federated_computation ( tff . type_at_server ( ( ) ) , tff . type_at_clients ( model ) , tff . type_at_clients ( tf . float32 ) , ) def weiszfeld_algorithm ( state : tff . Computation , weights : tf . Tensor , alpha : float ) -> tff . templates . MeasuredProcessOutput : mean = tff . federated_mean ( weights , weight = alpha ) for _ in range ( iterations - 1 ) : broadcast_mean = tff . federated_broadcast ( mean ) beta = tff . federated_map ( calculate_beta , ( alpha , broadcast_mean , weights ) ) mean = tff . federated_mean ( weights , weight = beta ) return tff . templates . MeasuredProcessOutput ( state , mean , tff . federated_value ( ( ) , tff . SERVER ) ) return tff . templates . MeasuredProcess ( initialize_fn = initialize_measured_process , next_fn = weiszfeld_algorithm )
Function that creates robust measured process used in federated aggregation.
Function that creates robust measured process used in federated aggregation .
"""Function that creates robust measured process used in federated aggregation.

    Args:
        model (tf.keras.Model): Model to train.\n
        iterations (int): Number of iterations.\n
        v (float): L2 threshold.\n
        compression (bool, optional): If the model should be compressed. Defaults to False.\n

    Returns:
        tff.templates.MeasuredProcess: Returns a `tff.templates.MeasuredProcess` which defines how to aggregate client updates.
    """
Returns a `tff.templates.MeasuredProcess` which defines how to aggregate client updates.
Returns a ` tff . templates . MeasuredProcess ` which defines how to aggregate client updates .
tff.templates.MeasuredProcess
model
tff.learning.Model
Model to train.\n
Model to train . \ n
iterations
int
Number of iterations.\n
Number of iterations . \ n
v
float
None
compression
bool
If the model should be compressed. Defaults to False.\n
If the model should be compressed . Defaults to False . \ n
False