فصل ۱۲: مدل‌های سفارشی و آموزش سفارشی با TensorFlow

تا اینجا تقریباً تمام کارها با API سطح‌بالای TensorFlow یعنی Keras انجام شد و همین API برای بخش بزرگی از کاربردهای واقعی کافی است. با Keras شبکه‌های رگرسیون و طبقه‌بندی، معماری Wide & Deep، شبکه‌های Self-Normalizing و تکنیک‌هایی مانند Batch Normalization، Dropout و Learning-Rate Schedule را پیاده‌سازی کردیم. با این حال، بعضی پروژه‌ها به کنترل سطح پایین‌تری نیاز دارند: Loss یا Metric سفارشی، Layer و Model اختصاصی، Initializer، Regularizer و Weight Constraint سفارشی، تغییر مستقیم Gradientها یا حتی استفاده از چند Optimizer برای بخش‌های متفاوت شبکه. فصل ۱۲ برای همین سطح از کنترل به API سطح‌پایین Python در TensorFlow می‌رود و در ادامه نشان می‌دهد چگونه تولید خودکار Graph می‌تواند کد سفارشی را سریع و قابل‌انتقال کند.

مروری سریع بر TensorFlow

TensorFlow کتابخانه‌ای قدرتمند برای محاسبات عددی است که به‌طور ویژه برای یادگیری ماشین در مقیاس بزرگ بهینه شده است. هستهٔ آن از نظر مدل برنامه‌نویسی به NumPy شبیه است، اما اجرای GPU و TPU، محاسبات توزیع‌شده، Reverse-Mode Automatic Differentiation و Optimizerهای پیشرفته را در اختیار می‌گذارد. TensorFlow در سال ۲۰۱۵ متن‌باز شد و در صنعت برای مسائلی مانند طبقه‌بندی تصویر، پردازش زبان طبیعی، سیستم‌های پیشنهاددهنده و پیش‌بینی سری زمانی به‌طور گسترده استفاده می‌شود.

قابلیت‌های اصلی TensorFlow را می‌توان چنین خلاصه کرد:

• هسته‌ای شبیه NumPy، اما با Kernelهای بهینه برای CPU، GPU و TPU.
• پشتیبانی از اجرای توزیع‌شده روی چند Device و چند Server.
• کامپایل و بهینه‌سازی Graph برای سرعت و مصرف حافظه؛ Nodeهای غیرضروری حذف و Operationهای مستقل تا حد امکان موازی می‌شوند.
• امکان Export کردن Computation Graph در قالبی قابل‌انتقال؛ Model می‌تواند در یک محیط Train و در محیط دیگری اجرا شود.
• Reverse-Mode Autodiff و Optimizerهایی مانند RMSProp و Nadam برای کمینه‌سازی Lossهای پیچیده.
• APIهای تکمیلی مانند tf.data و tf.io برای Data Pipeline، tf.image برای پردازش تصویر، tf.signal برای سیگنال و tf.summary برای TensorBoard.

Operationهای سطح پایین با C++ پیاده‌سازی شده‌اند و برای Deviceهای مختلف Kernelهای اختصاصی دارند. در عمل بیشتر کد با Keras و tf.data نوشته می‌شود و فقط وقتی انعطاف بیشتر لازم است مستقیماً Tensorها و Operationها مدیریت می‌شوند. Execution Engine مسئول اجرای کارآمد Graph است و در صورت پیکربندی می‌تواند آن را میان چند Device یا Machine توزیع کند.

TensorFlow علاوه بر خود کتابخانه، یک اکوسیستم کامل دارد: TensorBoard برای Visualization، مجموعهٔ TensorFlow Extended یا TFX برای Productionization و Data Validation/Preprocessing/Model Analysis/Serving، TensorFlow Hub برای استفادهٔ مجدد از Modelهای Pretrained، Model Garden برای معماری‌های آماده و TensorFlow Lite برای اجرای Model روی Mobile. APIهایی برای C++، Java و Swift وجود دارد و TensorFlow.js نیز امکان اجرای Model در Browser را فراهم می‌کند.

استفاده از TensorFlow مانند NumPy

Tensorها و Operationها

هستهٔ TensorFlow حول Tensor می‌چرخد. Tensor معمولاً یک آرایهٔ چندبعدی شبیه numpy.ndarray است، اما می‌تواند Scalar هم باشد. برای ساخت Tensor ثابت از tf.constant() استفاده می‌شود:

import tensorflow as tf

t = tf.constant([[1., 2., 3.], [4., 5., 6.]])
print(t.shape)   # TensorShape([2, 3])
print(t.dtype)   # tf.float32

Indexing تقریباً مانند NumPy است:

t[:, 1:]
t[..., 1, tf.newaxis]

Operationهای ریاضی نیز مستقیم روی Tensorها کار می‌کنند:

t + 10
tf.square(t)
t @ tf.transpose(t)

عبارت t + 10 در نهایت به tf.add(t, 10) تبدیل می‌شود و Operator @ معادل tf.matmul() است. بسیاری از Functionها Alias دارند؛ مثلاً tf.add و tf.math.add یک Operation را ارجاع می‌دهند. یک Scalar Tensor Shape خالی دارد:

tf.constant(42)   # shape=(), dtype=int32

TensorFlow تقریباً همهٔ Operationهای پایهٔ لازم را دارد: tf.add، tf.multiply، tf.square، tf.exp، tf.sqrt، tf.reshape، tf.squeeze، tf.tile و موارد دیگر. بعضی نام‌ها با NumPy تفاوت دارند؛ برای نمونه tf.reduce_mean، tf.reduce_sum، tf.reduce_max و tf.math.log تقریباً همتای np.mean، np.sum، np.max و np.log هستند.

TensorFlow و NumPy

تبدیل میان Tensor و NumPy ساده است. می‌توان از NumPy Array، Tensor ساخت و بالعکس؛ حتی بسیاری از Operationهای TensorFlow روی Arrayهای NumPy و Operationهای NumPy روی Tensorها کار می‌کنند:

import numpy as np

a = np.array([2., 4., 5.])
tf.constant(a)
t.numpy()          # یا np.array(t)
tf.square(a)
np.square(t)

یک تفاوت عملی مهم Precision پیش‌فرض است: NumPy معمولاً Float64 و TensorFlow معمولاً Float32 را انتخاب می‌کند. در شبکه‌های عصبی Float32 اغلب کافی، سریع‌تر و کم‌حافظه‌تر است؛ بنابراین هنگام تبدیل Arrayهای NumPy بهتر است Type را آگاهانه تعیین کنید.

تبدیل نوع داده

TensorFlow برای جلوگیری از Conversionهای پنهان و پرهزینه، Type Conversion خودکار انجام نمی‌دهد. جمع کردن Tensor Float با Integer یا حتی Float32 با Float64 خطا ایجاد می‌کند. برای تبدیل صریح از tf.cast() استفاده کنید:

t2 = tf.constant(40., dtype=tf.float64)
result = tf.constant(2.0) + tf.cast(t2, tf.float32)

Variableها

tf.Tensor تغییرناپذیر است، اما Weightهای شبکه و Stateهای Optimizer باید در طول Training تغییر کنند. برای این منظور از tf.Variable استفاده می‌شود. Variable تقریباً همهٔ Operationهای Tensor را پشتیبانی می‌کند، با این تفاوت که می‌توان مقدار آن را درجا تغییر داد:

v = tf.Variable([[1., 2., 3.], [4., 5., 6.]])
v.assign(2 * v)
v[0, 1].assign(42)
v[:, 2].assign([0., 1.])
v.scatter_nd_update(indices=[[0, 0], [1, 2]], updates=[100., 200.])

Assignment معمول Python مانند v[1] = ... برای ResourceVariable معتبر نیست. در Keras معمولاً نیازی نیست Variable را دستی بسازید؛ متد add_weight() این کار را انجام می‌دهد و Optimizerها Weightها را به‌روزرسانی می‌کنند.

ساختارهای دادهٔ دیگر TensorFlow

علاوه بر Tensor معمولی، TensorFlow ساختارهای دیگری برای داده‌های خاص ارائه می‌کند:

• tf.SparseTensor: نمایش کارآمد Tensorهایی که بیشتر مقادیر آن‌ها صفر است.
• tf.TensorArray: فهرستی از Tensorهای هم‌نوع و هم‌شکل؛ طول آن به‌صورت پیش‌فرض ثابت است اما می‌تواند توسعه‌پذیر باشد.
• tf.RaggedTensor: نمایش Sequenceهایی با Rank و Type یکسان اما طول‌های متفاوت.
• String Tensor: Tensor از نوع tf.string برای Byte String. Stringهای Unicode در Python هنگام تبدیل به UTF-8 Encode می‌شوند؛ tf.strings ابزارهای لازم برای کار با Byte/Unicode را ارائه می‌کند.
• Setها: با Tensor معمولی یا SparseTensor نمایش داده می‌شوند و Operationهای مرتبط در tf.sets قرار دارند.
• Queueها: انواع FIFO، Priority، RandomShuffle و PaddingFIFO برای نگهداری Tensor در چند Step.

با Tensorها، Operationها، Variableها و این ساختارها، زیرساخت لازم برای ساخت Componentها و Training Algorithmهای سفارشی فراهم شده است.

سفارشی‌سازی Model و الگوریتم Training

Loss Function سفارشی

فرض کنید یک Regression Model روی Dataset نویزی Train می‌شود و حذف Outlierها هم مسئله را کاملاً حل نکرده است. MSE خطاهای بزرگ را بیش از حد جریمه می‌کند و MAE در برابر Outlier مقاوم‌تر است اما ممکن است کندتر همگرا شود. Huber Loss ترکیبی از این دو رفتار است: برای Errorهای کوچک Quadratic و برای Errorهای بزرگ Linear. در Keras این Loss آماده است، اما برای نشان دادن روش ساخت Loss سفارشی آن را دستی می‌نویسیم:

def huber_fn(y_true, y_pred):
    error = y_true - y_pred
    is_small_error = tf.abs(error) < 1
    squared_loss = tf.square(error) / 2
    linear_loss = tf.abs(error) - 0.5
    return tf.where(is_small_error, squared_loss, linear_loss)

پیاده‌سازی Vectorized به‌مراتب سریع‌تر از Loop روی Sampleهاست. بهتر است Function فقط از Operationهای TensorFlow استفاده کند تا Graph Optimization ممکن باشد. همچنین بهتر است Loss یک مقدار برای هر Sample برگرداند و Mean را خود Keras محاسبه کند؛ در غیر این صورت استفادهٔ صحیح از Class Weight یا Sample Weight سخت می‌شود.

model.compile(loss=huber_fn, optimizer="nadam")
model.fit(X_train, y_train, ...)

در هر Batch، Keras این Function را فراخوانی می‌کند، Loss را می‌سازد، با Reverse-Mode Autodiff Gradientها را نسبت به Parameterها محاسبه می‌کند و Optimizer Step را اعمال می‌کند.

Save و Load کردن Modelهای دارای Component سفارشی

ذخیرهٔ Model دارای Loss سفارشی مشکلی ندارد، اما هنگام Load باید نام Component به Object واقعی Map شود:

model = tf.keras.models.load_model(
    "my_model_with_a_custom_loss",
    custom_objects={"huber_fn": huber_fn})

اگر Function با @tf.keras.utils.register_keras_serializable() ثبت شود، Keras آن را بدون نیاز به custom_objects پیدا می‌کند.

اگر Loss یک Hyperparameter داشته باشد، مثلاً Threshold، می‌توان Function Factory ساخت:

def create_huber(threshold=1.0):
    def huber_fn(y_true, y_pred):
        error = y_true - y_pred
        is_small_error = tf.abs(error) < threshold
        squared_loss = tf.square(error) / 2
        linear_loss = threshold * tf.abs(error) - threshold ** 2 / 2
        return tf.where(is_small_error, squared_loss, linear_loss)
    return huber_fn

اما Closure به‌تنهایی مقدار Threshold را همراه Model Serialize نمی‌کند. برای Persist شدن Hyperparameter بهتر است از Class ارث‌بری کنیم:

class HuberLoss(tf.keras.losses.Loss):
    def __init__(self, threshold=1.0, **kwargs):
        self.threshold = threshold
        super().__init__(**kwargs)

    def call(self, y_true, y_pred):
        error = y_true - y_pred
        is_small_error = tf.abs(error) < self.threshold
        squared_loss = tf.square(error) / 2
        linear_loss = self.threshold * tf.abs(error) - self.threshold**2 / 2
        return tf.where(is_small_error, squared_loss, linear_loss)

    def get_config(self):
        base_config = super().get_config()
        return {**base_config, "threshold": self.threshold}

get_config() Dictionary تنظیمات را برمی‌گرداند. Keras هنگام Save این Config را ذخیره و هنگام Load با from_config() Instance را بازسازی می‌کند.

Activation، Initializer، Regularizer و Constraint سفارشی

اغلب Componentهای Keras را می‌توان با Function ساده سفارشی کرد. مثال‌های زیر معادل نمونه‌های استاندارد Keras هستند:

def my_softplus(z):
    return tf.math.log(1.0 + tf.exp(z))

def my_glorot_initializer(shape, dtype=tf.float32):
    stddev = tf.sqrt(2. / (shape[0] + shape[1]))
    return tf.random.normal(shape, stddev=stddev, dtype=dtype)

def my_l1_regularizer(weights):
    return tf.reduce_sum(tf.abs(0.01 * weights))

def my_positive_weights(weights):
    return tf.where(weights < 0., tf.zeros_like(weights), weights)

layer = tf.keras.layers.Dense(
    1,
    activation=my_softplus,
    kernel_initializer=my_glorot_initializer,
    kernel_regularizer=my_l1_regularizer,
    kernel_constraint=my_positive_weights)

Activation روی Output Layer اعمال می‌شود؛ Initializer مقدار اولیهٔ Weight را می‌سازد؛ Regularizer در هر Training Step Loss تنظیمی را به Loss اصلی اضافه می‌کند؛ Constraint پس از Step روی Weight اعمال و مقدار Weight را با نسخهٔ محدودشده جایگزین می‌کند.

اگر Component Hyperparameterی دارد که باید همراه Model ذخیره شود، از Class مناسب ارث‌بری کنید؛ مثلاً Regularizer، Constraint، Initializer یا Layer. برای Loss، Layer و Model معمولاً call() و برای Regularizer/Initializer/Constraint معمولاً __call__() پیاده‌سازی می‌شود.

Metricهای سفارشی و Streaming Metric

Loss و Metric یک هدف ندارند. Loss برای Gradient Descent استفاده می‌شود و باید در نقاط لازم مشتق‌پذیر و دارای Gradient مفید باشد؛ Metric برای ارزیابی Model است و ممکن است نامشتق‌پذیر باشد. بسیاری از Metricهای ساده را می‌توان دقیقاً مانند Loss به شکل Function تعریف کرد.

اما بعضی Metricها را نمی‌توان Mean سادهٔ مقدار هر Batch در نظر گرفت. برای مثال Precision برابر است با True Positive تقسیم بر کل Positive Prediction. اگر Batch اول Precision برابر ۸۰٪ و Batch دوم ۰٪ باشد، Average سادهٔ ۴۰٪ الزاماً Precision واقعی مجموع دو Batch نیست. اگر Batch اول ۴ TP از ۵ Positive و Batch دوم ۰ TP از ۳ Positive داشته باشد، Precision کل 4/8 = 50% است. بنابراین باید State نگهداری شود.

precision = tf.keras.metrics.Precision()
precision(y_true_batch1, y_pred_batch1)
precision(y_true_batch2, y_pred_batch2)
precision.result()
precision.variables
precision.reset_states()

این نوع Metric را Streaming یا Stateful Metric می‌نامند. برای Metric سفارشی Stateful از tf.keras.metrics.Metric ارث‌بری می‌کنیم:

class HuberMetric(tf.keras.metrics.Metric):
    def __init__(self, threshold=1.0, **kwargs):
        super().__init__(**kwargs)
        self.threshold = threshold
        self.huber_fn = create_huber(threshold)
        self.total = self.add_weight("total", initializer="zeros")
        self.count = self.add_weight("count", initializer="zeros")

    def update_state(self, y_true, y_pred, sample_weight=None):
        sample_metrics = self.huber_fn(y_true, y_pred)
        self.total.assign_add(tf.reduce_sum(sample_metrics))
        self.count.assign_add(tf.cast(tf.size(y_true), tf.float32))

    def result(self):
        return self.total / self.count

    def get_config(self):
        base_config = super().get_config()
        return {**base_config, "threshold": self.threshold}

add_weight() State Metric را می‌سازد، update_state() آن را برای هر Batch به‌روزرسانی می‌کند و result() مقدار نهایی را محاسبه می‌کند. Keras Persist شدن Variableها را مدیریت می‌کند.

Layerهای سفارشی

برای Layer بدون Weight ساده‌ترین راه استفاده از Lambda است:

exponential_layer = tf.keras.layers.Lambda(lambda x: tf.exp(x))

برای Layer دارای State و Weight باید از tf.keras.layers.Layer ارث‌بری کرد. مثال زیر نسخه‌ای ساده‌شده از Dense است:

class MyDense(tf.keras.layers.Layer):
    def __init__(self, units, activation=None, **kwargs):
        super().__init__(**kwargs)
        self.units = units
        self.activation = tf.keras.activations.get(activation)

    def build(self, batch_input_shape):
        self.kernel = self.add_weight(
            name="kernel",
            shape=[batch_input_shape[-1], self.units],
            initializer="glorot_normal")
        self.bias = self.add_weight(
            name="bias", shape=[self.units], initializer="zeros")

    def call(self, X):
        return self.activation(X @ self.kernel + self.bias)

    def get_config(self):
        base_config = super().get_config()
        return {**base_config,
                "units": self.units,
                "activation": tf.keras.activations.serialize(self.activation)}

build() نخستین بار که Shape ورودی معلوم شود اجرا می‌شود و Variableهای Layer را با add_weight() می‌سازد. call() محاسبهٔ اصلی را تعریف می‌کند و get_config() Hyperparameterهای لازم برای Serialization را ثبت می‌کند. Keras در حالت معمول Output Shape را خودکار استنتاج می‌کند.

برای Layer با چند Input، آرگومان call() می‌تواند Tuple باشد و برای چند Output می‌توان Tuple/List برگرداند:

class MyMultiLayer(tf.keras.layers.Layer):
    def call(self, X):
        X1, X2 = X
        return X1 + X2, X1 * X2, X1 / X2

چنین Layerی با Functional API یا Subclassing API قابل استفاده است، اما Sequential API به Layerهای یک‌ورودی/یک‌خروجی محدود است.