 TensorFlow.org এ দেখুন |  Google Colab-এ চালান |  GitHub-এ উৎস দেখুন |  নোটবুক ডাউনলোড করুন |
এই টিউটোরিয়ালটি দেখায় কিভাবে TFF একটি খুব বড় মডেল যেখানে প্রতিটি ক্লায়েন্টের ডিভাইস শুধুমাত্র ডাউনলোডসমূহ প্রশিক্ষণের ব্যবহার করা যেতে পারে এবং মডেল একটি ছোট অংশ আপডেট ব্যবহার tff.federated_select এবং বিক্ষিপ্ত অ্যাগ্রিগেশন। স্বয়ংসম্পূর্ণ, যদিও এই টিউটোরিয়াল মোটামুটি হয় tff.federated_select টিউটোরিয়াল এবং কাস্টম এফএল আলগোরিদিম টিউটোরিয়াল কৌশল কিছু এখানে ব্যবহার করা ভাল ভূমিকা প্রদান।
এককভাবে, এই টিউটোরিয়ালে আমরা বহু-লেবেল শ্রেণিবিন্যাসের জন্য লজিস্টিক রিগ্রেশন বিবেচনা করি, কোন "ট্যাগগুলি" একটি ব্যাগ-অফ-শব্দ বৈশিষ্ট্য উপস্থাপনার উপর ভিত্তি করে একটি পাঠ্য স্ট্রিংয়ের সাথে সম্পর্কিত। গুরুত্বপূর্ণভাবে, যোগাযোগ এবং ক্লায়েন্ট-সাইড গণনার খরচ একটি নির্দিষ্ট ধ্রুবক (দ্বারা নিয়ন্ত্রিত হয় MAX_TOKENS_SELECTED_PER_CLIENT ), এবং সামগ্রিক শব্দভান্ডার আকার, যা ব্যবহারিক সেটিংসে অত্যন্ত বড় হতে পারে সঙ্গে স্কেল না।
!pip install --quiet --upgrade tensorflow-federated-nightly
!pip install --quiet --upgrade nest-asyncio
import nest_asyncio
nest_asyncio.apply()
import collections
import itertools
import numpy as np
from typing import Callable, List, Tuple
import tensorflow as tf
import tensorflow_federated as tff
tff.backends.native.set_local_python_execution_context()
প্রতিটি ক্লায়েন্ট হবে federated_select সর্বাধিক অনেক অনন্য টোকেন জন্য মডেল ওজন সারি। এই ক্লায়েন্টের স্থানীয় মডেল আকার এবং সার্ভারের পরিমাণ উপরের বাউন্ড -> ক্লায়েন্ট ( federated_select ) এবং ক্লায়েন্ট -> সার্ভার (federated_aggregate ) যোগাযোগ সঞ্চালিত।
এই টিউটোরিয়ালটি এখনও সঠিকভাবে চালানো উচিত যদি আপনি এটিকে 1-এর মতো ছোট সেট করেন (প্রত্যেক ক্লায়েন্ট থেকে সমস্ত টোকেন নির্বাচন করা না হয় তা নিশ্চিত করে) বা একটি বড় মান, যদিও মডেল কনভারজেন্স প্রভাবিত হতে পারে।
MAX_TOKENS_SELECTED_PER_CLIENT = 6
আমরা বিভিন্ন ধরণের জন্য কয়েকটি ধ্রুবক সংজ্ঞায়িত করি। এই colab, একটি টোকেন ডেটা সেটটি পার্স পর একটি নির্দিষ্ট শব্দের জন্য একটি পূর্ণসংখ্যা আইডেন্টিফায়ার।
# There are some constraints on types
# here that will require some explicit type conversions:
# - `tff.federated_select` requires int32
# - `tf.SparseTensor` requires int64 indices.
TOKEN_DTYPE = tf.int64
SELECT_KEY_DTYPE = tf.int32
# Type for counts of token occurences.
TOKEN_COUNT_DTYPE = tf.int32
# A sparse feature vector can be thought of as a map
# from TOKEN_DTYPE to FEATURE_DTYPE.
# Our features are {0, 1} indicators, so we could potentially
# use tf.int8 as an optimization.
FEATURE_DTYPE = tf.int32
সমস্যা সেট আপ করা হচ্ছে: ডেটাসেট এবং মডেল
এই টিউটোরিয়ালে সহজে পরীক্ষা-নিরীক্ষার জন্য আমরা একটি ছোট খেলনা ডেটাসেট তৈরি করি। যাইহোক, ডেটাসেটের বিন্যাস সঙ্গে সামঞ্জস্যপূর্ণ ফেডারেটেড Stackoverflow , এবং প্রাক প্রক্রিয়াকরণ এবং মডেল স্থাপত্য এর Stackoverflow ট্যাগ ভবিষ্যদ্বাণী সমস্যা থেকে গৃহীত হয় অভিযোজিত ফেডারেটেড অপ্টিমাইজেশান ।
ডেটাসেট পার্সিং এবং প্রাক-প্রক্রিয়াকরণ
NUM_OOV_BUCKETS = 1
BatchType = collections.namedtuple('BatchType', ['tokens', 'tags'])
def build_to_ids_fn(word_vocab: List[str],
tag_vocab: List[str]) -> Callable[[tf.Tensor], tf.Tensor]:
"""Constructs a function mapping examples to sequences of token indices."""
word_table_values = np.arange(len(word_vocab), dtype=np.int64)
word_table = tf.lookup.StaticVocabularyTable(
tf.lookup.KeyValueTensorInitializer(word_vocab, word_table_values),
num_oov_buckets=NUM_OOV_BUCKETS)
tag_table_values = np.arange(len(tag_vocab), dtype=np.int64)
tag_table = tf.lookup.StaticVocabularyTable(
tf.lookup.KeyValueTensorInitializer(tag_vocab, tag_table_values),
num_oov_buckets=NUM_OOV_BUCKETS)
def to_ids(example):
"""Converts a Stack Overflow example to a bag-of-words/tags format."""
sentence = tf.strings.join([example['tokens'], example['title']],
separator=' ')
# We represent that label (output tags) densely.
raw_tags = example['tags']
tags = tf.strings.split(raw_tags, sep='|')
tags = tag_table.lookup(tags)
tags, _ = tf.unique(tags)
tags = tf.one_hot(tags, len(tag_vocab) + NUM_OOV_BUCKETS)
tags = tf.reduce_max(tags, axis=0)
# We represent the features as a SparseTensor of {0, 1}s.
words = tf.strings.split(sentence)
tokens = word_table.lookup(words)
tokens, _ = tf.unique(tokens)
# Note: We could choose to use the word counts as the feature vector
# instead of just {0, 1} values (see tf.unique_with_counts).
tokens = tf.reshape(tokens, shape=(tf.size(tokens), 1))
tokens_st = tf.SparseTensor(
tokens,
tf.ones(tf.size(tokens), dtype=FEATURE_DTYPE),
dense_shape=(len(word_vocab) + NUM_OOV_BUCKETS,))
tokens_st = tf.sparse.reorder(tokens_st)
return BatchType(tokens_st, tags)
return to_ids
def build_preprocess_fn(word_vocab, tag_vocab):
@tf.function
def preprocess_fn(dataset):
to_ids = build_to_ids_fn(word_vocab, tag_vocab)
# We *don't* shuffle in order to make this colab deterministic for
# easier testing and reproducibility.
# But real-world training should use `.shuffle()`.
return dataset.map(to_ids, num_parallel_calls=tf.data.experimental.AUTOTUNE)
return preprocess_fn
একটি ছোট খেলনা ডেটাসেট
আমরা 12টি শব্দ এবং 3টি ক্লায়েন্টের বিশ্বব্যাপী শব্দভাণ্ডার সহ একটি ছোট খেলনা ডেটাসেট তৈরি করি। এই অতি ক্ষুদ্র উদাহরণ প্রান্ত মামলা পরীক্ষার জন্য উপযোগী (উদাহরণস্বরূপ, আমরা কম সঙ্গে দুই ক্লায়েন্ট MAX_TOKENS_SELECTED_PER_CLIENT = 6 এবং কোড উন্নয়নশীল এক স্বতন্ত্র টোকেন, এবং আরও সহ)।
যাইহোক, এই পদ্ধতির বাস্তব-বিশ্ব ব্যবহারের ক্ষেত্রে 10 মিলিয়ন বা তার বেশি বিশ্বব্যাপী শব্দভাণ্ডার হবে, প্রতিটি ক্লায়েন্টে সম্ভবত 1000 এর মতো স্বতন্ত্র টোকেন প্রদর্শিত হবে। কারণ ডেটার বিন্যাস একই, আরো বাস্তবসম্মত টেস্টবেড সমস্যা এক্সটেনশন, যেমন tff.simulation.datasets.stackoverflow.load_data() ডেটা সেটটি, সোজাসাপটা হওয়া উচিত।
প্রথমত, আমরা আমাদের শব্দ এবং ট্যাগ শব্দভান্ডার সংজ্ঞায়িত করি।
# Features
FRUIT_WORDS = ['apple', 'orange', 'pear', 'kiwi']
VEGETABLE_WORDS = ['carrot', 'broccoli', 'arugula', 'peas']
FISH_WORDS = ['trout', 'tuna', 'cod', 'salmon']
WORD_VOCAB = FRUIT_WORDS + VEGETABLE_WORDS + FISH_WORDS
# Labels
TAG_VOCAB = ['FRUIT', 'VEGETABLE', 'FISH']
এখন, আমরা ছোট স্থানীয় ডেটাসেট সহ 3 ক্লায়েন্ট তৈরি করি। আপনি যদি colab-এ এই টিউটোরিয়ালটি চালান, তাহলে এই সেল এবং এর আউটপুটকে পিন করতে "ট্যাবে মিরর সেল" বৈশিষ্ট্যটি ব্যবহার করা উপযোগী হতে পারে যাতে নীচের বিকাশকৃত ফাংশনগুলির আউটপুট ব্যাখ্যা/চেক করা যায়।
preprocess_fn = build_preprocess_fn(WORD_VOCAB, TAG_VOCAB)
def make_dataset(raw):
d = tf.data.Dataset.from_tensor_slices(
# Matches the StackOverflow formatting
collections.OrderedDict(
tokens=tf.constant([t[0] for t in raw]),
tags=tf.constant([t[1] for t in raw]),
title=['' for _ in raw]))
d = preprocess_fn(d)
return d
# 4 distinct tokens
CLIENT1_DATASET = make_dataset([
('apple orange apple orange', 'FRUIT'),
('carrot trout', 'VEGETABLE|FISH'),
('orange apple', 'FRUIT'),
('orange', 'ORANGE|CITRUS') # 2 OOV tag
])
# 6 distinct tokens
CLIENT2_DATASET = make_dataset([
('pear cod', 'FRUIT|FISH'),
('arugula peas', 'VEGETABLE'),
('kiwi pear', 'FRUIT'),
('sturgeon', 'FISH'), # OOV word
('sturgeon bass', 'FISH') # 2 OOV words
])
# A client with all possible words & tags (13 distinct tokens).
# With MAX_TOKENS_SELECTED_PER_CLIENT = 6, we won't download the model
# slices for all tokens that occur on this client.
CLIENT3_DATASET = make_dataset([
(' '.join(WORD_VOCAB + ['oovword']), '|'.join(TAG_VOCAB)),
# Mathe the OOV token and 'salmon' occur in the largest number
# of examples on this client:
('salmon oovword', 'FISH|OOVTAG')
])
print('Word vocab')
for i, word in enumerate(WORD_VOCAB):
print(f'{i:2d} {word}')
print('\nTag vocab')
for i, tag in enumerate(TAG_VOCAB):
print(f'{i:2d} {tag}')
Word vocab 0 apple 1 orange 2 pear 3 kiwi 4 carrot 5 broccoli 6 arugula 7 peas 8 trout 9 tuna 10 cod 11 salmon Tag vocab 0 FRUIT 1 VEGETABLE 2 FISH
ইনপুট বৈশিষ্ট্য (টোকেন/শব্দ) এবং লেবেল (পোস্ট ট্যাগ) এর কাঁচা সংখ্যার জন্য ধ্রুবক সংজ্ঞায়িত করুন। আমাদের প্রকৃত ইনপুট / আউটপুট শূণ্যস্থান আছে NUM_OOV_BUCKETS = 1 , কারণ আমরা একটি OOV টোকেন / ট্যাগ যুক্ত বড় করা হয়েছে।
NUM_WORDS = len(WORD_VOCAB)
NUM_TAGS = len(TAG_VOCAB)
WORD_VOCAB_SIZE = NUM_WORDS + NUM_OOV_BUCKETS
TAG_VOCAB_SIZE = NUM_TAGS + NUM_OOV_BUCKETS
ডেটাসেটগুলির ব্যাচ করা সংস্করণ এবং পৃথক ব্যাচগুলি তৈরি করুন, যা আমরা যাওয়ার সাথে সাথে কোড পরীক্ষা করার জন্য কার্যকর হবে৷
batched_dataset1 = CLIENT1_DATASET.batch(2)
batched_dataset2 = CLIENT2_DATASET.batch(3)
batched_dataset3 = CLIENT3_DATASET.batch(2)
batch1 = next(iter(batched_dataset1))
batch2 = next(iter(batched_dataset2))
batch3 = next(iter(batched_dataset3))
স্পার্স ইনপুট সহ একটি মডেল সংজ্ঞায়িত করুন
আমরা প্রতিটি ট্যাগের জন্য একটি সাধারণ স্বাধীন লজিস্টিক রিগ্রেশন মডেল ব্যবহার করি।
def create_logistic_model(word_vocab_size: int, vocab_tags_size: int):
model = tf.keras.models.Sequential([
tf.keras.layers.InputLayer(input_shape=(word_vocab_size,), sparse=True),
tf.keras.layers.Dense(
vocab_tags_size,
activation='sigmoid',
kernel_initializer=tf.keras.initializers.zeros,
# For simplicity, don't use a bias vector; this means the model
# is a single tensor, and we only need sparse aggregation of
# the per-token slices of the model. Generalizing to also handle
# other model weights that are fully updated
# (non-dense broadcast and aggregate) would be a good exercise.
use_bias=False),
])
return model
প্রথমে ভবিষ্যদ্বাণী করে এটি কাজ করে তা নিশ্চিত করুন:
model = create_logistic_model(WORD_VOCAB_SIZE, TAG_VOCAB_SIZE)
p = model.predict(batch1.tokens)
print(p)
[[0.5 0.5 0.5 0.5] [0.5 0.5 0.5 0.5]]
এবং কিছু সাধারণ কেন্দ্রীভূত প্রশিক্ষণ:
model.compile(optimizer=tf.keras.optimizers.Adagrad(learning_rate=0.001),
loss=tf.keras.losses.BinaryCrossentropy())
model.train_on_batch(batch1.tokens, batch1.tags)
ফেডারেটেড গণনার জন্য বিল্ডিং ব্লক
আমরা একটি সহজ সংস্করণ বাস্তবায়ন করবে ফেডারেটেড গড় কী পার্থক্য সঙ্গে অ্যালগরিদম প্রতিটি ডিভাইস শুধুমাত্র মডেলের একটি প্রাসঙ্গিক উপসেট ডাউনলোড, এবং শুধুমাত্র যে উপসেট যাও যাও আপডেট ভূমিকা রাখে।
আমরা ব্যবহার M জন্য সংক্ষেপে MAX_TOKENS_SELECTED_PER_CLIENT । একটি উচ্চ স্তরে, প্রশিক্ষণের এক রাউন্ডে এই পদক্ষেপগুলি অন্তর্ভুক্ত থাকে:
প্রতিটি অংশগ্রহণকারী ক্লায়েন্ট তার স্থানীয় ডেটাসেটের উপর স্ক্যান করে, ইনপুট স্ট্রিংগুলি পার্স করে এবং সঠিক টোকেনগুলিতে (int indexes) ম্যাপ করে। এই সর্বব্যাপী (বড়) অভিধান অ্যাক্সেস (এই সম্ভাব্য ব্যবহার এড়ানো যেত প্রয়োজন বৈশিষ্ট্য হ্যাশ কৌশল)। তারপরে প্রতিটি টোকেন কতবার ঘটে তা আমরা খুব কমই গণনা করি। যদি
Uঅনন্য টোকেন ডিভাইসে ঘটে, আমরা নির্বাচনnum_actual_tokens = min(U, M)ট্রেন সবচেয়ে ঘন টোকেন।ক্লায়েন্ট ব্যবহার
federated_selectজন্য মডেল কোফিসিয়েন্টস পুনরুদ্ধার করতেnum_actual_tokensসার্ভার থেকে নির্বাচিত টোকেন। প্রতিটি মডেল ফালি আকৃতি একটি টেন্সর হয়(TAG_VOCAB_SIZE, ), তাই মোট ক্লায়েন্টের সাথে প্রেরিত ডেটা আকার সবচেয়ে হয়TAG_VOCAB_SIZE * M(নীচে নোট দেখুন)।ক্লায়েন্ট একটি ম্যাপিং গঠন করা
global_token -> local_tokenযেখানে স্থানীয় টোকেন (int- সূচক) নির্বাচিত টোকেন তালিকায় বিশ্বব্যাপী টোকেন এর সূচি।ক্লায়েন্ট বিশ্বব্যাপী মডেল শুধুমাত্র সর্বাধিক জন্য কোফিসিয়েন্টস আছে একটি "ছোট" সংস্করণ ব্যবহার
Mপরিসীমা থেকে, টোকেন[0, num_actual_tokens)।global -> localম্যাপিং নির্বাচিত মডেল টুকরা থেকে এই মডেলের ঘন পরামিতি আরম্ভ করতে ব্যবহৃত হয়।ক্লায়েন্টদের সাথে preprocessed ডেটার উপর SGD ব্যবহার তাদের স্থানীয় মডেল প্রশিক্ষণ
global -> localম্যাপিং।ক্লায়েন্টদের মধ্যে তাদের স্থানীয় মডেলের পরামিতি চালু
IndexedSlicesব্যবহার আপডেটlocal -> globalসারি সূচক ম্যাপিং। সার্ভার একটি স্পার্স সমষ্টি ব্যবহার করে এই আপডেটগুলিকে একত্রিত করে।সার্ভার উপরের সমষ্টির (ঘন) ফলাফল নেয়, অংশগ্রহণকারী ক্লায়েন্টদের সংখ্যা দ্বারা ভাগ করে এবং গ্লোবাল মডেলে ফলাফল গড় আপডেট প্রয়োগ করে।
এই বিভাগে আমরা এই পদক্ষেপগুলি যা পরে একটি চূড়ান্ত সংমিশ্রণে হবে বিল্ডিং ব্লক গঠন করা federated_computation যে যেমনটি এক প্রশিক্ষণ বৃত্তাকার পূর্ণ যুক্তি।
ক্লায়েন্ট টোকেন গণনা এবং যা মডেল টুকরা করার সিদ্ধান্ত নেন federated_select
প্রতিটি ডিভাইসকে সিদ্ধান্ত নিতে হবে যে মডেলের কোন "স্লাইস" তার স্থানীয় প্রশিক্ষণ ডেটাসেটের সাথে প্রাসঙ্গিক। আমাদের সমস্যার জন্য, আমরা ক্লায়েন্ট প্রশিক্ষণ ডেটা সেটে প্রতিটি টোকেন কতগুলি উদাহরণ রয়েছে তা গণনা করে (অল্প পরিমাণে!) এটি করি।
@tf.function
def token_count_fn(token_counts, batch):
"""Adds counts from `batch` to the running `token_counts` sum."""
# Sum across the batch dimension.
flat_tokens = tf.sparse.reduce_sum(
batch.tokens, axis=0, output_is_sparse=True)
flat_tokens = tf.cast(flat_tokens, dtype=TOKEN_COUNT_DTYPE)
return tf.sparse.add(token_counts, flat_tokens)
# Simple tests
# Create the initial zero token counts using empty tensors.
initial_token_counts = tf.SparseTensor(
indices=tf.zeros(shape=(0, 1), dtype=TOKEN_DTYPE),
values=tf.zeros(shape=(0,), dtype=TOKEN_COUNT_DTYPE),
dense_shape=(WORD_VOCAB_SIZE,))
client_token_counts = batched_dataset1.reduce(initial_token_counts,
token_count_fn)
tokens = tf.reshape(client_token_counts.indices, (-1,)).numpy()
print('tokens:', tokens)
np.testing.assert_array_equal(tokens, [0, 1, 4, 8])
# The count is the number of *examples* in which the token/word
# occurs, not the total number of occurences, since we still featurize
# multiple occurences in the same example as a "1".
counts = client_token_counts.values.numpy()
print('counts:', counts)
np.testing.assert_array_equal(counts, [2, 3, 1, 1])
tokens: [0 1 4 8] counts: [2 3 1 1]
আমরা সংশ্লিষ্ট মডেল পরামিতি নির্বাচন করব MAX_TOKENS_SELECTED_PER_CLIENT সবচেয়ে ঘন ঘন ডিভাইসে টোকেন occuring। ডিভাইসে এই অনেক টোকেন কম ঘটতে থাকে, তাহলে আমরা প্যাড তালিকা এর ব্যবহার নিশ্চিত করার federated_select ।
মনে রাখবেন যে অন্যান্য কৌশলগুলি সম্ভবত আরও ভাল, উদাহরণস্বরূপ, এলোমেলোভাবে টোকেন নির্বাচন করা (সম্ভবত তাদের সংঘটনের সম্ভাবনার উপর ভিত্তি করে)। এটি নিশ্চিত করবে যে মডেলের সমস্ত স্লাইস (যার জন্য ক্লায়েন্টের ডেটা আছে) আপডেট হওয়ার কিছু সুযোগ রয়েছে।
@tf.function
def keys_for_client(client_dataset, max_tokens_per_client):
"""Computes a set of max_tokens_per_client keys."""
initial_token_counts = tf.SparseTensor(
indices=tf.zeros((0, 1), dtype=TOKEN_DTYPE),
values=tf.zeros((0,), dtype=TOKEN_COUNT_DTYPE),
dense_shape=(WORD_VOCAB_SIZE,))
client_token_counts = client_dataset.reduce(initial_token_counts,
token_count_fn)
# Find the most-frequently occuring tokens
tokens = tf.reshape(client_token_counts.indices, shape=(-1,))
counts = client_token_counts.values
perm = tf.argsort(counts, direction='DESCENDING')
tokens = tf.gather(tokens, perm)
counts = tf.gather(counts, perm)
num_raw_tokens = tf.shape(tokens)[0]
actual_num_tokens = tf.minimum(max_tokens_per_client, num_raw_tokens)
selected_tokens = tokens[:actual_num_tokens]
paddings = [[0, max_tokens_per_client - tf.shape(selected_tokens)[0]]]
padded_tokens = tf.pad(selected_tokens, paddings=paddings)
# Make sure the type is statically determined
padded_tokens = tf.reshape(padded_tokens, shape=(max_tokens_per_client,))
# We will pass these tokens as keys into `federated_select`, which
# requires SELECT_KEY_DTYPE=tf.int32 keys.
padded_tokens = tf.cast(padded_tokens, dtype=SELECT_KEY_DTYPE)
return padded_tokens, actual_num_tokens
# Simple test
# Case 1: actual_num_tokens > max_tokens_per_client
selected_tokens, actual_num_tokens = keys_for_client(batched_dataset1, 3)
assert tf.size(selected_tokens) == 3
assert actual_num_tokens == 3
# Case 2: actual_num_tokens < max_tokens_per_client
selected_tokens, actual_num_tokens = keys_for_client(batched_dataset1, 10)
assert tf.size(selected_tokens) == 10
assert actual_num_tokens == 4
স্থানীয় টোকেনগুলিতে বিশ্বব্যাপী টোকেনগুলি ম্যাপ করুন
উপরে নির্বাচন আমাদের সীমার মধ্যে টোকেন একটি ঘন সেট দেয় [0, actual_num_tokens) যা আমরা অন ডিভাইস মডেল জন্য ব্যবহার করবে। যাইহোক, ডেটা সেটটি আমরা পড়েছি অনেক বড় বিশ্বব্যাপী শব্দভান্ডার ব্যাপ্তি থেকে টোকেন হয়েছে [0, WORD_VOCAB_SIZE) ।
এইভাবে, আমাদের গ্লোবাল টোকেনগুলিকে তাদের সংশ্লিষ্ট স্থানীয় টোকেনগুলিতে ম্যাপ করতে হবে। স্থানীয় টোকেন আইডি কেবল মধ্যে ইনডেক্সে দেওয়া হয় selected_tokens টেন্সর পূর্ববর্তী ধাপে নির্ণিত।
@tf.function
def map_to_local_token_ids(client_data, client_keys):
global_to_local = tf.lookup.StaticHashTable(
# Note int32 -> int64 maps are not supported
tf.lookup.KeyValueTensorInitializer(
keys=tf.cast(client_keys, dtype=TOKEN_DTYPE),
# Note we need to use tf.shape, not the static
# shape client_keys.shape[0]
values=tf.range(0, limit=tf.shape(client_keys)[0],
dtype=TOKEN_DTYPE)),
# We use -1 for tokens that were not selected, which can occur for clients
# with more than MAX_TOKENS_SELECTED_PER_CLIENT distinct tokens.
# We will simply remove these invalid indices from the batch below.
default_value=-1)
def to_local_ids(sparse_tokens):
indices_t = tf.transpose(sparse_tokens.indices)
batch_indices = indices_t[0] # First column
tokens = indices_t[1] # Second column
tokens = tf.map_fn(
lambda global_token_id: global_to_local.lookup(global_token_id), tokens)
# Remove tokens that aren't actually available (looked up as -1):
available_tokens = tokens >= 0
tokens = tokens[available_tokens]
batch_indices = batch_indices[available_tokens]
updated_indices = tf.transpose(
tf.concat([[batch_indices], [tokens]], axis=0))
st = tf.sparse.SparseTensor(
updated_indices,
tf.ones(tf.size(tokens), dtype=FEATURE_DTYPE),
dense_shape=sparse_tokens.dense_shape)
st = tf.sparse.reorder(st)
return st
return client_data.map(lambda b: BatchType(to_local_ids(b.tokens), b.tags))
# Simple test
client_keys, actual_num_tokens = keys_for_client(
batched_dataset3, MAX_TOKENS_SELECTED_PER_CLIENT)
client_keys = client_keys[:actual_num_tokens]
d = map_to_local_token_ids(batched_dataset3, client_keys)
batch = next(iter(d))
all_tokens = tf.gather(batch.tokens.indices, indices=1, axis=1)
# Confirm we have local indices in the range [0, MAX):
assert tf.math.reduce_max(all_tokens) < MAX_TOKENS_SELECTED_PER_CLIENT
assert tf.math.reduce_max(all_tokens) >= 0
প্রতিটি ক্লায়েন্টকে স্থানীয় (সাব) মডেলকে প্রশিক্ষণ দিন
নোট federated_select একটি হিসেবে নির্বাচিত টুকরা ফিরে আসবে tf.data.Dataset নির্বাচন কীসমূহ হিসাবে একই যাতে। সুতরাং, আমরা প্রথমে এই ধরনের ডেটাসেট নেওয়ার জন্য একটি ইউটিলিটি ফাংশন সংজ্ঞায়িত করি এবং এটিকে একটি একক ঘন টেনসরে রূপান্তর করি যা ক্লায়েন্ট মডেলের মডেল ওজন হিসাবে ব্যবহার করা যেতে পারে।
@tf.function
def slices_dataset_to_tensor(slices_dataset):
"""Convert a dataset of slices to a tensor."""
# Use batching to gather all of the slices into a single tensor.
d = slices_dataset.batch(MAX_TOKENS_SELECTED_PER_CLIENT,
drop_remainder=False)
iter_d = iter(d)
tensor = next(iter_d)
# Make sure we have consumed everything
opt = iter_d.get_next_as_optional()
tf.Assert(tf.logical_not(opt.has_value()), data=[''], name='CHECK_EMPTY')
return tensor
# Simple test
weights = np.random.random(
size=(MAX_TOKENS_SELECTED_PER_CLIENT, TAG_VOCAB_SIZE)).astype(np.float32)
model_slices_as_dataset = tf.data.Dataset.from_tensor_slices(weights)
weights2 = slices_dataset_to_tensor(model_slices_as_dataset)
np.testing.assert_array_equal(weights, weights2)
এখন আমাদের কাছে একটি সাধারণ স্থানীয় প্রশিক্ষণ লুপ সংজ্ঞায়িত করার জন্য প্রয়োজনীয় সমস্ত উপাদান রয়েছে যা প্রতিটি ক্লায়েন্টে চলবে।
@tf.function
def client_train_fn(model, client_optimizer,
model_slices_as_dataset, client_data,
client_keys, actual_num_tokens):
initial_model_weights = slices_dataset_to_tensor(model_slices_as_dataset)
assert len(model.trainable_variables) == 1
model.trainable_variables[0].assign(initial_model_weights)
# Only keep the "real" (unpadded) keys.
client_keys = client_keys[:actual_num_tokens]
client_data = map_to_local_token_ids(client_data, client_keys)
loss_fn = tf.keras.losses.BinaryCrossentropy()
for features, labels in client_data:
with tf.GradientTape() as tape:
predictions = model(features)
loss = loss_fn(labels, predictions)
grads = tape.gradient(loss, model.trainable_variables)
client_optimizer.apply_gradients(zip(grads, model.trainable_variables))
model_weights_delta = model.trainable_weights[0] - initial_model_weights
model_weights_delta = tf.slice(model_weights_delta, begin=[0, 0],
size=[actual_num_tokens, -1])
return client_keys, model_weights_delta
# Simple test
# Note if you execute this cell a second time, you need to also re-execute
# the preceeding cell to avoid "tf.function-decorated function tried to
# create variables on non-first call" errors.
on_device_model = create_logistic_model(MAX_TOKENS_SELECTED_PER_CLIENT,
TAG_VOCAB_SIZE)
client_optimizer = tf.keras.optimizers.SGD(learning_rate=0.001)
client_keys, actual_num_tokens = keys_for_client(
batched_dataset2, MAX_TOKENS_SELECTED_PER_CLIENT)
model_slices_as_dataset = tf.data.Dataset.from_tensor_slices(
np.zeros((MAX_TOKENS_SELECTED_PER_CLIENT, TAG_VOCAB_SIZE),
dtype=np.float32))
keys, delta = client_train_fn(
on_device_model,
client_optimizer,
model_slices_as_dataset,
client_data=batched_dataset3,
client_keys=client_keys,
actual_num_tokens=actual_num_tokens)
print(delta)
সমষ্টি ইনডেক্সড স্লাইস
আমরা ব্যবহার tff.federated_aggregate জন্য একটি ফেডারেট বিক্ষিপ্ত সমষ্টি গঠন করা IndexedSlices । এই সহজ বাস্তবায়নের বাধ্যতা যে হয়েছে dense_shape আগাম স্ট্যাটিক্যালি পরিচিত হয়। লক্ষণীয় যে, এই সমষ্টি কেবল অর্ধ- বিক্ষিপ্ত, অর্থে যে ক্লায়েন্ট -> সার্ভার যোগাযোগ বিক্ষিপ্ত, কিন্তু সার্ভার সমষ্টি একটি ঘন উপস্থাপনা বজায় রাখে accumulate এবং merge , এবং এই ঘন উপস্থাপনা আউটপুট।
def federated_indexed_slices_sum(slice_indices, slice_values, dense_shape):
"""
Sumes IndexedSlices@CLIENTS to a dense @SERVER Tensor.
Intermediate aggregation is performed by converting to a dense representation,
which may not be suitable for all applications.
Args:
slice_indices: An IndexedSlices.indices tensor @CLIENTS.
slice_values: An IndexedSlices.values tensor @CLIENTS.
dense_shape: A statically known dense shape.
Returns:
A dense tensor placed @SERVER representing the sum of the client's
IndexedSclies.
"""
slices_dtype = slice_values.type_signature.member.dtype
zero = tff.tf_computation(
lambda: tf.zeros(dense_shape, dtype=slices_dtype))()
@tf.function
def accumulate_slices(dense, client_value):
indices, slices = client_value
# There is no built-in way to add `IndexedSlices`, but
# tf.convert_to_tensor is a quick way to convert to a dense representation
# so we can add them.
return dense + tf.convert_to_tensor(
tf.IndexedSlices(slices, indices, dense_shape))
return tff.federated_aggregate(
(slice_indices, slice_values),
zero=zero,
accumulate=tff.tf_computation(accumulate_slices),
merge=tff.tf_computation(lambda d1, d2: tf.add(d1, d2, name='merge')),
report=tff.tf_computation(lambda d: d))
একটি ন্যূনতম আঁকো federated_computation একটি পরীক্ষা হিসেবে
dense_shape = (6, 2)
indices_type = tff.TensorType(tf.int64, (None,))
values_type = tff.TensorType(tf.float32, (None, 2))
client_slice_type = tff.type_at_clients(
(indices_type, values_type))
@tff.federated_computation(client_slice_type)
def test_sum_indexed_slices(indices_values_at_client):
indices, values = indices_values_at_client
return federated_indexed_slices_sum(indices, values, dense_shape)
print(test_sum_indexed_slices.type_signature)
({<int64[?],float32[?,2]>}@CLIENTS -> float32[6,2]@SERVER)
x = tf.IndexedSlices(
values=np.array([[2., 2.1], [0., 0.1], [1., 1.1], [5., 5.1]],
dtype=np.float32),
indices=[2, 0, 1, 5],
dense_shape=dense_shape)
y = tf.IndexedSlices(
values=np.array([[0., 0.3], [3.1, 3.2]], dtype=np.float32),
indices=[1, 3],
dense_shape=dense_shape)
# Sum one.
result = test_sum_indexed_slices([(x.indices, x.values)])
np.testing.assert_array_equal(tf.convert_to_tensor(x), result)
# Sum two.
expected = [[0., 0.1], [1., 1.4], [2., 2.1], [3.1, 3.2], [0., 0.], [5., 5.1]]
result = test_sum_indexed_slices([(x.indices, x.values), (y.indices, y.values)])
np.testing.assert_array_almost_equal(expected, result)
একটি মধ্যে সবাই একসঙ্গে এটি নির্বাণ federated_computation
এখন আমরা একটি মধ্যে উপাদান একসঙ্গে বাঁধাই করার TFF ব্যবহার tff.federated_computation ।
DENSE_MODEL_SHAPE = (WORD_VOCAB_SIZE, TAG_VOCAB_SIZE)
client_data_type = tff.SequenceType(batched_dataset1.element_spec)
model_type = tff.TensorType(tf.float32, shape=DENSE_MODEL_SHAPE)
আমরা ফেডারেটেড এভারেজিংয়ের উপর ভিত্তি করে একটি বেসিক সার্ভার প্রশিক্ষণ ফাংশন ব্যবহার করি, 1.0 সার্ভার শেখার হার সহ আপডেট প্রয়োগ করি। এটি গুরুত্বপূর্ণ যে আমরা মডেলটিতে একটি আপডেট (ডেল্টা) প্রয়োগ করি, কেবলমাত্র ক্লায়েন্ট দ্বারা সরবরাহ করা মডেলগুলির গড় না করে, অন্যথায় যদি মডেলটির একটি প্রদত্ত স্লাইস প্রদত্ত রাউন্ডে কোনও ক্লায়েন্ট দ্বারা প্রশিক্ষিত না হয় তবে এর সহগ শূন্য করা যেতে পারে। আউট
@tff.tf_computation
def server_update(current_model_weights, update_sum, num_clients):
average_update = update_sum / num_clients
return current_model_weights + average_update
আমরা একটি দম্পতি বেশি প্রয়োজন tff.tf_computation উপাদান:
# Function to select slices from the model weights in federated_select:
select_fn = tff.tf_computation(
lambda model_weights, index: tf.gather(model_weights, index))
# We need to wrap `client_train_fn` as a `tff.tf_computation`, making
# sure we do any operations that might construct `tf.Variable`s outside
# of the `tf.function` we are wrapping.
@tff.tf_computation
def client_train_fn_tff(model_slices_as_dataset, client_data, client_keys,
actual_num_tokens):
# Note this is amaller than the global model, using
# MAX_TOKENS_SELECTED_PER_CLIENT which is much smaller than WORD_VOCAB_SIZE.
# W7e would like a model of size `actual_num_tokens`, but we
# can't build the model dynamically, so we will slice off the padded
# weights at the end.
client_model = create_logistic_model(MAX_TOKENS_SELECTED_PER_CLIENT,
TAG_VOCAB_SIZE)
client_optimizer = tf.keras.optimizers.SGD(learning_rate=0.1)
return client_train_fn(client_model, client_optimizer,
model_slices_as_dataset, client_data, client_keys,
actual_num_tokens)
@tff.tf_computation
def keys_for_client_tff(client_data):
return keys_for_client(client_data, MAX_TOKENS_SELECTED_PER_CLIENT)
আমরা এখন একসাথে সব টুকরা করা প্রস্তুত!
@tff.federated_computation(
tff.type_at_server(model_type), tff.type_at_clients(client_data_type))
def sparse_model_update(server_model, client_data):
max_tokens = tff.federated_value(MAX_TOKENS_SELECTED_PER_CLIENT, tff.SERVER)
keys_at_clients, actual_num_tokens = tff.federated_map(
keys_for_client_tff, client_data)
model_slices = tff.federated_select(keys_at_clients, max_tokens, server_model,
select_fn)
update_keys, update_slices = tff.federated_map(
client_train_fn_tff,
(model_slices, client_data, keys_at_clients, actual_num_tokens))
dense_update_sum = federated_indexed_slices_sum(update_keys, update_slices,
DENSE_MODEL_SHAPE)
num_clients = tff.federated_sum(tff.federated_value(1.0, tff.CLIENTS))
updated_server_model = tff.federated_map(
server_update, (server_model, dense_update_sum, num_clients))
return updated_server_model
print(sparse_model_update.type_signature)
(<server_model=float32[13,4]@SERVER,client_data={<tokens=<indices=int64[?,2],values=int32[?],dense_shape=int64[2]>,tags=float32[?,4]>*}@CLIENTS> -> float32[13,4]@SERVER)
এর একটি মডেল প্রশিক্ষণ দেওয়া যাক!
এখন যেহেতু আমাদের প্রশিক্ষণ ফাংশন আছে, আসুন এটি চেষ্টা করে দেখি।
server_model = create_logistic_model(WORD_VOCAB_SIZE, TAG_VOCAB_SIZE)
server_model.compile( # Compile to make evaluation easy.
optimizer=tf.keras.optimizers.Adagrad(learning_rate=0.0), # Unused
loss=tf.keras.losses.BinaryCrossentropy(),
metrics=[
tf.keras.metrics.Precision(name='precision'),
tf.keras.metrics.AUC(name='auc'),
tf.keras.metrics.Recall(top_k=2, name='recall_at_2'),
])
def evaluate(model, dataset, name):
metrics = model.evaluate(dataset, verbose=0)
metrics_str = ', '.join([f'{k}={v:.2f}' for k, v in
(zip(server_model.metrics_names, metrics))])
print(f'{name}: {metrics_str}')
print('Before training')
evaluate(server_model, batched_dataset1, 'Client 1')
evaluate(server_model, batched_dataset2, 'Client 2')
evaluate(server_model, batched_dataset3, 'Client 3')
model_weights = server_model.trainable_weights[0]
client_datasets = [batched_dataset1, batched_dataset2, batched_dataset3]
for _ in range(10): # Run 10 rounds of FedAvg
# We train on 1, 2, or 3 clients per round, selecting
# randomly.
cohort_size = np.random.randint(1, 4)
clients = np.random.choice([0, 1, 2], cohort_size, replace=False)
print('Training on clients', clients)
model_weights = sparse_model_update(
model_weights, [client_datasets[i] for i in clients])
server_model.set_weights([model_weights])
print('After training')
evaluate(server_model, batched_dataset1, 'Client 1')
evaluate(server_model, batched_dataset2, 'Client 2')
evaluate(server_model, batched_dataset3, 'Client 3')
Before training Client 1: loss=0.69, precision=0.00, auc=0.50, recall_at_2=0.60 Client 2: loss=0.69, precision=0.00, auc=0.50, recall_at_2=0.50 Client 3: loss=0.69, precision=0.00, auc=0.50, recall_at_2=0.40 Training on clients [0 1] Training on clients [0 2 1] Training on clients [2 0] Training on clients [1 0 2] Training on clients [2] Training on clients [2 0] Training on clients [1 2 0] Training on clients [0] Training on clients [2] Training on clients [1 2] After training Client 1: loss=0.67, precision=0.80, auc=0.91, recall_at_2=0.80 Client 2: loss=0.68, precision=0.67, auc=0.96, recall_at_2=1.00 Client 3: loss=0.65, precision=1.00, auc=0.93, recall_at_2=0.80