nvflare.app_common.aggregators.weighted_aggregation_helper module

class AggregationStatsKey

Bases: object

Keys of the per-round aggregation stats dict produced by WeightedAggregationHelper and aggregators.

ACCEPTED_CONTRIBUTIONS = 'accepted_contributions'

CONTRIBUTORS = 'contributors'

FULLY_MATCHED_KEYS = 'fully_matched_keys'

KEYS_AGGREGATED = 'keys_aggregated'

KEYS_SEEN = 'keys_seen'

PARTIALLY_MATCHED_KEYS = 'partially_matched_keys'

ROUND = 'round'

SKIPPED_KEYS = 'skipped_keys'

class WeightedAggregationHelper(exclude_vars: str | None = None, weigh_by_local_iter: bool = True)

Bases: object

Perform weighted aggregation.

Parameters:

• exclude_vars (str, optional) – regex string to match excluded vars during aggregation. Defaults to None.

• weigh_by_local_iter (bool, optional) – Whether to weight the contributions by the number of iterations performed in local training in the current round. Defaults to True. Setting it to False can be useful in applications such as homomorphic encryption to reduce the number of computations on encrypted ciphertext. The aggregated sum will still be divided by the provided weights and aggregation_weights for the resulting weighted sum to be valid.

add(data, weight, contributor_name, contribution_round)

Compute weighted sum and sum of weights.

get_aggregation_stats() → dict

Return key-matching stats of the contributions accumulated so far in this round.

A key is “fully matched” when every accepted contribution included it, “partially matched” when only some contributions included it, and “skipped” when it was excluded by exclude_vars.

get_history()

get_len()

get_result()

Divide weighted sum by sum of weights.

reset_stats()

compute_key_match_stats(contributions: Dict[str, Any]) → dict

Compute key-match aggregation stats from each contributor’s param keys.

Parameters:

contributions – dict of contributor name -> iterable of the param keys that contributor supplied.

Returns:

An aggregation stats dict (AggregationStatsKey fields). Since no exclusion is applied here, keys seen equals keys aggregated and skipped keys is 0.

filter_aggregatable_metrics(metrics: Dict[str, Any] | None, warn_skipped: Callable[[str, str], None] | None = None, warned_metric_keys: Set[str] | None = None) → Dict[str, Any]

Return metric entries that support weighted aggregation.

Note

Boolean metric values are included and aggregate as binary rates.

Parameters:

• metrics – Dict of metric name -> value.

• warn_skipped – Optional callback invoked as warn_skipped(key, type_name) for skipped metrics.

• warned_metric_keys – Optional set of keys already warned about. If provided, warnings are emitted at most once per key and newly warned keys are added to this set.