nvflare.app_common.aggregators package

Submodules

• nvflare.app_common.aggregators.accumulate_model_aggregator module
  • AccumulateWeightedAggregator()
• nvflare.app_common.aggregators.assembler module
  • Assembler
    • Assembler.assemble()
    • Assembler.collection
    • Assembler.get_expected_data_kind()
    • Assembler.get_model_params()
    • Assembler.initialize()
    • Assembler.reset()
• nvflare.app_common.aggregators.collect_and_assemble_aggregator module
  • CollectAndAssembleAggregator
    • CollectAndAssembleAggregator.accept()
    • CollectAndAssembleAggregator.aggregate()
• nvflare.app_common.aggregators.collect_and_assemble_model_aggregator module
  • CollectAndAssembleModelAggregator
    • CollectAndAssembleModelAggregator.accept_model()
    • CollectAndAssembleModelAggregator.aggregate_model()
    • CollectAndAssembleModelAggregator.reset_stats()
• nvflare.app_common.aggregators.dxo_aggregator module
  • DXOAggregator
    • DXOAggregator.accept()
    • DXOAggregator.aggregate()
    • DXOAggregator.reset_aggregation_helper()
• nvflare.app_common.aggregators.dxo_collector module
  • DXOCollector
    • DXOCollector.accept()
    • DXOCollector.aggregate()
    • DXOCollector.reset()
• nvflare.app_common.aggregators.intime_accumulate_model_aggregator module
  • InTimeAccumulateWeightedAggregator
    • InTimeAccumulateWeightedAggregator.accept()
    • InTimeAccumulateWeightedAggregator.aggregate()
    • InTimeAccumulateWeightedAggregator.handle_event()
• nvflare.app_common.aggregators.lazy module
  • LazyAggregator
    • LazyAggregator.accept()
    • LazyAggregator.aggregate()
    • LazyAggregator.reset()
• nvflare.app_common.aggregators.model_aggregator module
  • ModelAggregator
    • ModelAggregator.accept()
    • ModelAggregator.accept_model()
    • ModelAggregator.aggregate()
    • ModelAggregator.aggregate_model()
    • ModelAggregator.error()
    • ModelAggregator.exception()
    • ModelAggregator.handle_event()
    • ModelAggregator.info()
    • ModelAggregator.reset()
    • ModelAggregator.reset_stats()
    • ModelAggregator.warning()
• nvflare.app_common.aggregators.weighted_aggregation_helper module
  • WeightedAggregationHelper
    • WeightedAggregationHelper.add()
    • WeightedAggregationHelper.get_history()
    • WeightedAggregationHelper.get_len()
    • WeightedAggregationHelper.get_result()
    • WeightedAggregationHelper.reset_stats()
  • filter_aggregatable_metrics()

Module contents

AccumulateWeightedAggregator(exclude_vars: str | Dict[str, str] | None = None, aggregation_weights: Dict[str, Any] | Dict[str, Dict[str, Any]] | None = None, expected_data_kind: DataKind | Dict[str, DataKind] = DataKind.WEIGHT_DIFF, weigh_by_local_iter: bool = True)

class CollectAndAssembleAggregator(assembler_id: str)

Bases: Aggregator

Perform collection and flexible assemble aggregation

This is used for methods needing a special assemble mechanism on the client submissions. It first collects all submissions from clients, then delegates the assembling functionality to assembler, which is specific to a particular algorithm. Note that the aggregation in this case is not in-time, since the assembling function may not be arithmetic mean.

Init FLComponent.

The FLComponent is the base class of all FL Components. (executors, controllers, responders, filters, aggregators, and widgets are all FLComponents)

FLComponents have the capability to handle and fire events and contain various methods for logging.

accept(shareable: Shareable, fl_ctx: FLContext) → bool

Accept the shareable submitted by the client.

Parameters:

• shareable – submitted Shareable object

• fl_ctx – FLContext

Returns:

first boolean to indicate if the contribution has been accepted.

aggregate(fl_ctx: FLContext) → Shareable

Perform the aggregation for all the received Shareable from the clients.

Parameters:

fl_ctx – FLContext

Returns:

shareable

class CollectAndAssembleModelAggregator(assembler_id: str)

Bases: ModelAggregator

ModelAggregator adapter for CollectAndAssemble pattern.

This aggregator bridges the gap between FLModel-based workflows (FedAvg) and Assembler-based custom aggregation logic (e.g., K-Means, SVM).

It wraps an Assembler component and: 1. Collects FLModel results from clients 2. Converts them to the format expected by the Assembler 3. Delegates aggregation to the Assembler 4. Returns the aggregated result as FLModel

This enables custom aggregation algorithms to work with the modern FedAvg workflow while maintaining InTime aggregation where possible.

Parameters:

assembler_id – ID of the Assembler component to use for aggregation.

Init FLComponent.

The FLComponent is the base class of all FL Components. (executors, controllers, responders, filters, aggregators, and widgets are all FLComponents)

FLComponents have the capability to handle and fire events and contain various methods for logging.

accept_model(model: FLModel) → None

Accept one FLModel from a client.

Parameters:

model – FLModel received from a client

aggregate_model() → FLModel

Aggregate all accepted models using the Assembler.

Returns:

Aggregated model

Return type:

FLModel

reset_stats() → None

Reset aggregation statistics for next round.

class InTimeAccumulateWeightedAggregator(exclude_vars: str | Dict[str, str] | None = None, aggregation_weights: Dict[str, Any] | Dict[str, Dict[str, Any]] | None = None, expected_data_kind: DataKind | Dict[str, DataKind] = DataKind.WEIGHT_DIFF, weigh_by_local_iter: bool = True)

Bases: Aggregator

Perform accumulated weighted aggregation.

This is often used as the default aggregation method and can be used for FedAvg. It parses the shareable and aggregates the contained DXO(s).

Parameters:

• exclude_vars (Union[str, Dict[str, str]], optional) – Regular expression string to match excluded vars during aggregation. Defaults to None. Can be one string or a dict of {dxo_name: regex strings} corresponding to each aggregated DXO when processing a DXO of DataKind.COLLECTION.

• aggregation_weights (Union[Dict[str, Any], Dict[str, Dict[str, Any]]], optional) – Aggregation weight for each contributor. Defaults to None. Can be one dict of {contrib_name: aggr_weight} or a dict of dicts corresponding to each aggregated DXO when processing a DXO of DataKind.COLLECTION.

• expected_data_kind (Union[DataKind, Dict[str, DataKind]]) – DataKind for DXO. Defaults to DataKind.WEIGHT_DIFF Can be one DataKind or a dict of {dxo_name: DataKind} corresponding to each aggregated DXO when processing a DXO of DataKind.COLLECTION. Only the keys in this dict will be processed.

• 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.

accept(shareable: Shareable, fl_ctx: FLContext) → bool

Store shareable and update aggregator’s internal state

Parameters:

• shareable – information from contributor

• fl_ctx – context provided by workflow

Returns:

The first boolean indicates if this shareable is accepted. The second boolean indicates if aggregate can be called.

aggregate(fl_ctx: FLContext) → Shareable

Called when workflow determines to generate shareable to send back to contributors

Parameters:

fl_ctx (FLContext) – context provided by workflow

Returns:

the weighted mean of accepted shareables from contributors

Return type:

Shareable

handle_event(event_type: str, fl_ctx: FLContext)

Handles events.

Parameters:

• event_type (str) – event type fired by workflow.

• fl_ctx (FLContext) – FLContext information.