 TensorFlow.orgで見る |  Google Colabで実行 |  GitHubでソースを表示する |
このチュートリアルでは、我々は、TFFの連合学習（FL）API層を導入する古典MNISTトレーニング例を使用tff.learningなど連合学習課題の一般的なタイプを実行するために使用することができる高レベルのインターフェイスのセットを- TensorFlowに実装されたユーザー提供モデルに対するフェデレーショントレーニング。
このチュートリアルとFederated Learning APIは、主にTFFに独自のTensorFlowモデルをプラグインし、後者を主にブラックボックスとして扱いたいユーザーを対象としています。 TFFの詳細と独自の連携学習アルゴリズムの実装方法については、FCコアAPI- カスタム連携アルゴリズムパート1およびパート2のチュートリアルを参照してください。
tff.learning詳細については、 Federated Learning for Text Generationチュートリアルにtff.learningでください。このチュートリアルでは、反復モデルをカバーするだけでなく、事前学習済みのシリアル化されたKerasモデルを読み込んで、Kerasを使用した評価と組み合わせた連携学習による改良を行います。
始める前に
始める前に、次のコマンドを実行して、環境が正しく設定されていることを確認してください。挨拶が表示されない場合は、 インストールガイドを参照してください。
!pip install --quiet --upgrade tensorflow_federated_nightly
!pip install --quiet --upgrade nest_asyncio
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リポジトリにいくつかのデータセットをシードしました。これには、 Leafを使用して再処理された元のNISTデータセットのバージョンを含むMNISTのフェデレーションバージョンが含まれるため、データは数字。各ライターには独自のスタイルがあるため、このデータセットは、フェデレーションデータセットに期待される一種の非iid動作を示します。
これをロードする方法は次のとおりです。
emnist_train, emnist_test = tff.simulation.datasets.emnist.load_data()
返されたデータセットload_data()のインスタンスであるtff.simulation.ClientData 、あなたは、ユーザーのセットを列挙するために構築することを可能にするインタフェースtf.data.Dataset特定のユーザのデータを表し、および照会します個々の要素の構造。このインターフェースを使用してデータセットのコンテンツを探索する方法は次のとおりです。このインターフェイスではクライアントIDを反復処理できますが、これはシミュレーションデータの機能にすぎないことに注意してください。すぐにわかるように、クライアントIDは連携学習フレームワークでは使用されません。その唯一の目的は、シミュレーション用のデータのサブセットを選択できるようにすることです。
len(emnist_train.client_ids)
3383
emnist_train.element_type_structure
OrderedDict([('pixels', TensorSpec(shape=(28, 28), dtype=tf.float32, name=None)), ('label', TensorSpec(shape=(), dtype=tf.int32, 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データを使用して、フェデレーテッドシステムに典型的なデータの異質性のこの概念を探りましょう。これはクライアントのデータのこの詳細な分析は私たちしか利用できないことに注意することが重要です。これはすべてのデータがローカルで利用できるシミュレーション環境だからです。実際の本番環境では、単一のクライアントのデータを検査することはできません。
まず、1つのクライアントのデータのサンプリングを取得して、1つのシミュレートされたデバイスでの例の感触をつかみましょう。私たちが使用しているデータセットは一意のライターによってキー付けされているため、1人のクライアントのデータは、0〜9の数字のサンプルに対する1人の手書きを表し、1人のユーザーの一意の「使用パターン」をシミュレートします。
## 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ラベルのクライアントごとの平均画像を視覚化します。このコードは、1つのラベルに対するユーザーのすべての例の各ピクセル値の平均を生成します。各人の固有の手書きスタイルにより、あるクライアントの数字の平均画像は、同じ数字の別のクライアントの平均画像とは異なって見えることがわかります。各ローカルラウンドでそのユーザー自身の一意のデータから学習しているため、各ローカルトレーニングラウンドが各クライアントで異なる方向にモデルをナッジする方法について考察することができます。チュートリアルの後半で、すべてのクライアントからモデルへの各更新を取得し、それらをクライアントの独自の各データから学習した新しいグローバルモデルに集約する方法について説明します。
# 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')
ユーザーデータはノイズが多く、信頼性の低いラベルが付けられる可能性があります。たとえば、上のクライアント＃2のデータを見ると、ラベル2について、ノイズの多い平均画像を作成する誤ったラベルの例があった可能性があることがわかります。
入力データの前処理
データはすでにtf.data.Datasetであるため、データセット変換を使用して前処理を実行できます。ここでは、 28x28画像を784要素の配列にフラット化し、個々の例をシャッフルし、それらをバッチに編成し、Kerasで使用できるようにpixelsとlabel機能の名前をxとyします。また、データセットを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).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],
[2],
[3],
[6],
[0],
[1],
[4],
[1],
[0],
[6],
[9],
[9],
[3],
[6],
[1],
[4],
[8],
[0],
[2]], dtype=int32))])
フェデレーションデータセットを構築するためのほぼすべての構成要素が用意されています。
シミュレーションでフェデレーションデータをTFFに供給する方法の1つは、単純なPythonリストであり、リストの各要素は、リストまたは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.Input(shape=(784,)),
tf.keras.layers.Dense(10, kernel_initializer='zeros'),
tf.keras.layers.Softmax(),
])
TFFで任意のモデルを使用するには、 tff.learning.Modelインターフェースのインスタンスでラップする必要があります。これは、 tff.learning.Modelと同様に、モデルの転送パス、メタデータプロパティなどにスタンプをtff.learning.Modelメソッドを公開しますが、追加のフェデレーテッドメトリックの計算プロセスを制御する方法などの要素。今のところ、これについて心配する必要はありません。上記で定義したようなKerasモデルがある場合は、 tff.learning.from_keras_modelを呼び出して、以下に示すようにモデルとサンプルデータバッチを引数として渡すことで、TFFでラップすることができます。
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で使用するモデルをtff.learning.Modelとしてtff.learning.Modelしたtff.learning.build_federated_averaging_process 、次のようにヘルパー関数tff.learning.build_federated_averaging_processを呼び出すことにより、TFFにtff.learning.build_federated_averaging_process平均アルゴリズムを構築させることができます。
引数はコンストラクター（上記のmodel_fnなど）ではなく、既に構築されたインスタンスである必要があることに注意してください。これにより、モデルの構築がTFFによって制御されるコンテキストで行われるようになります（理由についてmodel_fn場合）これについては、 カスタムアルゴリズムに関するフォローアップチュートリアルを読むことをお勧めします）。
_client オプティマイザと_server オプティマイザ ：以下フェデレーテッド平均化アルゴリズムの1つの重要なノートでは、2つのオプティマイザがあります。 _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は1組の連合計算を構築し、それらをtff.templates.IterativeProcessにパッケージ化しました。これらの計算は、ペアのプロパティinitializeおよびnextとして使用できます。
一言で言えば、 連合計算は、さまざまな連合アルゴリズムを表現できるTFFの内部言語のプログラムです（これについては、 カスタムアルゴリズムのチュートリアルで詳しく説明しています）。この場合、生成されてiterative_processパックされた2つの計算はFederated Averagingを実装します。
TFFの目標は、実際の統合学習設定で実行できるように計算を定義することですが、現在はローカル実行シミュレーションランタイムのみが実装されています。シミュレータで計算を実行するには、Python関数のように呼び出すだけです。このデフォルトのインタープリター環境は、高パフォーマンス用に設計されていませんが、このチュートリアルでは十分です。今後のリリースで大規模な研究を促進するために、より高性能なシミュレーションランタイムを提供する予定です。
まず、 initialize計算から始めましょう。すべての連合計算の場合と同様に、それは関数と考えることができます。計算は引数を取らず、1つの結果（サーバー上のフェデレーテッドアベレージングプロセスの状態の表現）を返します。 TFFの詳細については詳しく説明しませんが、この状態がどのように見えるかを確認することは有益です。次のように視覚化できます。
str(iterative_process.initialize.type_signature)
'( -> <model=<trainable=<float32[784,10],float32[10]>,non_trainable=<>>,optimizer_state=<int64>,delta_aggregate_state=<>,model_broadcast_state=<>>@SERVER)'
上記のタイプシグネチャは最初は少しわかりにくいかもしれませんが、サーバーの状態はmodel （すべてのデバイスに配布されるMNISTの初期モデルパラメーター）とoptimizer_state （サーバーによって維持される追加情報）で構成されていることを認識できます。ハイパーパラメータスケジュールなどに使用するラウンド数など）。
サーバー状態を構築するためにinitialize計算を呼び出しましょう。
state = iterative_process.initialize()
フェデレーション計算のペアの2番目、 nextは、サーバーの状態（モデルパラメーターを含む）をクライアントにプッシュすること、ローカルデータのデバイス上のトレーニング、モデルの更新の収集と平均化で構成されるフェデレーテッドアベレージングの単一ラウンドを表します、サーバーで新しい更新モデルを作成します。
概念的には、 nextはnextような関数型シグネチャを持つと考えることができます。
SERVER_STATE, FEDERATED_DATA -> SERVER_STATE, TRAINING_METRICS
特に、 next()はサーバー上で実行される関数ではなく、分散型計算全体の宣言的な関数表現であると考える必要があります-一部の入力はサーバー（ SERVER_STATE ）によって提供されますが、それぞれが関与していますデバイスは独自のローカルデータセットを提供します。
1回のトレーニングを実行して、結果を視覚化しましょう。上記ですでに生成したフェデレーションデータをユーザーのサンプルに使用できます。
state, metrics = iterative_process.next(state, federated_train_data)
print('round 1, metrics={}'.format(metrics))
round 1, metrics=<broadcast=<>,aggregation=<>,train=<sparse_categorical_accuracy=0.12037037312984467,loss=3.0108425617218018>>
もう少しラウンドを実行してみましょう。前述のように、通常、この時点で、ユーザーが継続的に出入りする現実的な展開をシミュレートするために、ランダムに選択された各ラウンドのユーザーのサンプルからシミュレーションデータのサブセットを選択しますが、このインタラクティブなノートブックでは、デモンストレーションのために、同じユーザーを再利用して、システムがすばやく収束するようにします。
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=<broadcast=<>,aggregation=<>,train=<sparse_categorical_accuracy=0.14814814925193787,loss=2.8865506649017334>> round 3, metrics=<broadcast=<>,aggregation=<>,train=<sparse_categorical_accuracy=0.148765429854393,loss=2.9079062938690186>> round 4, metrics=<broadcast=<>,aggregation=<>,train=<sparse_categorical_accuracy=0.17633745074272156,loss=2.724686622619629>> round 5, metrics=<broadcast=<>,aggregation=<>,train=<sparse_categorical_accuracy=0.20226337015628815,loss=2.6334855556488037>> round 6, metrics=<broadcast=<>,aggregation=<>,train=<sparse_categorical_accuracy=0.22427983582019806,loss=2.5482592582702637>> round 7, metrics=<broadcast=<>,aggregation=<>,train=<sparse_categorical_accuracy=0.24094650149345398,loss=2.4472343921661377>> round 8, metrics=<broadcast=<>,aggregation=<>,train=<sparse_categorical_accuracy=0.259876549243927,loss=2.3809611797332764>> round 9, metrics=<broadcast=<>,aggregation=<>,train=<sparse_categorical_accuracy=0.29814815521240234,loss=2.156442403793335>> round 10, metrics=<broadcast=<>,aggregation=<>,train=<sparse_categorical_accuracy=0.31687241792678833,loss=2.122845411300659>>
フェデレーショントレーニングの各ラウンドの後、トレーニングの損失は減少しており、モデルが収束していることを示しています。これらのトレーニング指標にはいくつかの重要な注意事項がありますが、このチュートリアルの後半の評価に関するセクションを参照してください。
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._asdict().items():
tf.summary.scalar(name, value, step=round_num)
上記で指定したルートログディレクトリを使用してTensorBoardを起動します。データの読み込みには数秒かかる場合があります。
%tensorboard --logdir /tmp/logs/scalars/ --port=0
# Run this this cell to clean your directory of old output for future graphs from this directory.
!rm -R /tmp/logs/scalars/*
同じ方法で評価指標を表示するには、「logs / scalars / eval」のような個別のevalフォルダーを作成して、TensorBoardに書き込むことができます。
モデル実装のカスタマイズ
KerasはTensorFlowの推奨される高レベルモデルAPIであり 、可能な限りTFFで（ tff.learning.from_keras_modelを介して） tff.learning.from_keras_modelモデルを使用することをお勧めします。
しかし、 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として表します。これにより、後の段階での型変換の必要がなくなるためです。変数初期化子をラムダとしてラップすることは、 リソース変数によって課せられる要件です 。
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 mnist_forward_pass(variables, batch):
y = tf.nn.softmax(tf.matmul(batch['x'], variables.weights) + variables.bias)
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だけ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使用して、各デバイスによって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だけ、それはあなたがもはやTensorFlowを使用してそれらを操作することができますクリアしない作るために、S tff.federated_meanやtff.federated_sumなどのTFFの連合演算子を使用する。返されるグローバル集計のディクショナリは、サーバーで使用できる一連のメトリックを定義します。
tff.learning.Modelインスタンスのtff.learning.Model
上記のすべての準備が整ったら、TFFにKerasモデルを取り込んだときに生成されるものと同様のTFFで使用するモデル表現を構築する準備が整いました。
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 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
ご覧のとおり、 tff.learning.Modelによって定義された抽象メソッドとプロパティは、変数を導入し、損失と統計を定義した前のセクションのコードスニペットに対応しています。
ここで強調する価値があるいくつかのポイントがあります：
- TFFは実行時にPythonを使用しないため、モデルが使用するすべての状態をTensorFlow変数としてキャプチャする必要があります（モバイルデバイスにデプロイできるようにコードを記述する必要があります。詳細については、 カスタムアルゴリズムのチュートリアルを参照してください）理由についての解説）。
- 一般に、TFFは強く型付けされた環境であり、すべてのコンポーネントの型シグネチャを決定する必要があるため、モデルは、受け入れるデータの形式（
input_spec）を記述する必要があります。モデルの入力のフォーマットを宣言することは、モデルの重要な部分です。 - 技術的には必須ではありませんが、すべてのTensorFlowロジック（転送パス、メトリック計算など）を
tf.functionとしてラップすることをお勧めします。これにより、TensorFlowを確実にシリアル化でき、明示的な制御依存関係の必要性がなくなります。
上記は、Federated SGDのような評価とアルゴリズムには十分です。ただし、フェデレーテッドアベレージングの場合、モデルが各バッチでローカルにトレーニングする方法を指定する必要があります。フェデレーテッドアベレージングアルゴリズムを構築するときに、ローカルオプティマイザーを指定します。
新しいモデルを使用したフェデレーショントレーニングのシミュレーション
上記のすべてが整ったので、プロセスの残りの部分はすでに見たもののように見えます-モデルコンストラクターを新しいモデルクラスのコンストラクターに置き換え、作成した反復プロセスで2つのフェデレーション計算を使用して循環しますトレーニングラウンド。
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=<broadcast=<>,aggregation=<>,train=<num_examples=4860.0,loss=2.9713594913482666,accuracy=0.13518518209457397>>
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=<broadcast=<>,aggregation=<>,train=<num_examples=4860.0,loss=2.975412607192993,accuracy=0.14032921195030212>> round 3, metrics=<broadcast=<>,aggregation=<>,train=<num_examples=4860.0,loss=2.9395227432250977,accuracy=0.1594650149345398>> round 4, metrics=<broadcast=<>,aggregation=<>,train=<num_examples=4860.0,loss=2.710164785385132,accuracy=0.17139917612075806>> round 5, metrics=<broadcast=<>,aggregation=<>,train=<num_examples=4860.0,loss=2.5891618728637695,accuracy=0.20267489552497864>> round 6, metrics=<broadcast=<>,aggregation=<>,train=<num_examples=4860.0,loss=2.5148487091064453,accuracy=0.21666666865348816>> round 7, metrics=<broadcast=<>,aggregation=<>,train=<num_examples=4860.0,loss=2.2816808223724365,accuracy=0.2580246925354004>> round 8, metrics=<broadcast=<>,aggregation=<>,train=<num_examples=4860.0,loss=2.3656885623931885,accuracy=0.25884774327278137>> round 9, metrics=<broadcast=<>,aggregation=<>,train=<num_examples=4860.0,loss=2.23549222946167,accuracy=0.28477364778518677>> round 10, metrics=<broadcast=<>,aggregation=<>,train=<num_examples=4860.0,loss=1.974222183227539,accuracy=0.35329216718673706>>
TensorBoard内でこれらのメトリックを表示するには、上記の「TensorBoardでのモデルメトリックの表示」にリストされている手順を参照してください。
評価
これまでのすべての実験では、フェデレーショントレーニングメトリックのみを表示していました-ラウンドのすべてのクライアントでトレーニングされたデータのすべてのバッチの平均メトリック。これは、特に単純化のために各ラウンドで同じクライアントのセットを使用したため、過剰適合に関する通常の懸念をもたらしますが、フェデレーテッドアベレージングアルゴリズムに固有のトレーニングメトリックには過剰適合の概念があります。これは、各クライアントに単一のデータバッチがあり、そのバッチで多くの反復（エポック）をトレーニングすると想定すると、最も簡単です。この場合、ローカルモデルはその1つのバッチにすばやく正確にフィットするため、平均化するローカル精度メトリックは1.0に近づきます。したがって、これらのトレーニング指標は、トレーニングが進んでいることを示すものと見なすことができますが、それ以上ではありません。
フェデレーションデータの評価を実行するには、 tff.learning.build_federated_evaluation関数を使用して、モデルコンストラクターを引数として渡すだけで、この目的のために設計された別のフェデレーション計算を構築できます。 MnistTrainableModelを使用したMnistTrainableModelとは異なり、 MnistModelを渡すだけで十分MnistModel 。評価は勾配降下法を実行せず、オプティマイザを構築する必要はありません。
実験と研究のために、集中テストデータセットが利用可能な場合、 テキスト生成用の統合学習は、別の評価オプションを示します：統合学習からトレーニング済みの重みを取得し、標準のKerasモデルに適用してから、単にtf.keras.models.Model.evaluate()呼び出しtf.keras.models.Model.evaluate()一元化されたデータセットに対するtf.keras.models.Model.evaluate() 。
evaluation = tff.learning.build_federated_evaluation(MnistModel)
評価関数の抽象型シグネチャは、次のように検査できます。
str(evaluation.type_signature)
'(<<trainable=<float32[784,10],float32[10]>,non_trainable=<>>@SERVER,{<x=float32[?,784],y=int32[?,1]>*}@CLIENTS> -> <num_examples=float32@SERVER,loss=float32@SERVER,accuracy=float32@SERVER>)'
この時点では詳細について気にする必要はありませんtff.templates.IterativeProcess.nextに似ていtff.templates.IterativeProcess.nextが、2つの重要な違いがあるという次の一般的な形式をとることに注意してください。まず、評価によってモデルや状態の他の側面が変更されることはないため、サーバーの状態は返されません。これはステートレスと考えることができます。第2に、評価にはモデルのみが必要であり、オプティマイザ変数など、トレーニングに関連する可能性のあるサーバーの状態の他の部分は必要ありません。
SERVER_MODEL, FEDERATED_DATA -> TRAINING_METRICS
トレーニング中に到達した最新の状態の評価を呼び出しましょう。サーバーの状態から最新のトレーニング済みモデルを抽出するには、次のように.modelメンバーにアクセスするだけです。
train_metrics = evaluation(state.model, federated_train_data)
ここに私たちが得るものがあります。上記のトレーニングの最後のラウンドで報告されたものよりわずかに良く見えることに注意してください。慣例により、反復トレーニングプロセスによって報告されるトレーニング指標は、通常、トレーニングラウンドの開始時のモデルのパフォーマンスを反映しているため、評価指標は常に1ステップ先にあります。
str(train_metrics)
'<num_examples=4860.0,loss=1.7142657041549683,accuracy=0.38683128356933594>'
ここで、フェデレーションデータのテストサンプルをコンパイルして、テストデータの評価を再実行してみましょう。データは実際のユーザーの同じサンプルから取得されますが、明確に保持されたデータセットから取得されます。
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)
'<num_examples=580.0,loss=1.861915111541748,accuracy=0.3362068831920624>'
これでチュートリアルは終了です。パラメーター（バッチサイズ、ユーザー数、エポック、学習率など）を試して、上記のコードを変更し、各ラウンドでユーザーのランダムサンプルのトレーニングをシミュレートし、他のチュートリアルを探索することをお勧めします私たちは開発しました。