I am using multiple .CSV files to train a multi-calssification federated learning model using TensorFlow federated. and I want to use a custom metric to measure the performance. But I am not sure how to feed the data which are provided in different files into this custom metric. The code is provided below:
dataset_paths = {
'file0': '/content/file0.csv',
'file1': '/content/file1.csv',
'file2': '/content/file2.csv',
}
@tf.function
def add_parsing(dataset):
def parse_dataset(*x):
return OrderedDict([('x', x[1:-1]),('y', x[-1])])
return dataset.map(parse_dataset)
data = tff.simulation.datasets.FilePerUserClientData(
dataset_paths, create_tf_dataset_for_client_fn)
file_ids = sorted(data.file_ids)
data._file_ids = [tf.cast(c, tf.string) for c in data.client_ids]
def create_keras_model():
initializer = tf.keras.initializers.Zeros()
return tf.keras.models.Sequential([
tf.keras.layers.Input(shape=(7,)),
tf.keras.layers.Dense(4, kernel_initializer=initializer),
tf.keras.layers.Softmax(),
])
def model_fn():
keras_model = create_keras_model()
return tff.learning.from_keras_model(
keras_model,
input_spec=train_data[0].element_spec,
loss=tf.keras.losses.SparseCategoricalCrossentropy(),
metrics=[MulticlassTruePositives()]
)
## compile the model
iterative_process = tff.learning.build_federated_averaging_process(
model_fn,
client_optimizer_fn=lambda: tf.keras.optimizers.SGD(learning_rate=0.06),
server_optimizer_fn=lambda: tf.keras.optimizers.SGD(learning_rate=0.9))
The Custom Metric code is:
class MulticlassTruePositives(tf.keras.metrics.Metric):
def __init__(self, name='multiclass_true_positives', **kwargs):
super(MulticlassTruePositives, self).__init__(name=name, **kwargs)
self.true_positives = self.add_weight(name='tp', initializer='zeros')
def update_state(self, y_true, y_pred, sample_weight=None):
y_pred = tf.argmax(y_pred, axis=1)
values = tf.cast(y_true, 'int32') == tf.cast(y_pred, 'int32')
if sample_weight is not None:
sample_weight = tf.cast(sample_weight, 'float32')
values = tf.multiply(values, sample_weight)
self.true_positives.assign_add(tf.reduce_sum(values))
def result(self):
return self.true_positives
def reset_states(self):
# The state of the metric will be reset at the start of each epoch.
self.true_positives.assign(0.)
and when I compile it I got this error:
---------------------------------------------------------------------------
TypeError Traceback (most recent call last)
<ipython-input-25-3023533c0d6e> in <module>()
2 model_fn,
3 client_optimizer_fn=lambda: tf.keras.optimizers.SGD(learning_rate=0.06),
----> 4 server_optimizer_fn=lambda: tf.keras.optimizers.SGD(learning_rate=0.9))
6 frames
/usr/local/lib/python3.7/dist-packages/tensorflow_federated/python/learning/federated_averaging.py in build_federated_averaging_process(model_fn, client_optimizer_fn, server_optimizer_fn, client_weighting, broadcast_process, model_update_aggregation_factory, metrics_aggregator, use_experimental_simulation_loop)
289 broadcast_process=broadcast_process,
290 model_update_aggregation_factory=model_update_aggregation_factory,
--> 291 metrics_aggregator=metrics_aggregator)
292
293 server_state_type = iter_proc.state_type.member
/usr/local/lib/python3.7/dist-packages/tensorflow_federated/python/learning/framework/optimizer_utils.py in build_model_delta_optimizer_process(model_fn, model_to_client_delta_fn, server_optimizer_fn, broadcast_process, model_update_aggregation_factory, metrics_aggregator)
622 broadcast_process=broadcast_process,
623 aggregation_process=aggregation_process,
--> 624 metrics_aggregator=metrics_aggregator)
625
626 return iterative_process.IterativeProcess(
/usr/local/lib/python3.7/dist-packages/tensorflow_federated/python/learning/framework/optimizer_utils.py in _build_one_round_computation(model_fn, server_optimizer_fn, model_to_client_delta_fn, broadcast_process, aggregation_process, metrics_aggregator)
366
367 @computations.tf_computation(dataset_type, model_weights_type)
--> 368 @tf.function
369 def _compute_local_training_and_client_delta(dataset, initial_model_weights):
370 """Performs client local model optimization.
/usr/local/lib/python3.7/dist-packages/tensorflow_federated/python/core/impl/wrappers/computation_wrapper.py in __call__(self, tff_internal_types, *args)
493 parameter_type = _parameter_type(parameters, parameter_types)
494 wrapped_func = self._strategy(
--> 495 fn_to_wrap, fn_name, parameter_type, unpack=None)
496
497 # Copy the __doc__ attribute with the documentation in triple-quotes from
/usr/local/lib/python3.7/dist-packages/tensorflow_federated/python/core/impl/wrappers/computation_wrapper.py in __call__(self, fn_to_wrap, fn_name, parameter_type, unpack)
220 try:
221 args, kwargs = unpack_arguments_fn(packed_args)
--> 222 result = fn_to_wrap(*args, **kwargs)
223 if result is None:
224 raise ComputationReturnedNoneError(fn_to_wrap)
/usr/local/lib/python3.7/dist-packages/tensorflow/python/util/traceback_utils.py in error_handler(*args, **kwargs)
151 except Exception as e:
152 filtered_tb = _process_traceback_frames(e.__traceback__)
--> 153 raise e.with_traceback(filtered_tb) from None
154 finally:
155 del filtered_tb
/usr/local/lib/python3.7/dist-packages/tensorflow/python/framework/func_graph.py in autograph_handler(*args, **kwargs)
1145 except Exception as e: # pylint:disable=broad-except
1146 if hasattr(e, "ag_error_metadata"):
-> 1147 raise e.ag_error_metadata.to_exception(e)
1148 else:
1149 raise
TypeError: in user code:
File "/usr/local/lib/python3.7/dist-packages/tensorflow_federated/python/learning/framework/optimizer_utils.py", line 382, in _compute_local_training_and_client_delta *
client_output = client_delta_fn(dataset, initial_model_weights)
File "/usr/local/lib/python3.7/dist-packages/tensorflow_federated/python/learning/federated_averaging.py", line 113, in reduce_fn *
output = model.forward_pass(batch, training=True)
File "/usr/local/lib/python3.7/dist-packages/tensorflow_federated/python/learning/framework/dataset_reduce.py", line 33, in _dataset_reduce_fn *
return dataset.reduce(initial_state=initial_state_fn(), reduce_func=reduce_fn)
File "/usr/local/lib/python3.7/dist-packages/tensorflow_federated/python/learning/keras_utils.py", line 455, in forward_pass *
return self._forward_pass(batch_input, training=training)
File "/usr/local/lib/python3.7/dist-packages/tensorflow_federated/python/learning/keras_utils.py", line 443, in _forward_pass *
metric.update_state(y_true=y_true, y_pred=predictions)
File "<ipython-input-17-69c4467a9ad6>", line 20, in decorated *
update_op = update_state_fn(*args, **kwargs)
File "<ipython-input-17-69c4467a9ad6>", line 12, in update_state *
self.true_positives.assign_add(tf.reduce_sum(values))
TypeError: Value passed to parameter 'input' has DataType bool not in list of allowed values: float32, float64, int32, uint8, int16, int8, complex64, int64, qint8, quint8, qint32, bfloat16, uint16, complex128, float16, uint32, uint64
I think it is because of passing the whole file to the model. how can I fix it? any help will be appreciated.
CodePudding user response：
The problem is this line:
values = tf.cast(y_true, 'int32') == tf.cast(y_pred, 'int32')
It is generating a boolean and tf.reduce_sum cannot be applied to it:
y_true = [0, 1, 0, 0]
y_pred = [-18.6, 0.51, 2.94, -12.8]
values = tf.cast(y_true, 'int32') == tf.cast(y_pred, 'int32')
print(values)
# tf.Tensor([False False False False], shape=(4,), dtype=bool)
tf.reduce_sum(values) # throws error