آموزش فدراسیون برای طبقه بندی تصویر

مشاهده در TensorFlow.org

در Google Colab اجرا شود

مشاهده منبع در GitHub

دانلود دفترچه یادداشت

در این آموزش، ما به عنوان مثال آموزش MNIST کلاسیک استفاده به معرفی آموزش فدرال (FL) لایه API از TFF، tff.learning - مجموعه ای از رابط های سطح بالا است که می تواند مورد استفاده برای انجام انواع معمول وظایف یادگیری فدرال، مانند آموزش فدرال، در برابر مدل های ارائه شده توسط کاربر پیاده سازی شده در TensorFlow.

این آموزش و API یادگیری فدرال عمدتاً برای کاربرانی در نظر گرفته شده است که می‌خواهند مدل‌های TensorFlow خود را به TFF وصل کنند و مدل دوم را بیشتر به عنوان یک جعبه سیاه در نظر می‌گیرند. - برای درک بیشتر در عمق از TFF و چگونگی پیاده سازی الگوریتم های یادگیری خود فدرال خود را، آموزش بر روی API FC هسته دیدن سفارشی فدرال الگوریتم قسمت 1 و قسمت 2 .

برای اطلاعات بیشتر در tff.learning ، با ادامه آموزش فدرال برای متن ، آموزش که علاوه بر پوشش مدل های مکرر، همچنین نشان می دهد در حال بارگذاری یک مدل Keras مرتب قبل از آموزش دیده برای پالایش با یادگیری فدرال همراه با ارزیابی با استفاده از Keras.

قبل از اینکه شروع کنیم

قبل از شروع، لطفاً موارد زیر را اجرا کنید تا مطمئن شوید که محیط شما به درستی تنظیم شده است. اگر شما یک تبریک نمی بینم، لطفا به مراجعه نصب و راه اندازی راهنمای دستورالعمل.

# tensorflow_federated_nightly also bring in tf_nightly, which
# can causes a duplicate tensorboard install, leading to errors.
!pip uninstall --yes tensorboard tb-nightly

!pip install --quiet --upgrade tensorflow-federated-nightly
!pip install --quiet --upgrade nest-asyncio
!pip install --quiet --upgrade tb-nightly  # or tensorboard, but not both

import nest_asyncio
nest_asyncio.apply()

%load_ext tensorboard

import collections

import numpy as np
import tensorflow as tf
import tensorflow_federated as tff

np.random.seed(0)

tff.federated_computation(lambda: 'Hello, World!')()

b'Hello, World!'

آماده سازی داده های ورودی

بیایید با داده ها شروع کنیم. یادگیری فدرال نیاز به مجموعه داده های فدرال دارد، به عنوان مثال، مجموعه ای از داده ها از چندین کاربر. اطلاعات فدرال معمولا غیر IID ، که به شمار مجموعه ای منحصر به فرد از چالش است.

به منظور تسهیل آزمایش، ما کشت مخزن TFF با چند مجموعه داده، از جمله یک نسخه فدرال MNIST که حاوی یک نسخه از مجموعه داده NIST اصلی شده است که دوباره پردازش با استفاده از برگ به طوری که داده است نویسنده اصلی از کوک ارقام از آنجایی که هر نویسنده دارای سبک منحصر به فردی است، این مجموعه داده نوع رفتار غیر iid مورد انتظار از مجموعه داده های فدرال را نشان می دهد.

در اینجا نحوه بارگذاری آن آمده است.

emnist_train, emnist_test = tff.simulation.datasets.emnist.load_data()

مجموعه داده های بازگردانده شده توسط load_data() موارد هستند tff.simulation.ClientData ، یک رابط است که اجازه می دهد تا شما را به شمردن مجموعه ای از کاربران، برای ساخت یک tf.data.Dataset که نشان دهنده داده ها از یک کاربر خاص، و به پرس و جو ساختار عناصر منفرد در اینجا نحوه استفاده از این رابط برای کاوش در محتوای مجموعه داده آمده است. به خاطر داشته باشید که در حالی که این رابط به شما امکان می دهد روی شناسه های مشتری تکرار کنید، این تنها یکی از ویژگی های داده های شبیه سازی است. همانطور که به زودی خواهید دید، هویت های مشتری توسط چارچوب یادگیری فدرال استفاده نمی شود - تنها هدف آنها این است که به شما اجازه دهند زیر مجموعه هایی از داده ها را برای شبیه سازی انتخاب کنید.

len(emnist_train.client_ids)

3383

emnist_train.element_type_structure

OrderedDict([('label', TensorSpec(shape=(), dtype=tf.int32, name=None)), ('pixels', TensorSpec(shape=(28, 28), dtype=tf.float32, name=None))])

example_dataset = emnist_train.create_tf_dataset_for_client(
    emnist_train.client_ids[0])

example_element = next(iter(example_dataset))

example_element['label'].numpy()

1

from matplotlib import pyplot as plt
plt.imshow(example_element['pixels'].numpy(), cmap='gray', aspect='equal')
plt.grid(False)
_ = plt.show()

بررسی ناهمگونی در داده های فدرال

اطلاعات فدرال معمولا غیر IID ، کاربران به طور معمول دارای توزیع های مختلف داده بسته به الگوهای استفاده. برخی از مشتریان ممکن است نمونه‌های آموزشی کمتری روی دستگاه داشته باشند، در حالی که از کمبود داده‌ها به صورت محلی رنج می‌برند، در حالی که برخی از مشتریان نمونه‌های آموزشی بیش از اندازه کافی خواهند داشت. بیایید این مفهوم ناهمگونی داده‌ها را با داده‌های EMNIST که در دسترس داریم بررسی کنیم. توجه به این نکته مهم است که این تجزیه و تحلیل عمیق از داده های مشتری فقط برای ما در دسترس است زیرا این یک محیط شبیه سازی است که در آن تمام داده ها به صورت محلی در دسترس ما هستند. در یک محیط فدرال تولید واقعی، نمی‌توانید داده‌های یک مشتری را بررسی کنید.

ابتدا، بیایید نمونه‌ای از داده‌های یک مشتری بگیریم تا نمونه‌های موجود در یک دستگاه شبیه‌سازی‌شده را احساس کنیم. از آنجایی که مجموعه داده ای که ما استفاده می کنیم توسط یک نویسنده منحصر به فرد کلید زده شده است، داده های یک مشتری دستخط یک فرد را برای نمونه ای از ارقام 0 تا 9 نشان می دهد و "الگوی استفاده" منحصر به فرد یک کاربر را شبیه سازی می کند.

## Example MNIST digits for one client
figure = plt.figure(figsize=(20, 4))
j = 0

for example in example_dataset.take(40):
  plt.subplot(4, 10, j+1)
  plt.imshow(example['pixels'].numpy(), cmap='gray', aspect='equal')
  plt.axis('off')
  j += 1

حالا بیایید تعداد نمونه ها را در هر مشتری برای هر برچسب رقمی MNIST تجسم کنیم. در محیط فدرال، بسته به رفتار کاربر، تعداد نمونه‌ها روی هر مشتری می‌تواند کمی متفاوت باشد.

# Number of examples per layer for a sample of clients
f = plt.figure(figsize=(12, 7))
f.suptitle('Label Counts for a Sample of Clients')
for i in range(6):
  client_dataset = emnist_train.create_tf_dataset_for_client(
      emnist_train.client_ids[i])
  plot_data = collections.defaultdict(list)
  for example in client_dataset:
    # Append counts individually per label to make plots
    # more colorful instead of one color per plot.
    label = example['label'].numpy()
    plot_data[label].append(label)
  plt.subplot(2, 3, i+1)
  plt.title('Client {}'.format(i))
  for j in range(10):
    plt.hist(
        plot_data[j],
        density=False,
        bins=[0, 1, 2, 3, 4, 5, 6, 7, 8, 9, 10])

حالا بیایید میانگین تصویر به ازای هر مشتری را برای هر برچسب MNIST تجسم کنیم. این کد میانگین هر پیکسل را برای تمام مثال های کاربر برای یک برچسب تولید می کند. خواهیم دید که تصویر میانگین یک مشتری برای یک رقم، به دلیل سبک دست خط منحصر به فرد هر فرد، متفاوت از تصویر میانگین مشتری دیگر برای همان رقم خواهد بود. ما می‌توانیم در مورد اینکه چگونه هر دور آموزشی محلی، مدل را در جهتی متفاوت در هر مشتری هدایت می‌کند، فکر کنیم، زیرا از داده‌های منحصربه‌فرد خود کاربر در آن دور محلی یاد می‌گیریم. بعداً در آموزش خواهیم دید که چگونه می‌توانیم هر به‌روزرسانی را به مدل از همه کلاینت‌ها ببریم و آنها را با هم در مدل جهانی جدید خود که از داده‌های منحصربه‌فرد هر یک از مشتری‌هایمان آموخته‌ایم جمع کنیم.

# Each client has different mean images, meaning each client will be nudging
# the model in their own directions locally.

for i in range(5):
  client_dataset = emnist_train.create_tf_dataset_for_client(
      emnist_train.client_ids[i])
  plot_data = collections.defaultdict(list)
  for example in client_dataset:
    plot_data[example['label'].numpy()].append(example['pixels'].numpy())
  f = plt.figure(i, figsize=(12, 5))
  f.suptitle("Client #{}'s Mean Image Per Label".format(i))
  for j in range(10):
    mean_img = np.mean(plot_data[j], 0)
    plt.subplot(2, 5, j+1)
    plt.imshow(mean_img.reshape((28, 28)))
    plt.axis('off')

داده‌های کاربر می‌توانند نویز داشته باشند و برچسب نامطمئن داشته باشند. به عنوان مثال، با نگاه کردن به داده‌های Client #2 در بالا، می‌توانیم ببینیم که برای برچسب 2، ممکن است نمونه‌هایی با برچسب نادرست وجود داشته باشد که یک تصویر متوسط ​​نویزتر ایجاد کرده است.

پیش پردازش داده های ورودی

از آنجا که داده در حال حاضر یک tf.data.Dataset ، فرآیندی می تواند با استفاده از تحولات مجموعه داده انجام می شود. در اینجا، ما پهن 28x28 تصاویر را به 784 آرایه -Element، زدن مثال فردی، سازماندهی آنها را به دسته، و تغییر نام از ویژگی های از pixels و label به x و y برای استفاده با Keras. ما همچنین در یک پرتاب repeat بیش از مجموعه داده ها برای اجرای چندین دوره.

NUM_CLIENTS = 10
NUM_EPOCHS = 5
BATCH_SIZE = 20
SHUFFLE_BUFFER = 100
PREFETCH_BUFFER = 10

def preprocess(dataset):

  def batch_format_fn(element):
    """Flatten a batch `pixels` and return the features as an `OrderedDict`."""
    return collections.OrderedDict(
        x=tf.reshape(element['pixels'], [-1, 784]),
        y=tf.reshape(element['label'], [-1, 1]))

  return dataset.repeat(NUM_EPOCHS).shuffle(SHUFFLE_BUFFER, seed=1).batch(
      BATCH_SIZE).map(batch_format_fn).prefetch(PREFETCH_BUFFER)

بیایید بررسی کنیم که این کار کرده است.

preprocessed_example_dataset = preprocess(example_dataset)

sample_batch = tf.nest.map_structure(lambda x: x.numpy(),
                                     next(iter(preprocessed_example_dataset)))

sample_batch

OrderedDict([('x', array([[1., 1., 1., ..., 1., 1., 1.],
       [1., 1., 1., ..., 1., 1., 1.],
       [1., 1., 1., ..., 1., 1., 1.],
       ...,
       [1., 1., 1., ..., 1., 1., 1.],
       [1., 1., 1., ..., 1., 1., 1.],
       [1., 1., 1., ..., 1., 1., 1.]], dtype=float32)), ('y', array([[2],
       [1],
       [5],
       [7],
       [1],
       [7],
       [7],
       [1],
       [4],
       [7],
       [4],
       [2],
       [2],
       [5],
       [4],
       [1],
       [1],
       [0],
       [0],
       [9]], dtype=int32))])

ما تقریباً تمام بلوک‌های ساختمانی را برای ساخت مجموعه‌های داده فدرال در اختیار داریم.

یکی از راه های برای تغذیه پیوسته داده ها به TFF در یک شبیه سازی است که به سادگی به عنوان یک لیست پایتون، با هر عنصر از لیست برگزاری داده ها از کاربر افراد، چه به عنوان یک لیست و یا به عنوان یک tf.data.Dataset . از آنجایی که ما قبلاً رابطی داریم که دومی را فراهم می کند، بیایید از آن استفاده کنیم.

در اینجا یک تابع کمکی ساده وجود دارد که لیستی از مجموعه داده ها را از مجموعه داده شده از کاربران به عنوان ورودی برای دور آموزش یا ارزیابی ایجاد می کند.

def make_federated_data(client_data, client_ids):
  return [
      preprocess(client_data.create_tf_dataset_for_client(x))
      for x in client_ids
  ]

حال، چگونه مشتریان را انتخاب کنیم؟

در یک سناریوی آموزشی معمولی فدرال، ما با جمعیت بسیار بزرگی از دستگاه‌های کاربر سر و کار داریم، که تنها بخشی از آن ممکن است در یک مقطع زمانی معین برای آموزش در دسترس باشد. به عنوان مثال، هنگامی که دستگاه‌های سرویس گیرنده تلفن‌های همراهی هستند که فقط زمانی که به منبع برق وصل هستند، شبکه‌های اندازه‌گیری شده و غیرفعال هستند، در آموزش شرکت می‌کنند.

البته ما در یک محیط شبیه سازی هستیم و تمام داده ها به صورت محلی در دسترس هستند. معمولاً، هنگام اجرای شبیه‌سازی‌ها، ما به سادگی از یک زیرمجموعه تصادفی از مشتریان برای شرکت در هر دور آموزشی نمونه‌برداری می‌کنیم، که معمولاً در هر دور متفاوت است.

گفت که، به عنوان شما می توانید با مطالعه این مقاله بر روی پیدا کردن به طور متوسط فدرال الگوریتم، دستیابی به همگرایی در یک سیستم با زیر مجموعه های به طور تصادفی نمونه برداری از مشتریان در هر دور می تواند در حالی که، و آن را غیر عملی خواهد بود که به اجرا صدها دور در این آموزش تعاملی

کاری که ما به جای آن انجام خواهیم داد این است که یک بار از مجموعه مشتریان نمونه برداری کرده و از همان مجموعه در سراسر دور برای سرعت بخشیدن به همگرایی استفاده مجدد کنیم (تطبیق عمدی با داده های این چند کاربر). ما آن را به عنوان تمرینی برای خواننده می‌گذاریم که این آموزش را برای شبیه‌سازی نمونه‌گیری تصادفی تغییر دهد - انجام آن نسبتاً آسان است (پس از انجام، به خاطر داشته باشید که همگرا شدن مدل ممکن است مدتی طول بکشد).

sample_clients = emnist_train.client_ids[0:NUM_CLIENTS]

federated_train_data = make_federated_data(emnist_train, sample_clients)

print('Number of client datasets: {l}'.format(l=len(federated_train_data)))
print('First dataset: {d}'.format(d=federated_train_data[0]))

Number of client datasets: 10
First dataset: <DatasetV1Adapter shapes: OrderedDict([(x, (None, 784)), (y, (None, 1))]), types: OrderedDict([(x, tf.float32), (y, tf.int32)])>

ساخت مدل با Keras

اگر از Keras استفاده می کنید، احتمالاً قبلاً کدی دارید که یک مدل Keras را می سازد. در اینجا یک نمونه از یک مدل ساده است که برای نیازهای ما کافی است.

def create_keras_model():
  return tf.keras.models.Sequential([
      tf.keras.layers.InputLayer(input_shape=(784,)),
      tf.keras.layers.Dense(10, kernel_initializer='zeros'),
      tf.keras.layers.Softmax(),
  ])

به منظور استفاده از هر مدل با TFF، آن را باید در یک نمونه از پیچیده می شود tff.learning.Model رابط، که در معرض روش به مهر پاس رو به جلو مدل، خواص ابرداده، و غیره، به طور مشابه به Keras، اما همچنین به معرفی اضافی عناصری مانند روش هایی برای کنترل فرآیند محاسبه معیارهای فدرال. فعلا نگران این نباشیم. اگر شما یک مدل Keras مانند یکی از ما فقط در بالا تعریف شده است، شما می توانید TFF قرار دادن آن را برای شما با استناد به tff.learning.from_keras_model ، عبور از مدل و دسته ای از داده های نمونه را به عنوان آرگومان، به عنوان زیر نشان داده شده.

def model_fn():
  # We _must_ create a new model here, and _not_ capture it from an external
  # scope. TFF will call this within different graph contexts.
  keras_model = create_keras_model()
  return tff.learning.from_keras_model(
      keras_model,
      input_spec=preprocessed_example_dataset.element_spec,
      loss=tf.keras.losses.SparseCategoricalCrossentropy(),
      metrics=[tf.keras.metrics.SparseCategoricalAccuracy()])

آموزش مدل بر روی داده های فدرال

حالا که ما یک مدل به عنوان پیچیده tff.learning.Model برای استفاده با TFF، ما می توانید اجازه دهید TFF ساخت یک الگوریتم به طور متوسط فدرال با توسل به تابع کمکی tff.learning.build_federated_averaging_process ، به شرح زیر.

به خاطر داشته باشید که استدلال نیاز به یک سازنده (مانند model_fn بالا)، نه یک نمونه در حال حاضر ساخته شده، به طوری که ساخت و ساز از مدل خود را می توانید در یک زمینه توسط TFF کنترل اتفاق می افتد (اگر شما کنجکاو هستید در مورد دلایل این کار، ما شما را تشویق به خواندن آموزش پیگیری الگوریتم سفارشی ).

یک نکته مهم در الگوریتم به طور متوسط فدرال زیر، 2 بهینه وجود دارد: یک بهینه ساز _client و بهینه ساز _SERVER. بهینه ساز _client است تنها مورد استفاده برای محاسبه به روز رسانی مدل محلی در هر مشتری. بهینه ساز _SERVER به روز رسانی به طور متوسط به مدل جهانی در سرور اعمال می شود. به طور خاص، این بدان معنی است که انتخاب بهینه‌ساز و نرخ یادگیری مورد استفاده ممکن است با مواردی که برای آموزش مدل بر روی مجموعه داده استاندارد iid استفاده کرده‌اید متفاوت باشد. توصیه می کنیم با SGD معمولی شروع کنید، احتمالاً با نرخ یادگیری کمتر از حد معمول. نرخ یادگیری ما به دقت تنظیم نشده است، با خیال راحت آزمایش کنید.

iterative_process = tff.learning.build_federated_averaging_process(
    model_fn,
    client_optimizer_fn=lambda: tf.keras.optimizers.SGD(learning_rate=0.02),
    server_optimizer_fn=lambda: tf.keras.optimizers.SGD(learning_rate=1.0))

چه اتفاقی افتاده؟ TFF ساخته شده است یک جفت از محاسبات فدراسیونی و آنها را به یک بسته بندی tff.templates.IterativeProcess که در آن این محاسبات در دسترس هستند به عنوان یک جفت از خواص initialize و next .

به طور خلاصه، محاسبات فدرال برنامه در زبان داخلی TFF است که می تواند الگوریتم های مختلف فدرال اکسپرس (شما می توانید بیشتر در مورد این در پیدا سفارشی الگوریتم آموزش). در این مورد، دو محاسبات تولید و بسته بندی شده به iterative_process اجرای فدرال به طور متوسط .

هدف TFF تعریف محاسبات به گونه ای است که بتوان آنها را در تنظیمات یادگیری فدرال واقعی اجرا کرد، اما در حال حاضر فقط زمان اجرا شبیه سازی اجرای محلی پیاده سازی شده است. برای اجرای یک محاسبات در یک شبیه ساز، شما به سادگی آن را مانند یک تابع پایتون فراخوانی می کنید. این محیط تفسیری پیش فرض برای عملکرد بالا طراحی نشده است، اما برای این آموزش کافی است. ما انتظار داریم زمان‌های اجرا شبیه‌سازی با عملکرد بالاتر را برای تسهیل تحقیقات در مقیاس بزرگتر در نسخه‌های آینده ارائه کنیم.

بیایید با شروع initialize محاسبات. همانطور که در مورد همه محاسبات فدرال وجود دارد، می توانید آن را به عنوان یک تابع در نظر بگیرید. محاسبات هیچ آرگومان نمی گیرد و یک نتیجه را برمی گرداند - نمایش وضعیت فرآیند میانگین گیری فدرال در سرور. در حالی که ما نمی خواهیم به جزئیات TFF بپردازیم، ممکن است آموزنده باشد که ببینیم این حالت چگونه به نظر می رسد. می توانید آن را به صورت زیر تجسم کنید.

str(iterative_process.initialize.type_signature)

'( -> <model=<trainable=<float32[784,10],float32[10]>,non_trainable=<>>,optimizer_state=<int64>,delta_aggregate_state=<value_sum_process=<>,weight_sum_process=<>>,model_broadcast_state=<>>@SERVER)'

در حالی که امضای نوع بالا در ابتدا ممکن است به نظر می رسد مرموز کمی، شما می توانید تشخیص دهند که حالت سرور شامل یک model (پارامترهای مدل اولیه برای MNIST خواهد شد که به تمام دستگاه های توزیع شده)، و optimizer_state (اطلاعات اضافی که توسط سرور، مانند تعداد دورهای مورد استفاده برای زمانبندی هایپرپارامتر و غیره).

بیایید فراخوان initialize محاسبات برای ساخت حالت سرور.

state = iterative_process.initialize()

دوم از جفت محاسبات فدرال، next ، نشان دهنده یک دور از فدرال به طور متوسط، که متشکل از هل دادن به وضعیت سرور (از جمله پارامترهای مدل) به مشتریان، بر روی دستگاه آموزش بر روی داده های محلی خود را، جمع آوری و به روز رسانی مدل متوسط ، و تولید یک مدل جدید به روز شده در سرور.

از لحاظ مفهومی، شما می توانید از فکر می کنم next به عنوان داشتن یک امضای نوع کاربردی است که به نظر می رسد شرح زیر است.

SERVER_STATE, FEDERATED_DATA -> SERVER_STATE, TRAINING_METRICS

به طور خاص، باید در مورد فکر می کنم next() برخی از ورودی ها توسط سرور (ارائه - و نه به عنوان یک تابع است که قابل اجرا بر روی سرور، بلکه یک نمایندگی کاربردی اعلانی از کل محاسبات غیر متمرکز SERVER_STATE )، اما هر شرکت کننده دستگاه داده های محلی خود را ارائه می دهد.

بیایید یک دور تمرین را اجرا کنیم و نتایج را تجسم کنیم. می‌توانیم از داده‌های فدرالی که قبلاً در بالا تولید کرده‌ایم برای نمونه‌ای از کاربران استفاده کنیم.

state, metrics = iterative_process.next(state, federated_train_data)
print('round  1, metrics={}'.format(metrics))

round  1, metrics=OrderedDict([('broadcast', ()), ('aggregation', OrderedDict([('mean_value', ()), ('mean_weight', ())])), ('train', OrderedDict([('sparse_categorical_accuracy', 0.12345679), ('loss', 3.1193738)])), ('stat', OrderedDict([('num_examples', 4860)]))])

بیایید چند دور دیگر اجرا کنیم. همانطور که قبلاً ذکر شد، معمولاً در این مرحله شما زیرمجموعه‌ای از داده‌های شبیه‌سازی خود را از یک نمونه جدید به‌طور تصادفی انتخاب شده از کاربران برای هر دور انتخاب می‌کنید تا یک استقرار واقعی را شبیه‌سازی کنید که در آن کاربران پیوسته می‌آیند و می‌روند، اما در این نوت‌بوک تعاملی، برای نمایش ما فقط از همان کاربران استفاده مجدد می کنیم تا سیستم به سرعت همگرا شود.

NUM_ROUNDS = 11
for round_num in range(2, NUM_ROUNDS):
  state, metrics = iterative_process.next(state, federated_train_data)
  print('round {:2d}, metrics={}'.format(round_num, metrics))

round  2, metrics=OrderedDict([('broadcast', ()), ('aggregation', OrderedDict([('mean_value', ()), ('mean_weight', ())])), ('train', OrderedDict([('sparse_categorical_accuracy', 0.13518518), ('loss', 2.9834728)])), ('stat', OrderedDict([('num_examples', 4860)]))])
round  3, metrics=OrderedDict([('broadcast', ()), ('aggregation', OrderedDict([('mean_value', ()), ('mean_weight', ())])), ('train', OrderedDict([('sparse_categorical_accuracy', 0.14382716), ('loss', 2.861665)])), ('stat', OrderedDict([('num_examples', 4860)]))])
round  4, metrics=OrderedDict([('broadcast', ()), ('aggregation', OrderedDict([('mean_value', ()), ('mean_weight', ())])), ('train', OrderedDict([('sparse_categorical_accuracy', 0.17407407), ('loss', 2.7957022)])), ('stat', OrderedDict([('num_examples', 4860)]))])
round  5, metrics=OrderedDict([('broadcast', ()), ('aggregation', OrderedDict([('mean_value', ()), ('mean_weight', ())])), ('train', OrderedDict([('sparse_categorical_accuracy', 0.19917695), ('loss', 2.6146567)])), ('stat', OrderedDict([('num_examples', 4860)]))])
round  6, metrics=OrderedDict([('broadcast', ()), ('aggregation', OrderedDict([('mean_value', ()), ('mean_weight', ())])), ('train', OrderedDict([('sparse_categorical_accuracy', 0.21975309), ('loss', 2.529761)])), ('stat', OrderedDict([('num_examples', 4860)]))])
round  7, metrics=OrderedDict([('broadcast', ()), ('aggregation', OrderedDict([('mean_value', ()), ('mean_weight', ())])), ('train', OrderedDict([('sparse_categorical_accuracy', 0.2409465), ('loss', 2.4053504)])), ('stat', OrderedDict([('num_examples', 4860)]))])
round  8, metrics=OrderedDict([('broadcast', ()), ('aggregation', OrderedDict([('mean_value', ()), ('mean_weight', ())])), ('train', OrderedDict([('sparse_categorical_accuracy', 0.2611111), ('loss', 2.315389)])), ('stat', OrderedDict([('num_examples', 4860)]))])
round  9, metrics=OrderedDict([('broadcast', ()), ('aggregation', OrderedDict([('mean_value', ()), ('mean_weight', ())])), ('train', OrderedDict([('sparse_categorical_accuracy', 0.30823046), ('loss', 2.1240263)])), ('stat', OrderedDict([('num_examples', 4860)]))])
round 10, metrics=OrderedDict([('broadcast', ()), ('aggregation', OrderedDict([('mean_value', ()), ('mean_weight', ())])), ('train', OrderedDict([('sparse_categorical_accuracy', 0.33312756), ('loss', 2.1164262)])), ('stat', OrderedDict([('num_examples', 4860)]))])

از دست دادن تمرین پس از هر دور تمرین فدرال کاهش می یابد، که نشان می دهد مدل همگرا است. برخی از هشدارهای مهم با این معیارها آموزش وجود دارد، با این حال، به بخش ارزیابی بعد در این آموزش.

نمایش معیارهای مدل در TensorBoard

سپس، بیایید معیارهای این محاسبات فدرال را با استفاده از Tensorboard تجسم کنیم.

بیایید با ایجاد دایرکتوری و خلاصه‌نویس مربوطه برای نوشتن معیارها شروع کنیم.

logdir = "/tmp/logs/scalars/training/"
summary_writer = tf.summary.create_file_writer(logdir)
state = iterative_process.initialize()

معیارهای اسکالر مربوطه را با همان خلاصه نویس ترسیم کنید.

with summary_writer.as_default():
  for round_num in range(1, NUM_ROUNDS):
    state, metrics = iterative_process.next(state, federated_train_data)
    for name, value in metrics['train'].items():
      tf.summary.scalar(name, value, step=round_num)

TensorBoard را با دایرکتوری root log مشخص شده در بالا شروع کنید. ممکن است چند ثانیه طول بکشد تا داده ها بارگیری شوند.

!ls {logdir}
%tensorboard --logdir {logdir} --port=0

events.out.tfevents.1629557449.ebe6e776479e64ea-4903924a278.borgtask.google.com.458912.1.v2
Launching TensorBoard...
Reusing TensorBoard on port 50681 (pid 292785), started 0:30:30 ago. (Use '!kill 292785' to kill it.)
<IPython.core.display.Javascript at 0x7fd6617e02d0>

# Uncomment and run this this cell to clean your directory of old output for
# future graphs from this directory. We don't run it by default so that if 
# you do a "Runtime > Run all" you don't lose your results.

# !rm -R /tmp/logs/scalars/*

به منظور مشاهده معیارهای ارزیابی به صورت یکسان، می توانید یک پوشه eval جداگانه مانند "logs/scalars/eval" ایجاد کنید تا در TensorBoard بنویسید.

سفارشی سازی پیاده سازی مدل

Keras است توصیه می شود-API سطح بالا مدل برای TensorFlow ، و ما را تشویق استفاده از مدل Keras (از طریق tff.learning.from_keras_model ) در TFF دوخته.

با این حال، tff.learning یک رابط مدل های پایین تر، فراهم می کند tff.learning.Model ، که در معرض قابلیت حداقل لازم برای استفاده از یک مدل برای یادگیری فدرال. به طور مستقیم اجرای این رابط (احتمالا هنوز هم با استفاده بلوک مانند tf.keras.layers ) اجازه می دهد تا برای حداکثر سفارشی بدون تغییر داخلی از الگوریتم های یادگیری فدرال.

پس بیایید همه چیز را دوباره از ابتدا انجام دهیم.

تعریف متغیرهای مدل، پاس رو به جلو و معیارها

اولین قدم شناسایی متغیرهای TensorFlow است که قرار است با آنها کار کنیم. برای اینکه کد زیر خواناتر شود، اجازه دهید ساختار داده ای را برای نمایش کل مجموعه تعریف کنیم. این شامل متغیرهایی مانند weights و bias که ما آموزش، و همچنین به عنوان متغیرهای که آمار های مختلف تجمعی و شمارنده ما در طول آموزش به روز رسانی، مانند نگه loss_sum ، accuracy_sum و num_examples .

MnistVariables = collections.namedtuple(
    'MnistVariables', 'weights bias num_examples loss_sum accuracy_sum')

در اینجا روشی وجود دارد که متغیرها را ایجاد می کند. به خاطر سادگی، ما نمایندگی از تمام آمار به عنوان tf.float32 ، همان طور که نیاز به نوع تبدیل در مرحله بعد از بین بردن. بسته بندی Initializer نام متغیر را به عنوان یک نیاز لامبداها اعمال شده توسط است متغیرهای منابع .

def create_mnist_variables():
  return MnistVariables(
      weights=tf.Variable(
          lambda: tf.zeros(dtype=tf.float32, shape=(784, 10)),
          name='weights',
          trainable=True),
      bias=tf.Variable(
          lambda: tf.zeros(dtype=tf.float32, shape=(10)),
          name='bias',
          trainable=True),
      num_examples=tf.Variable(0.0, name='num_examples', trainable=False),
      loss_sum=tf.Variable(0.0, name='loss_sum', trainable=False),
      accuracy_sum=tf.Variable(0.0, name='accuracy_sum', trainable=False))

با وجود متغیرهای پارامترهای مدل و آمار تجمعی، اکنون می‌توانیم روش عبور پیش رو را تعریف کنیم که تلفات را محاسبه می‌کند، پیش‌بینی‌ها را منتشر می‌کند و آمار تجمعی را برای یک دسته از داده‌های ورودی به‌روزرسانی می‌کند.

def predict_on_batch(variables, x):
  return tf.nn.softmax(tf.matmul(x, variables.weights) + variables.bias)

def mnist_forward_pass(variables, batch):
  y = predict_on_batch(variables, batch['x'])
  predictions = tf.cast(tf.argmax(y, 1), tf.int32)

  flat_labels = tf.reshape(batch['y'], [-1])
  loss = -tf.reduce_mean(
      tf.reduce_sum(tf.one_hot(flat_labels, 10) * tf.math.log(y), axis=[1]))
  accuracy = tf.reduce_mean(
      tf.cast(tf.equal(predictions, flat_labels), tf.float32))

  num_examples = tf.cast(tf.size(batch['y']), tf.float32)

  variables.num_examples.assign_add(num_examples)
  variables.loss_sum.assign_add(loss * num_examples)
  variables.accuracy_sum.assign_add(accuracy * num_examples)

  return loss, predictions

سپس، تابعی را تعریف می‌کنیم که مجموعه‌ای از معیارهای محلی را با استفاده از TensorFlow برمی‌گرداند. اینها مقادیری هستند (علاوه بر به‌روزرسانی‌های مدل، که به‌طور خودکار مدیریت می‌شوند) که واجد شرایط جمع‌آوری در سرور در فرآیند یادگیری یا ارزیابی فدرال هستند.

در اینجا، ما به سادگی به طور متوسط بازگشت loss و accuracy ، و همچنین به عنوان num_examples ، که ما باید به درستی وزن سهم از کاربران مختلف در هنگام محاسبه مصالح فدرال.

def get_local_mnist_metrics(variables):
  return collections.OrderedDict(
      num_examples=variables.num_examples,
      loss=variables.loss_sum / variables.num_examples,
      accuracy=variables.accuracy_sum / variables.num_examples)

در نهایت، ما نیاز به تعیین چگونگی جمع معیارهای محلی از طریق ساطع شده از هر یک از دستگاه get_local_mnist_metrics . این تنها بخشی از کد است که در TensorFlow نوشته نشده است - این یک محاسبه فدرال بیان شده در TFF است. اگر شما می خواهم به حفاری عمیق تر، خامه ای بر سفارشی الگوریتم آموزش، اما در اکثر برنامه های کاربردی، شما واقعا نیاز دارید که نه؛ انواع الگوی نشان داده شده در زیر باید کافی باشد. در اینجا به نظر می رسد:

@tff.federated_computation
def aggregate_mnist_metrics_across_clients(metrics):
  return collections.OrderedDict(
      num_examples=tff.federated_sum(metrics.num_examples),
      loss=tff.federated_mean(metrics.loss, metrics.num_examples),
      accuracy=tff.federated_mean(metrics.accuracy, metrics.num_examples))

ورودی metrics مربوط آرگومان به OrderedDict بازگردانده شده توسط get_local_mnist_metrics بالا، اما به شدت از ارزش دیگر هستند tf.Tensors - آنها "جعبه" را به عنوان tff.Value S، آن را به روشن شما دیگر نمی تواند آنها را دستکاری با استفاده از TensorFlow، اما تنها با استفاده از عملگرهای فدرال TFF مانند tff.federated_mean و tff.federated_sum . فرهنگ لغت برگردانده شده از مجموعات جهانی مجموعه معیارهایی را که در سرور در دسترس خواهد بود را تعریف می کند.

ساخت یک نمونه از tff.learning.Model

با در نظر گرفتن همه موارد فوق، ما آماده ساختن یک نمایش مدل برای استفاده با TFF هستیم که مشابه نمونه ای است که برای شما ایجاد می شود زمانی که به TFF اجازه می دهید یک مدل Keras را جذب کند.

from typing import Callable, List, OrderedDict

class MnistModel(tff.learning.Model):

  def __init__(self):
    self._variables = create_mnist_variables()

  @property
  def trainable_variables(self):
    return [self._variables.weights, self._variables.bias]

  @property
  def non_trainable_variables(self):
    return []

  @property
  def local_variables(self):
    return [
        self._variables.num_examples, self._variables.loss_sum,
        self._variables.accuracy_sum
    ]

  @property
  def input_spec(self):
    return collections.OrderedDict(
        x=tf.TensorSpec([None, 784], tf.float32),
        y=tf.TensorSpec([None, 1], tf.int32))

  @tf.function
  def predict_on_batch(self, x, training=True):
    del training
    return predict_on_batch(self._variables, x)

  @tf.function
  def forward_pass(self, batch, training=True):
    del training
    loss, predictions = mnist_forward_pass(self._variables, batch)
    num_exmaples = tf.shape(batch['x'])[0]
    return tff.learning.BatchOutput(
        loss=loss, predictions=predictions, num_examples=num_exmaples)

  @tf.function
  def report_local_outputs(self):
    return get_local_mnist_metrics(self._variables)

  @property
  def federated_output_computation(self):
    return aggregate_mnist_metrics_across_clients

  @tf.function
  def report_local_unfinalized_metrics(
      self) -> OrderedDict[str, List[tf.Tensor]]:
    """Creates an `OrderedDict` of metric names to unfinalized values."""
    return collections.OrderedDict(
        num_examples=[self._variables.num_examples],
        loss=[self._variables.loss_sum, self._variables.num_examples],
        accuracy=[self._variables.accuracy_sum, self._variables.num_examples])

  def metric_finalizers(
      self) -> OrderedDict[str, Callable[[List[tf.Tensor]], tf.Tensor]]:
    """Creates an `OrderedDict` of metric names to finalizers."""
    return collections.OrderedDict(
        num_examples=tf.function(func=lambda x: x[0]),
        loss=tf.function(func=lambda x: x[0] / x[1]),
        accuracy=tf.function(func=lambda x: x[0] / x[1]))

همانطور که می بینید، روش انتزاعی و خواص تعریف شده توسط tff.learning.Model در بخش قبل که متغیرهای معرفی و از دست دادن و آمار تعریف به قطعه کد مربوط.

در اینجا چند نکته قابل ذکر است:

• همه دولت که مدل خود را استفاده خواهد کرد باید به عنوان متغیرهای TensorFlow دستگیر می شود، به عنوان TFF کند پایتون در زمان اجرا استفاده نکنید (به یاد داشته باشید کد شما باید به گونه ای است که می تواند آن را به دستگاه های تلفن همراه مستقر نوشته شود؛ ببینید سفارشی الگوریتم آموزش برای یک بیشتر در عمق تفسیر دلایل).
• مدل شما باید آنچه را شکل داده آن را می پذیرد ( input_spec به طور کلی)، به عنوان، TFF یک محیط به شدت تایپ است و می خواهد برای تعیین امضا نوع برای تمام اجزای. اعلام فرمت ورودی مدل شما یک بخش اساسی از آن است.
• اگر چه از لحاظ فنی مورد نیاز، توصیه می کنیم بسته بندی تمام منطق TensorFlow (پاس رو به جلو، محاسبات متریک، و غیره) به عنوان tf.function بازدید کنندگان، به عنوان این کمک می کند تا اطمینان حاصل شود که TensorFlow می توان سریال، حذف نیاز به وابستگی کنترل صریح است.

موارد فوق برای ارزیابی و الگوریتم هایی مانند Federated SGD کافی است. با این حال، برای میانگین گیری فدرال، باید مشخص کنیم که چگونه مدل باید در هر دسته به صورت محلی آموزش ببیند. هنگام ساخت الگوریتم میانگین‌گیری فدرال، یک بهینه‌ساز محلی را مشخص می‌کنیم.

شبیه سازی آموزش فدرال با مدل جدید

با وجود همه موارد فوق، بقیه فرآیند شبیه چیزی است که قبلاً دیده‌ایم - فقط سازنده مدل را با سازنده کلاس مدل جدید خود جایگزین کنید و از دو محاسبات فدرال در فرآیند تکراری که ایجاد کرده‌اید برای چرخه استفاده کنید. دورهای آموزشی

iterative_process = tff.learning.build_federated_averaging_process(
    MnistModel,
    client_optimizer_fn=lambda: tf.keras.optimizers.SGD(learning_rate=0.02))

state = iterative_process.initialize()

state, metrics = iterative_process.next(state, federated_train_data)
print('round  1, metrics={}'.format(metrics))

round  1, metrics=OrderedDict([('broadcast', ()), ('aggregation', OrderedDict([('mean_value', ()), ('mean_weight', ())])), ('train', OrderedDict([('num_examples', 4860.0), ('loss', 3.0708053), ('accuracy', 0.12777779)])), ('stat', OrderedDict([('num_examples', 4860)]))])

for round_num in range(2, 11):
  state, metrics = iterative_process.next(state, federated_train_data)
  print('round {:2d}, metrics={}'.format(round_num, metrics))

round  2, metrics=OrderedDict([('broadcast', ()), ('aggregation', OrderedDict([('mean_value', ()), ('mean_weight', ())])), ('train', OrderedDict([('num_examples', 4860.0), ('loss', 3.011699), ('accuracy', 0.13024691)])), ('stat', OrderedDict([('num_examples', 4860)]))])
round  3, metrics=OrderedDict([('broadcast', ()), ('aggregation', OrderedDict([('mean_value', ()), ('mean_weight', ())])), ('train', OrderedDict([('num_examples', 4860.0), ('loss', 2.7408307), ('accuracy', 0.15576132)])), ('stat', OrderedDict([('num_examples', 4860)]))])
round  4, metrics=OrderedDict([('broadcast', ()), ('aggregation', OrderedDict([('mean_value', ()), ('mean_weight', ())])), ('train', OrderedDict([('num_examples', 4860.0), ('loss', 2.6761012), ('accuracy', 0.17921811)])), ('stat', OrderedDict([('num_examples', 4860)]))])
round  5, metrics=OrderedDict([('broadcast', ()), ('aggregation', OrderedDict([('mean_value', ()), ('mean_weight', ())])), ('train', OrderedDict([('num_examples', 4860.0), ('loss', 2.675567), ('accuracy', 0.1855967)])), ('stat', OrderedDict([('num_examples', 4860)]))])
round  6, metrics=OrderedDict([('broadcast', ()), ('aggregation', OrderedDict([('mean_value', ()), ('mean_weight', ())])), ('train', OrderedDict([('num_examples', 4860.0), ('loss', 2.5664043), ('accuracy', 0.20329218)])), ('stat', OrderedDict([('num_examples', 4860)]))])
round  7, metrics=OrderedDict([('broadcast', ()), ('aggregation', OrderedDict([('mean_value', ()), ('mean_weight', ())])), ('train', OrderedDict([('num_examples', 4860.0), ('loss', 2.4179392), ('accuracy', 0.24382716)])), ('stat', OrderedDict([('num_examples', 4860)]))])
round  8, metrics=OrderedDict([('broadcast', ()), ('aggregation', OrderedDict([('mean_value', ()), ('mean_weight', ())])), ('train', OrderedDict([('num_examples', 4860.0), ('loss', 2.3237286), ('accuracy', 0.26687244)])), ('stat', OrderedDict([('num_examples', 4860)]))])
round  9, metrics=OrderedDict([('broadcast', ()), ('aggregation', OrderedDict([('mean_value', ()), ('mean_weight', ())])), ('train', OrderedDict([('num_examples', 4860.0), ('loss', 2.1861682), ('accuracy', 0.28209877)])), ('stat', OrderedDict([('num_examples', 4860)]))])
round 10, metrics=OrderedDict([('broadcast', ()), ('aggregation', OrderedDict([('mean_value', ()), ('mean_weight', ())])), ('train', OrderedDict([('num_examples', 4860.0), ('loss', 2.046388), ('accuracy', 0.32037038)])), ('stat', OrderedDict([('num_examples', 4860)]))])

برای مشاهده این معیارها در TensorBoard، به مراحل ذکر شده در بالا در «نمایش معیارهای مدل در TensorBoard» مراجعه کنید.

ارزیابی

همه آزمایش‌های ما تاکنون تنها معیارهای آموزشی یکپارچه را ارائه کرده‌اند - میانگین معیارها در تمام دسته‌های داده‌های آموزش‌دیده شده در همه مشتریان در دور. این موضوع نگرانی‌های معمولی در مورد اضافه‌برازش را معرفی می‌کند، به‌ویژه از آنجایی که ما از مجموعه مشتریان یکسانی در هر دور برای سادگی استفاده می‌کنیم، اما مفهوم دیگری از تطبیق بیش‌ازحد در معیارهای آموزشی خاص برای الگوریتم میانگین‌گیری فدرال وجود دارد. این آسان‌ترین کار است که ببینیم آیا تصور می‌کنیم هر مشتری یک دسته از داده‌ها دارد یا خیر، و ما در آن دسته برای چندین تکرار (دوران) آموزش می‌دهیم. در این مورد، مدل محلی به سرعت دقیقاً با آن دسته مطابقت خواهد داشت، و بنابراین معیار دقت محلی ما به طور میانگین به 1.0 نزدیک می شود. بنابراین، این معیارهای آموزشی را می توان نشانه ای از پیشرفت آموزش دانست، اما نه خیلی بیشتر.

برای انجام ارزیابی بر روی داده های فدرال، شما می توانید یکی دیگر از محاسبات فدرال طراحی فقط برای این منظور ساخت، با استفاده از tff.learning.build_federated_evaluation تابع، و عبور در سازنده مدل خود را به عنوان یک استدلال. توجه داشته باشید که بر خلاف با فدرال به طور متوسط، که در آن ما استفاده کرده ایم MnistTrainableModel ، کافی است به تصویب MnistModel . ارزیابی نزول گرادیان را انجام نمی دهد و نیازی به ساخت بهینه سازها نیست.

برای آزمایش و پژوهش، هنگامی که یک مجموعه داده آزمون متمرکز در دسترس است، فدرال آموزشی برای متن ایجاد نشان می دهد گزینه ارزیابی دیگر: در نظر گرفتن وزن آموزش دیده از یادگیری فدرال، استفاده از آنها را به یک مدل استاندارد Keras، و سپس به سادگی با تماس tf.keras.models.Model.evaluate() در یک مجموعه داده متمرکز است.

evaluation = tff.learning.build_federated_evaluation(MnistModel)

می توانید امضای نوع انتزاعی تابع ارزیابی را به صورت زیر بررسی کنید.

str(evaluation.type_signature)

'(<server_model_weights=<trainable=<float32[784,10],float32[10]>,non_trainable=<>>@SERVER,federated_dataset={<x=float32[?,784],y=int32[?,1]>*}@CLIENTS> -> <eval=<num_examples=float32,loss=float32,accuracy=float32>,stat=<num_examples=int64>>@SERVER)'

بدون نیاز به در مورد جزئیات در این نقطه مربوط می شود، فقط آگاه باشید که آن طول می کشد به شکل زیر به طور کلی، شبیه به tff.templates.IterativeProcess.next اما با دو تفاوت مهم است. اول، ما وضعیت سرور را بر نمی‌گردانیم، زیرا ارزیابی مدل یا هیچ جنبه دیگری از حالت را تغییر نمی‌دهد - می‌توانید آن را بدون حالت در نظر بگیرید. دوم، ارزیابی فقط به مدل نیاز دارد، و به هیچ بخش دیگری از حالت سرور که ممکن است با آموزش مرتبط باشد، مانند متغیرهای بهینه ساز نیاز ندارد.

SERVER_MODEL, FEDERATED_DATA -> TRAINING_METRICS

بیایید از آخرین وضعیتی که در طول آموزش به آن رسیدیم، ارزیابی کنیم. به منظور استخراج آخرین مدل آموزش دیده از حالت سرور، شما به سادگی دسترسی به .model عضو، به شرح زیر.

train_metrics = evaluation(state.model, federated_train_data)

این چیزی است که ما به دست می آوریم. توجه داشته باشید که اعداد بسیار بهتر از آنچه در آخرین دور آموزش بالا گزارش شده است به نظر می رسند. طبق قرارداد، معیارهای آموزشی گزارش شده توسط فرآیند آموزشی تکراری عموماً منعکس کننده عملکرد مدل در ابتدای دور آموزشی هستند، بنابراین معیارهای ارزیابی همیشه یک قدم جلوتر خواهند بود.

str(train_metrics)

"OrderedDict([('eval', OrderedDict([('num_examples', 4860.0), ('loss', 1.7510437), ('accuracy', 0.2788066)])), ('stat', OrderedDict([('num_examples', 4860)]))])"

حال، بیایید یک نمونه آزمایشی از داده‌های فدرال جمع‌آوری کنیم و ارزیابی را روی داده‌های آزمون دوباره اجرا کنیم. داده‌ها از همان نمونه‌ای از کاربران واقعی، اما از مجموعه داده‌های نگهداری‌شده مجزا به دست می‌آیند.

federated_test_data = make_federated_data(emnist_test, sample_clients)

len(federated_test_data), federated_test_data[0]

(10,
 <DatasetV1Adapter shapes: OrderedDict([(x, (None, 784)), (y, (None, 1))]), types: OrderedDict([(x, tf.float32), (y, tf.int32)])>)

test_metrics = evaluation(state.model, federated_test_data)

str(test_metrics)

"OrderedDict([('eval', OrderedDict([('num_examples', 580.0), ('loss', 1.8361608), ('accuracy', 0.2413793)])), ('stat', OrderedDict([('num_examples', 580)]))])"

این آموزش را به پایان می رساند. ما شما را تشویق می‌کنیم که با پارامترها (به عنوان مثال، اندازه دسته، تعداد کاربران، دوره‌ها، نرخ یادگیری و غیره) بازی کنید، کد بالا را برای شبیه‌سازی آموزش بر روی نمونه‌های تصادفی کاربران در هر دور تغییر دهید، و سایر آموزش‌ها را بررسی کنید. ما توسعه داده ایم.