Implement Bulyan (#1817)

doc/source/changelog.md

@@ -4,7 +4,11 @@
 
 ### What's new?
 
--  **Add new** `FedTrimmedAvg` **strategy** ([#1769](https://github.com/adap/flower/pull/1769), [#1853](https://github.com/adap/flower/pull/1853))
+- **Add new** `Bulyan` **strategy** ([#1817](https://github.com/adap/flower/pull/1817), [#1891](https://github.com/adap/flower/pull/1891))
+
+  The new `Bulyan` strategy implements Bulyan by [El Mhamdi et al., 2018](https://arxiv.org/abs/1802.07927)
+
+- **Add new** `FedTrimmedAvg` **strategy** ([#1769](https://github.com/adap/flower/pull/1769), [#1853](https://github.com/adap/flower/pull/1853))
 
   The new `FedTrimmedAvg` strategy implements Trimmed Mean by [Dong Yin, 2018](https://arxiv.org/abs/1803.01498)
 

src/py/flwr/server/strategy/aggregate.py

@@ -16,7 +16,7 @@
 
 
 from functools import reduce
-from typing import List, Tuple
+from typing import Any, Callable, List, Tuple
 
 import numpy as np
 
@@ -56,7 +56,7 @@ def aggregate_median(results: List[Tuple[NDArrays, int]]) -> NDArrays:
 def aggregate_krum(
     results: List[Tuple[NDArrays, int]], num_malicious: int, to_keep: int
 ) -> NDArrays:
-    """Choose one parameter vector according to the Krum fucntion.
+    """Choose one parameter vector according to the Krum function.
 
     If to_keep is not None, then MultiKrum is applied.
     """
@@ -91,6 +91,88 @@ def aggregate_krum(
     return weights[np.argmin(scores)]
 
 
+# pylint: disable=too-many-locals
+def aggregate_bulyan(
+    results: List[Tuple[NDArrays, int]],
+    num_malicious: int,
+    aggregation_rule: Callable,  # type: ignore
+    **aggregation_rule_kwargs: Any,
+) -> NDArrays:
+    """Perform Bulyan aggregation.
+
+    Parameters
+    ----------
+    results: List[Tuple[NDArrays, int]]
+        Weights and number of samples for each of the client.
+    num_malicious: int
+        The maximum number of malicious clients.
+    aggregation_rule: Callable
+        Byzantine resilient aggregation rule used as the first step of the Bulyan
+    aggregation_rule_kwargs: Any
+        The arguments to the aggregation rule.
+
+    Returns
+    -------
+    aggregated_parameters: NDArrays
+        Aggregated parameters according to the Bulyan strategy.
+    """
+    byzantine_resilient_single_ret_model_aggregation = [aggregate_krum]
+    # also GeoMed (but not implemented yet)
+    byzantine_resilient_many_return_models_aggregation = []  # type: ignore
+    # Brute, Medoid (but not implemented yet)
+
+    num_clients = len(results)
+    if num_clients < 4 * num_malicious + 3:
+        raise ValueError(
+            "The Bulyan aggregation requires then number of clients to be greater or "
+            "equal to the 4 * num_malicious + 3. This is the assumption of this method."
+            "It is needed to ensure that the method reduces the attacker's leeway to "
+            "the one proved in the paper."
+        )
+    selected_models_set: List[Tuple[NDArrays, int]] = []
+
+    # Create a list of weights and ignore the number of examples
+    weights = [weights for weights, _ in results]
+
+    theta = len(weights) - 2 * num_malicious
+    beta = theta - 2 * num_malicious
+
+    for _ in range(theta):
+        best_model = aggregation_rule(
+            results=results, num_malicious=num_malicious, **aggregation_rule_kwargs
+        )
+        list_of_weights = [weights for weights, num_samples in results]
+        # This group gives exact result
+        if aggregation_rule in byzantine_resilient_single_ret_model_aggregation:
+            best_idx = _find_reference_weights(best_model, list_of_weights)
+        # This group requires finding the closest model to the returned one
+        # (weights distance wise)
+        elif aggregation_rule in byzantine_resilient_many_return_models_aggregation:
+            # when different aggregation strategies available
+            # write a function to find the closest model
+            best_idx = 0
+        else:
+            raise ValueError(
+                "The given aggregation rule is not added as Byzantine resilient. "
+                "Please choose from Byzantine resilient rules."
+            )
+
+        selected_models_set.append(results[best_idx])
+
+        # remove idx from tracker and weights_results
+        results.pop(best_idx)
+
+    # Compute median parameter vector across selected_models_set
+    median_vect = aggregate_median(selected_models_set)
+
+    # Take the averaged beta parameters of the closest distance to the median
+    # (coordinate-wise)
+    parameters_aggregated = _aggregate_n_closest_weights(
+        median_vect, selected_models_set, beta_closest=beta
+    )
+    return parameters_aggregated
+
+
 def weighted_loss_avg(results: List[Tuple[int, float]]) -> float:
     """Aggregate evaluation results obtained from multiple clients."""
     num_total_evaluation_examples = sum([num_examples for num_examples, _ in results])
@@ -170,3 +252,86 @@ def aggregate_trimmed_avg(
     ]
 
     return trimmed_w
+
+
+def _check_weights_equality(weights1: NDArrays, weights2: NDArrays) -> bool:
+    """Check if weights are the same."""
+    if len(weights1) != len(weights2):
+        return False
+    return all(
+        np.array_equal(layer_weights1, layer_weights2)
+        for layer_weights1, layer_weights2 in zip(weights1, weights2)
+    )
+
+
+def _find_reference_weights(
+    reference_weights: NDArrays, list_of_weights: List[NDArrays]
+) -> int:
+    """Loop through the `list_of_weights` to find the reference weights, raise
+    Error if not found.
+
+    Parameters
+    ----------
+    reference_weights: NDArrays
+        Weights that will be searched for.
+    list_of_weights: List[NDArrays]
+        List of weights that will be searched through.
+
+    Returns
+    -------
+    index: int
+        The index of `reference_weights` in the `list_of_weights`.
+
+    Raises
+    ------
+    ValueError
+        If `reference_weights` is not found in `list_of_weights`.
+    """
+    for idx, weights in enumerate(list_of_weights):
+        if _check_weights_equality(reference_weights, weights):
+            return idx
+    raise ValueError("The reference weights not found in list_of_weights.")
+
+
+def _aggregate_n_closest_weights(
+    reference_weights: NDArrays, results: List[Tuple[NDArrays, int]], beta_closest: int
+) -> NDArrays:
+    """It calculates element-wise mean of the `N` the closest values.
+
+    Note, each i-th coordinate of the result weight is the average of the beta_closest
+    -ith coordinates to the reference weights
+
+
+    Parameters
+    ----------
+    reference_weights: NDArrays
+        The weights from which the distances will be computed
+    results: List[Tuple[NDArrays, int]]
+        The weights from models
+    beta_closest: int
+        The number of the closest distance weights that will be averaged
+    Returns
+    -------
+    aggregated_weights: NDArrays
+        Averaged (element-wise) beta weights that have the closest distance to
+         reference weights
+    """
+    list_of_weights = [weights for weights, num_examples in results]
+    aggregated_weights = []
+
+    for layer_id, layer_weights in enumerate(reference_weights):
+        other_weights_layer_list = []
+        for other_w in list_of_weights:
+            other_weights_layer = other_w[layer_id]
+            other_weights_layer_list.append(other_weights_layer)
+        other_weights_layer_np = np.array(other_weights_layer_list)
+        diff_np = np.abs(layer_weights - other_weights_layer_np)
+        # Create indices of the smallest differences
+        # We do not need  the exact order but just the beta closest weights
+        # therefore np.argpartition is used instead of np.argsort
+        indices = np.argpartition(diff_np, kth=beta_closest - 1, axis=0)
+        # Take the weights (coordinate-wise) corresponding to the beta of the
+        # closest distances
+        beta_closest_weights = np.take_along_axis(other_weights_layer_np, indices=indices, axis=0)[:beta_closest]  # type: ignore[no-untyped-call]
+        aggregated_weights.append(np.mean(beta_closest_weights, axis=0))
+    return aggregated_weights

src/py/flwr/server/strategy/aggregate_test.py

@@ -19,7 +19,13 @@
 
 import numpy as np
 
-from .aggregate import aggregate, weighted_loss_avg
+from .aggregate import (
+    _aggregate_n_closest_weights,
+    _check_weights_equality,
+    _find_reference_weights,
+    aggregate,
+    weighted_loss_avg,
+)
 
 
 def test_aggregate() -> None:
@@ -65,3 +71,72 @@ def test_weighted_loss_avg_multiple_values() -> None:
 
     # Assert
     assert expected == actual
+
+
+def test_check_weights_equality_true() -> None:
+    """Check weights equality - the same weights."""
+    weights1 = [np.array([1, 2]), np.array([[1, 2], [3, 4]])]
+    weights2 = [np.array([1, 2]), np.array([[1, 2], [3, 4]])]
+    results = _check_weights_equality(weights1, weights2)
+    expected = True
+    assert expected == results
+
+
+def test_check_weights_equality_numeric_false() -> None:
+    """Check weights equality - different weights, same length."""
+    weights1 = [np.array([1, 2]), np.array([[1, 2], [3, 4]])]
+    weights2 = [np.array([2, 2]), np.array([[1, 2], [3, 4]])]
+    results = _check_weights_equality(weights1, weights2)
+    expected = False
+    assert expected == results
+
+
+def test_check_weights_equality_various_length_false() -> None:
+    """Check weights equality - the same first layer weights, different length."""
+    weights1 = [np.array([1, 2]), np.array([[1, 2], [3, 4]])]
+    weights2 = [np.array([1, 2])]
+    results = _check_weights_equality(weights1, weights2)
+    expected = False
+    assert expected == results
+
+
+def test_find_reference_weights() -> None:
+    """Check if the finding weights from list of weigths work."""
+    reference_weights = [np.array([1, 2]), np.array([[1, 2], [3, 4]])]
+    list_of_weights = [
+        [np.array([2, 2]), np.array([[1, 2], [3, 4]])],
+        [np.array([3, 2]), np.array([[1, 2], [3, 4]])],
+        [np.array([3, 2]), np.array([[1, 2], [10, 4]])],
+        [np.array([1, 2]), np.array([[1, 2], [3, 4]])],
+    ]
+
+    result = _find_reference_weights(reference_weights, list_of_weights)
+
+    expected = 3
+    assert result == expected
+
+
+def test_aggregate_n_closest_weights_mean() -> None:
+    """Check if aggregation of n closest weights to the reference works."""
+    beta_closest = 2
+    reference_weights = [np.array([1, 2]), np.array([[1, 2], [3, 4]])]
+
+    list_of_weights = [
+        [np.array([1, 2]), np.array([[1, 2], [3, 4]])],
+        [np.array([1.1, 2.1]), np.array([[1.1, 2.1], [3.1, 4.1]])],
+        [np.array([1.2, 2.2]), np.array([[1.2, 2.2], [3.2, 4.2]])],
+        [np.array([1.3, 2.3]), np.array([[0.9, 2.5], [3.4, 3.8]])],
+    ]
+    list_of_weights_and_samples = [(weights, 0) for weights in list_of_weights]
+
+    beta_closest_weights = _aggregate_n_closest_weights(
+        reference_weights, list_of_weights_and_samples, beta_closest=beta_closest
+    )
+    expected_averaged = [np.array([1.05, 2.05]), np.array([[0.95, 2.05], [3.05, 4.05]])]
+
+    assert all(
+        (
+            np.array_equal(expected, result)
+            for expected, result in zip(expected_averaged, beta_closest_weights)
+        )
+    )