custom_aggregators.ipynb

Kernel: Python 3

Copyright 2021 The TensorFlow Federated Authors.

#@title Licensed under the Apache License, Version 2.0 (the "License");
# you may not use this file except in compliance with the License.
# You may obtain a copy of the License at
#
# https://www.apache.org/licenses/LICENSE-2.0
#
# Unless required by applicable law or agreed to in writing, software
# distributed under the License is distributed on an "AS IS" BASIS,
# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
# See the License for the specific language governing permissions and
# limitations under the License.

Implementación de agregaciones personalizadas

Ver en TensorFlow.org

Ejecutar en Google Colab

Ver fuente en GitHub

Descargar bloc de notas

En este tutorial, explicamos los principios de diseño detrás del módulo tff.aggregators y las mejores prácticas para la implementación de agregación personalizada de valores de clientes a servidores.

Requisitos previos. En este tutorial se presupone que el lector ya está familiarizado con los conceptos básicos de núcleo federado, como las ubicaciones (tff.SERVER, tff.CLIENTS), la manera en que TFF representa los cálculos (tff.tf_computation, tff.federated_computation) y sus firmas de tipo.

#@test {"skip": true}
!pip install --quiet --upgrade tensorflow-federated

Resumen del diseño

En TFF, la "agregación" se refiere al movimiento de conjuntos de valores en tff.CLIENTS para producir un valor agregado del mismo tipo en tff.SERVER. Es decir, no es necesario que cada valor de cliente individual esté disponible. Por ejemplo, en el aprendizaje federado las actualizaciones del modelo del cliente se promedian para obtener una actualización del modelo agregado a fin de aplicarla al modelo global en el servidor.

TFF aporta los operadores que cumplen con este objetivo, como tff.federated_sum, y además tff.templates.AggregationProcess (un proceso con datos con estado) que formaliza la firma de tipo para el cálculo de agregación, de modo que se pueda generalizar en formas más complejas que la de una simple suma.

Los componentes principales del módulo tff.aggregators son las factorías para la creación del AggregationProcess, que están diseñadas para ser bloques de construcción útiles y reemplazables de TFF en dos aspectos:

1. Cálculos parametrizados. La agregación es un bloque de construcción independiente que se puede conectar con los módulos TFF diseñados para trabajar con tff.aggregators a fin de parametrizar su agregación necesaria.

Ejemplo:

learning_process = tff.learning.algorithms.build_weighted_fed_avg(
    ...,
    model_aggregator=tff.aggregators.MeanFactory())

1. Composición de agregación. Un bloque de construcción de agregación se puede componer con otros bloques de construcción de agregación para crear agregaciones compuestas más complejas.

Ejemplo:

secure_mean = tff.aggregators.MeanFactory(
    value_sum_factory=tff.aggregators.SecureSumFactory(...))

En el resto de este tutorial se explica cómo lograr estos dos objetivos.

Proceso de agregación

Primero, resumimos el tff.templates.AggregationProcess y a continuación, seguimos con el patrón de factoría para su creación.

El tff.templates.AggregationProcess es un tff.templates.MeasuredProcess con firmas de tipo especificadas para agregación. En particular, las funciones initialize y next tienen las siguientes firmas de tipo:

• ( -> state_type@SERVER)

• (<state_type@SERVER, {value_type}@CLIENTS, *> -> <state_type@SERVER, value_type@SERVER, measurements_type@SERVER>)

El estado (de tipo state_type) debe estar ubicado en el servidor. La función next toma como entrada al estado y un valor para ser agregado (de tipo value_type) que se ubica en los clientes. El código * indica que es opcional para otros argumentos de entrada, por ejemplo, para los pesos en una media ponderada. Devuelve un objeto de estado actualizado, el valor agregado del mismo tipo ubicado en el servidor y algunas mediciones.

Tenga en cuenta que tanto el estado para pasar entre ejecuciones de la función next como las mediciones previstas para reportar cualquier información dependiendo de una ejecución específica de la función next pueden estar vacías. Sin embargo, deben explicitarse para que otras partes de TFF tengan un convenio claro para seguir.

Se espera que otros módulos de TFF, como las actualizaciones del módulo en tff.learning, usen el tff.templates.AggregationProcess para parametrizar cómo se agregan los valores. Sin embargo, qué sean exactamente esos valores agregados y cuáles sean sus firmas de tipo dependerá de otros detalles del modelo que se entrena y del algoritmo de entrenamiento que se utilice.

Para hacer que la agregación sea independiente de otros aspectos de los cálculos, utilizamos el patrón de factoría. Creamos el tff.templates.AggregationProcess apropiado una vez que contamos con las firmas de tipo de los objetos que se agregarán. Para hacerlo, invocamos el método create de la factoría. Por lo tanto, la manipulación directa del proceso de agregación es necesaria solamente para los autores de la biblioteca, que son los responsables de la creación.

Factorías de proceso de agregación

Hay dos clases de factorías base abstractas para la agregación ponderada y subponderada. Su método create toma las firmas de tipo de los valores que se agregarán y devuelve un tff.templates.AggregationProcess para la agregación de tales valores.

El proceso creado por tff.aggregators.UnweightedAggregationFactory toma dos argumentos de entrada: (1) el estado en el servidor y (2) el valor del tipo especificado value_type.

tff.aggregators.SumFactory es un ejemplo de implementación.

El proceso creado por tff.aggregators.WeightedAggregationFactory toma tres argumentos de entrada: (1) el estado del servidor, (2) el valor del tipo especificado value_type y (3) el peso del tipo weight_type, tal como lo especifica el usuario de la factoría cuando invoca su método create.

tff.aggregators.MeanFactory es un ejemplo de implementación que calcula una media ponderada.

El patrón de factoría es lo que nos permite lograr el primer objetivo planteado arriba: que la agregación sea un bloque de construcción independiente. Por ejemplo, cuando cambiamos qué variables del modelo son entrenables, no necesariamente debemos cambiar la agregación compleja. La factoría que lo represente será invocada con una firma de tipo diferente, cada vez que se use un método como tff.learning.algorithms.build_weighted_fed_avg.

Composiciones

Recordemos que un proceso de agregación general puede encapsular (a) algún procesamiento de los valores en los clientes, (b) el movimiento de valores de clientes a servidores y (c) algún posprocesamiento de valores agregados en el servidor. El segundo objetivo definido en este tutorial es la composición de la agregación. Se obtiene dentro del módulo tff.aggregators estructurando la implementación de las factorías de agregación de modo tal que la parte (b) se pueda delegar a otra factoría de agregación.

En vez de implementar toda la lógica necesaria dentro de una sola clase de factoría, las implementaciones, por defecto, se centran en un único aspecto relevante para la agregación. Cuando es necesario, este patrón nos permite reemplazar los bloques de construcción de a uno por vez.

La tff.aggregators.MeanFactory es un ejemplo claro. Su implementación multiplica los valores provistos y los pesos en los clientes, después suma los valores ponderados y los pesos de forma independiente; finalmente, divide la suma de valores ponderados por la suma de pesos en el servidor. En vez de implementar las sumatorias usando directamente el operador tff.federated_sum, la sumatoria se delega a dos instancias de tff.aggregators.SumFactory.

Una estructura como esta permite que las dos sumatorias puedan ser reemplazadas por factorías diferentes, que realizan las sumas de forma diferente. Por ejemplo, una tff.aggregators.SecureSumFactory o una implementación personalizada de la tff.aggregators.UnweightedAggregationFactory. En cambio, esta vez, tff.aggregators.MeanFactory puede ser una agregación interna de otra factoría como tff.aggregators.clipping_factory, en caso de que los valores se vayan a recortar (clipped) antes de promediarlos.

Veamos el siguiente tutorial sobre agregaciones recomendadas para ajustes en aprendizaje en el que se sugiere cómo usar el mecanismo de composición con las factorías que se encuentran en el módulo tff.aggregators.

Mejores prácticas con ejemplos

Vamos a ilustrar los conceptos de tff.aggregators en detalle. Implementaremos una tarea simple de ejemplo y, progresivamente, la volveremos más general. Otra forma de aprender es mediante la observación de la implementación de las factorías que ya existen.

import collections
import tensorflow as tf
import tensorflow_federated as tff

En vez de sumar value, en la tarea de ejemplo se suma value * 2.0 y se divide por 2.0. La agregación que se obtiene como resultado, por lo tanto, es matemáticamente equivalente a sumar directamente el value y se podría pensar como una composición de tres partes: (1) el escalamiento en los clientes (2) la suma entre distintos clientes y (3) el desescalamiento en el servidor.

NOTA: Esta tarea no necesariamente es útil en la práctica. Sin embargo, sí ayuda a explicar los conceptos de base.

Continuando con el diseño explicado arriba, la lógica se implementará como una subclase de tff.aggregators.UnweightedAggregationFactory, que crea tff.templates.AggregationProcess apropiados cuando se le da un value_type para agregar:

Implementación mínima

En la tarea de ejemplo, los cálculos son siempre los mismos, así que no hay necesidad de utilizar el estado. Por lo tanto, está vacío y representado como tff.federated_value((), tff.SERVER). Por ahora, sucede lo mismo con las mediciones.

La implementación mínima de la tarea es, entonces, la siguiente:

class ExampleTaskFactory(tff.aggregators.UnweightedAggregationFactory):

  def create(self, value_type):
    @tff.federated_computation()
    def initialize_fn():
      return tff.federated_value((), tff.SERVER)

    @tff.federated_computation(initialize_fn.type_signature.result,
                               tff.type_at_clients(value_type))
    def next_fn(state, value):
      scaled_value = tff.federated_map(
          tff.tf_computation(lambda x: x * 2.0), value)
      summed_value = tff.federated_sum(scaled_value)
      unscaled_value = tff.federated_map(
          tff.tf_computation(lambda x: x / 2.0), summed_value)
      measurements = tff.federated_value((), tff.SERVER)
      return tff.templates.MeasuredProcessOutput(
          state=state, result=unscaled_value, measurements=measurements)

    return tff.templates.AggregationProcess(initialize_fn, next_fn)

Con el siguiente código se puede verificar si todo funciona como se espera:

client_data = [1.0, 2.0, 5.0]
factory = ExampleTaskFactory()
aggregation_process = factory.create(tff.TensorType(tf.float32))
print(f'Type signatures of the created aggregation process:\n'
      f'  - initialize: {aggregation_process.initialize.type_signature}\n'
      f'  - next: {aggregation_process.next.type_signature}\n')

state = aggregation_process.initialize()
output = aggregation_process.next(state, client_data)
print(f'Aggregation result: {output.result}  (expected 8.0)')

Type signatures of the created aggregation process:
  - initialize: ( -> <>@SERVER)
  - next: (<state=<>@SERVER,value={float32}@CLIENTS> -> <state=<>@SERVER,result=float32@SERVER,measurements=<>@SERVER>)

Aggregation result: 8.0  (expected 8.0)

Cálculos con estado y mediciones

Los estados se usan mucho en TFF para representar cálculos que se esperan ejecutar iterativamente y que se pretende cambiar con cada iteración. Por ejemplo, el estado de un cálculo de aprendizaje tiene los pesos del modelo que se está aprendiendo.

Para ilustrar cómo se utiliza el estado en un cálculo de agregación, modificamos la tarea de ejemplo. En vez de multiplicar el value por 2.0, lo multiplicamos por el índice de iteración, la cantidad de veces que se ha ejecutado la agregación.

Para lograrlo, debemos hallar una forma de dar seguimiento al índice de iteración, que se logra con el concepto de estado. En initialize_fn, en vez de crear un estado vacío, inicializamos el estado con un cero escalar. Luego, el estado se puede utilizar en next_fn en tres pasos: (1) aumento en 1.0, (2) multiplicación de value y (3) devolución como un estado actualizado nuevo.

Una vez que los pasos anteriores se han concretado, pensará lo siguiente: Si se puede usar exactamente el mismo código para verificar que todo funciona como se espera. ¿Cómo sé que algo ha cambiado realmente?

Buena pregunta. Aquí es donde el concepto de mediciones se vuelve útil. En general, las mediciones pueden informar cualquier valor relevante a una ejecución simple de la función next, que se podría usar para la monitorización. En este caso, puede ser el summed_value del ejemplo anterior. Es decir, el valor anterior al paso de "desescalamiento" que debería depender del índice de iteración. Una vez más, no es necesariamente útil en la práctica, sino que ilustra el mecanismo relevante.

Por lo tanto, la respuesta con estado para la tarea tiene el siguiente aspecto:

class ExampleTaskFactory(tff.aggregators.UnweightedAggregationFactory):

  def create(self, value_type):
    @tff.federated_computation()
    def initialize_fn():
      return tff.federated_value(0.0, tff.SERVER)

    @tff.federated_computation(initialize_fn.type_signature.result,
                               tff.type_at_clients(value_type))
    def next_fn(state, value):
      new_state = tff.federated_map(
          tff.tf_computation(lambda x: x + 1.0), state)
      state_at_clients = tff.federated_broadcast(new_state)
      scaled_value = tff.federated_map(
          tff.tf_computation(lambda x, y: x * y), (value, state_at_clients))
      summed_value = tff.federated_sum(scaled_value)
      unscaled_value = tff.federated_map(
          tff.tf_computation(lambda x, y: x / y), (summed_value, new_state))
      return tff.templates.MeasuredProcessOutput(
          state=new_state, result=unscaled_value, measurements=summed_value)

    return tff.templates.AggregationProcess(initialize_fn, next_fn)

Tenga en cuenta que el state que ingresa en next_fn como entrada se ubica en el servidor. Para usarlo en los clientes, primero debemos comunicarlo; esto se consigue con el operador tff.federated_broadcast.

Para verificar que todo funciona como fue previsto, podemos observar las measurements informadas, que deberían ser diferentes en cada ronda de ejecución, incluso aunque la ronda se ejecute con los mismos client_data.

client_data = [1.0, 2.0, 5.0]
factory = ExampleTaskFactory()
aggregation_process = factory.create(tff.TensorType(tf.float32))
print(f'Type signatures of the created aggregation process:\n'
      f'  - initialize: {aggregation_process.initialize.type_signature}\n'
      f'  - next: {aggregation_process.next.type_signature}\n')

state = aggregation_process.initialize()

output = aggregation_process.next(state, client_data)
print('| Round #1')
print(f'|       Aggregation result: {output.result}   (expected 8.0)')
print(f'| Aggregation measurements: {output.measurements}   (expected 8.0 * 1)')

output = aggregation_process.next(output.state, client_data)
print('\n| Round #2')
print(f'|       Aggregation result: {output.result}   (expected 8.0)')
print(f'| Aggregation measurements: {output.measurements}  (expected 8.0 * 2)')

output = aggregation_process.next(output.state, client_data)
print('\n| Round #3')
print(f'|       Aggregation result: {output.result}   (expected 8.0)')
print(f'| Aggregation measurements: {output.measurements}  (expected 8.0 * 3)')

Type signatures of the created aggregation process:
  - initialize: ( -> float32@SERVER)
  - next: (<state=float32@SERVER,value={float32}@CLIENTS> -> <state=float32@SERVER,result=float32@SERVER,measurements=float32@SERVER>)

| Round #1
|       Aggregation result: 8.0   (expected 8.0)
| Aggregation measurements: 8.0   (expected 8.0 * 1)

| Round #2
|       Aggregation result: 8.0   (expected 8.0)
| Aggregation measurements: 16.0  (expected 8.0 * 2)

| Round #3
|       Aggregation result: 8.0   (expected 8.0)
| Aggregation measurements: 24.0  (expected 8.0 * 3)

Tipos estructurados

Los pesos de un modelo entrenado con aprendizaje federado, por lo general, se representan como una colección de tensores, en vez de con un solo tensor. En TFF, se representa como tff.StructType y, por lo común, con factorías de agregación útiles que debe haber para poder aceptar tipos estructurados.

Sin embargo, en los ejemplos anteriores, solamente trabajamos con un objeto tff.TensorType. Si intentamos usar la factoría anterior para crear el proceso de agregación con un tff.StructType([(tf.float32, (2,)), (tf.float32, (3,))]), obtendremos error extraño, porque TensorFlow intentará multiplicar un tf.Tensor y una lista list.

El problema es que en vez de multiplicar la estructura de tensores por una constante, deberemos multiplicar cada tensor de la estructura por una constante. La solución verdadera a este problema es usar el módulo tf.nest dentro de los tff.tf_computation creados.

La versión anterior de ExampleTaskFactory compatible con tipos estructurados, por lo tanto, se ve de la siguiente manera:

@tff.tf_computation()
def scale(value, factor):
  return tf.nest.map_structure(lambda x: x * factor, value)

@tff.tf_computation()
def unscale(value, factor):
  return tf.nest.map_structure(lambda x: x / factor, value)

@tff.tf_computation()
def add_one(value):
  return value + 1.0

class ExampleTaskFactory(tff.aggregators.UnweightedAggregationFactory):

  def create(self, value_type):
    @tff.federated_computation()
    def initialize_fn():
      return tff.federated_value(0.0, tff.SERVER)

    @tff.federated_computation(initialize_fn.type_signature.result,
                               tff.type_at_clients(value_type))
    def next_fn(state, value):
      new_state = tff.federated_map(add_one, state)
      state_at_clients = tff.federated_broadcast(new_state)
      scaled_value = tff.federated_map(scale, (value, state_at_clients))
      summed_value = tff.federated_sum(scaled_value)
      unscaled_value = tff.federated_map(unscale, (summed_value, new_state))
      return tff.templates.MeasuredProcessOutput(
          state=new_state, result=unscaled_value, measurements=summed_value)

    return tff.templates.AggregationProcess(initialize_fn, next_fn)

En este ejemplo nos centramos en un patrón que puede ser útil para cuando estructuramos el código TFF. Cuando no trabajamos con operaciones muy simples, el código se vuelve más legible si los tff.tf_computation, que se usarán como bloques de construcción dentro de un tff.federated_computation, se crean en un lugar separado. Dentro de tff.federated_computation, estos bloques de construcción solamente están conectados con las operaciones intrínsecas.

Para verificar si funciona como fue previsto:

client_data = [[[1.0, 2.0], [3.0, 4.0, 5.0]],
               [[1.0, 1.0], [3.0, 0.0, -5.0]]]
factory = ExampleTaskFactory()
aggregation_process = factory.create(
    tff.to_type([(tf.float32, (2,)), (tf.float32, (3,))]))
print(f'Type signatures of the created aggregation process:\n'
      f'  - initialize: {aggregation_process.initialize.type_signature}\n'
      f'  - next: {aggregation_process.next.type_signature}\n')

state = aggregation_process.initialize()
output = aggregation_process.next(state, client_data)
print(f'Aggregation result: [{output.result[0]}, {output.result[1]}]\n'
      f'          Expected: [[2. 3.], [6. 4. 0.]]')

Type signatures of the created aggregation process:
  - initialize: ( -> float32@SERVER)
  - next: (<state=float32@SERVER,value={<float32[2],float32[3]>}@CLIENTS> -> <state=float32@SERVER,result=<float32[2],float32[3]>@SERVER,measurements=<float32[2],float32[3]>@SERVER>)

Aggregation result: [[2. 3.], [6. 4. 0.]]
          Expected: [[2. 3.], [6. 4. 0.]]

Agregaciones internas

El último paso consiste en permitir, como opción, la delegación de la agregación real a las factorías, a fin de lograr una composición simple de técnicas de agregación diferentes.

Esto se logra mediante la creación de un argumento inner_factory en el constructor de nuestra ExampleTaskFactory. A menos que se especifique lo contrario, se usa tff.aggregators.SumFactory, que aplica el operador tff.federated_sum que utilizamos directamente en la sección anterior.

Cuando llamamos a create, primero podemos llamar a create de inner_factory para crear el proceso de agregación interna con el mismo value_type.

El estado de nuestro proceso devuelto por initialize_fn está compuesto por dos partes: el estado creado por este proceso y el estado del proceso interno que acabamos de crear.

La implementación de next_fn difiere en que la agregación real se delega a la función next del proceso interno y en la manera en que está compuesta la salida final. El estado vuelve a estar compuesto por "aquel" estado y el "interno", y las mediciones se componen de un modo similar como un OrderedDict.

La siguiente es una implementación de un patrón como el mencionado:

@tff.tf_computation()
def scale(value, factor):
  return tf.nest.map_structure(lambda x: x * factor, value)

@tff.tf_computation()
def unscale(value, factor):
  return tf.nest.map_structure(lambda x: x / factor, value)

@tff.tf_computation()
def add_one(value):
  return value + 1.0

class ExampleTaskFactory(tff.aggregators.UnweightedAggregationFactory):

  def __init__(self, inner_factory=None):
    if inner_factory is None:
      inner_factory = tff.aggregators.SumFactory()
    self._inner_factory = inner_factory

  def create(self, value_type):
    inner_process = self._inner_factory.create(value_type)

    @tff.federated_computation()
    def initialize_fn():
      my_state = tff.federated_value(0.0, tff.SERVER)
      inner_state = inner_process.initialize()
      return tff.federated_zip((my_state, inner_state))

    @tff.federated_computation(initialize_fn.type_signature.result,
                               tff.type_at_clients(value_type))
    def next_fn(state, value):
      my_state, inner_state = state
      my_new_state = tff.federated_map(add_one, my_state)
      my_state_at_clients = tff.federated_broadcast(my_new_state)
      scaled_value = tff.federated_map(scale, (value, my_state_at_clients))

      # Delegation to an inner factory, returning values placed at SERVER.
      inner_output = inner_process.next(inner_state, scaled_value)

      unscaled_value = tff.federated_map(unscale, (inner_output.result, my_new_state))

      new_state = tff.federated_zip((my_new_state, inner_output.state))
      measurements = tff.federated_zip(
          collections.OrderedDict(
              scaled_value=inner_output.result,
              example_task=inner_output.measurements))

      return tff.templates.MeasuredProcessOutput(
          state=new_state, result=unscaled_value, measurements=measurements)

    return tff.templates.AggregationProcess(initialize_fn, next_fn)

Cuando delegamos a la función inner_process.next, la estructura de retorno que obtenemos es una tff.templates.MeasuredProcessOutput, con los mismos tres campos: estado, resultado y mediciones. Cuando creamos la estructura general de retorno del proceso de agregación compuesto, los campos state y measurements, por lo general, deberían estar compuestos y ser devueltos juntos. Por el contrario, el campo result corresponde a los valores que se agregan y "fluye por" la agregación compuesta.

El objeto state se debería ver como un detalle de la implementación de la factoría y, por lo tanto, la composición podría tener cualquier estructura. Sin embargo, las measurements corresponden a valores que se informarán al usuario en algún momento. Por lo tanto, recomendamos usar OrderedDict, con nombres compuestos de modo tal que quede claro de qué lugar de la composición proviene la métrica informada.

También tenga en cuenta que usamos el operador tff.federated_zip. El objeto state controlado por el proceso creado debería ser un tff.FederatedType. Si, en cambio hubiésemos devuelto (this_state, inner_state) en la initialize_fn, la firma de tipo del retorno sería un tff.StructType con dos tuplas de tff.FederatedType. Si usamos tff.federated_zip se "eleva" el tff.FederatedType al nivel más alto. Todo esto se usa de un modo similar en la next_fn cuando preparamos el estado y las mediciones que retornarán.

Finalmente, podemos ver cómo se puede aplicar con la agregación interna predeterminada:

client_data = [1.0, 2.0, 5.0]
factory = ExampleTaskFactory()
aggregation_process = factory.create(tff.TensorType(tf.float32))
state = aggregation_process.initialize()

output = aggregation_process.next(state, client_data)
print('| Round #1')
print(f'|           Aggregation result: {output.result}   (expected 8.0)')
print(f'| measurements[\'scaled_value\']: {output.measurements["scaled_value"]}')
print(f'| measurements[\'example_task\']: {output.measurements["example_task"]}')

output = aggregation_process.next(output.state, client_data)
print('\n| Round #2')
print(f'|           Aggregation result: {output.result}   (expected 8.0)')
print(f'| measurements[\'scaled_value\']: {output.measurements["scaled_value"]}')
print(f'| measurements[\'example_task\']: {output.measurements["example_task"]}')

| Round #1
|           Aggregation result: 8.0   (expected 8.0)
| measurements['scaled_value']: 8.0
| measurements['example_task']: ()

| Round #2
|           Aggregation result: 8.0   (expected 8.0)
| measurements['scaled_value']: 16.0
| measurements['example_task']: ()

... y con una agregación interna diferente. Por ejemplo, con una ExampleTaskFactory:

client_data = [1.0, 2.0, 5.0]
# Note the inner delegation can be to any UnweightedAggregaionFactory.
# In this case, each factory creates process that multiplies by the iteration
# index (1, 2, 3, ...), thus their combination multiplies by (1, 4, 9, ...).
factory = ExampleTaskFactory(ExampleTaskFactory())
aggregation_process = factory.create(tff.TensorType(tf.float32))
state = aggregation_process.initialize()

output = aggregation_process.next(state, client_data)
print('| Round #1')
print(f'|           Aggregation result: {output.result}   (expected 8.0)')
print(f'| measurements[\'scaled_value\']: {output.measurements["scaled_value"]}')
print(f'| measurements[\'example_task\']: {output.measurements["example_task"]}')

output = aggregation_process.next(output.state, client_data)
print('\n| Round #2')
print(f'|           Aggregation result: {output.result}   (expected 8.0)')
print(f'| measurements[\'scaled_value\']: {output.measurements["scaled_value"]}')
print(f'| measurements[\'example_task\']: {output.measurements["example_task"]}')

| Round #1
|           Aggregation result: 8.0   (expected 8.0)
| measurements['scaled_value']: 8.0
| measurements['example_task']: OrderedDict([('scaled_value', 8.0), ('example_task', ())])

| Round #2
|           Aggregation result: 8.0   (expected 8.0)
| measurements['scaled_value']: 16.0
| measurements['example_task']: OrderedDict([('scaled_value', 32.0), ('example_task', ())])

Resumen

En este tutorial explicamos las mejores prácticas que se pueden seguir para crear bloques de construcción de agregación para propósitos generales, representados en forma de una factoría de agregación. La generalidad aborda la intención de diseño de dos formas:

1. Cálculos parametrizados. La agregación es un bloque de construcción independiente que se puede conectar con los módulos TFF diseñados para trabajar con tff.aggregators a fin de parametrizar su agregación necesaria, como tff.learning.algorithms.build_weighted_fed_avg.

2. Composición de agregación. Un bloque de construcción de agregación se puede componer con otros bloques de construcción de agregación para crear agregaciones compuestas más complejas.