راهنمای نوشتن تماس های سفارشی TensorFlow/Keras

معرفی

فرض کنید می خواهید مدل Keras شما در حین آموزش، ارزیابی یا پیش بینی رفتار خاصی داشته باشد. به عنوان مثال، ممکن است بخواهید مدل خود را در هر دوره آموزشی ذخیره کنید. یکی از راه های انجام این کار استفاده از Callbacks است.

به طور کلی، Callback ها توابعی هستند که هنگام وقوع رویدادی فراخوانی می شوند و به عنوان آرگومان به توابع دیگر ارسال می شوند. در مورد Keras، آنها ابزاری برای سفارشی کردن رفتار مدل شما هستند – خواه در حین آموزش، ارزیابی یا استنتاج. برخی از برنامه ها عبارتند از ورود به سیستم، تداوم مدل، توقف زودهنگام یا تغییر نرخ یادگیری. این کار با ارسال لیستی از Callbacks به عنوان آرگومان انجام می شود keras.Model.fit()،keras.Model.evaluate() یا keras.Model.predict().

برخی از موارد استفاده متداول برای پاسخگویی به تماس عبارتند از: اصلاح نرخ یادگیری، ثبت گزارش، نظارت و توقف زودهنگام آموزش. Keras دارای تعدادی فراخوان داخلی است که به تفصیل می باشد
در مستندات.

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

کلاس Callback و روش های آن

Keras یک کلاس callback خاص دارد، keras.callbacks.Callback، با روش هایی که می توان در حین آموزش، تست و استنباط نام برد روی سطح جهانی، دسته ای یا دوره ای. به منظور. واسه اینکه. برای اینکه ایجاد تماس های سفارشی، باید یک زیر کلاس ایجاد کنیم و این متدها را لغو کنیم.

این keras.callbacks.Callback کلاس سه نوع متد دارد:

• روش های جهانی: در ابتدا یا در پایان نامیده می شود fit()، evaluate() و predict().
• روش های سطح دسته ای: در ابتدا یا در پایان پردازش یک دسته نامیده می شود.
• روش‌های سطح دوره: در ابتدا یا در پایان یک دسته آموزشی نامیده می‌شوند.

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

بیایید با تعریف یک مدل اسباب بازی شروع کنیم:

import tensorflow as tf
from tensorflow import keras
import numpy as np

model = keras.Sequential()
model.add(keras.layers.Dense(10, input_dim = 1, activation='relu'))
model.add(keras.layers.Dense(10, activation='relu'))
model.add(keras.layers.Dense(1))
model.compile(
    optimizer=keras.optimizers.RMSprop(learning_rate=0.1),
    loss = "mean_squared_error",
    metrics = ("mean_absolute_error")
)

x = np.random.uniform(low = 0, high = 10, size = 1000)
y = x**2
x_train, x_test = (x(:900),x(900:))
y_train, y_test = (y(:900),y(900:))

پاسخ به تماس آموزشی سفارشی

اولین تماس ما این است که در حین آموزش فراخوانی شود. بیایید زیر کلاس Callback کلاس:

class TrainingCallback(keras.callbacks.Callback):
    def __init__(self):
        self.tabulation = {"train":"", 'batch': " "*8, 'epoch':" "*4}
    def on_train_begin(self, logs=None):
        tab = self.tabulation('train')
        print(f"{tab}Training!")
        print(f"{tab}available logs: {logs}")

    def on_train_batch_begin(self, batch, logs=None):
        tab = self.tabulation('batch')
        print(f"{tab}Batch {batch}")
        print(f"{tab}available logs: {logs}")

    def on_train_batch_end(self, batch, logs=None):
        tab = self.tabulation('batch')
        print(f"{tab}End of Batch {batch}")
        print(f"{tab}available logs: {logs}")

    def on_epoch_begin(self, epoch, logs=None):
        tab = self.tabulation('epoch')
        print(f"{tab}Epoch {epoch} of training")
        print(f"{tab}available logs: {logs}")

    def on_epoch_end(self, epoch, logs=None):
        tab = self.tabulation('epoch')
        print(f"{tab}End of Epoch {epoch} of training")
        print(f"{tab}available logs: {logs}")

    def on_train_end(self, logs=None):
        tab = self.tabulation('train')
        print(f"{tab}Finishing training!")
        print(f"{tab}available logs: {logs}")

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

بیایید نگاهی به خروجی ها بیندازیم:

model.fit(
    x_train,
    y_train,
    batch_size=500,
    epochs=2,
    verbose=0,
    callbacks=(TrainingCallback()),
)

Training!
available logs: {}
    Epoch 0 of training
    available logs: {}
        Batch 0
        available logs: {}
        End of Batch 0
        available logs: {'loss': 2172.373291015625, 'mean_absolute_error': 34.79669952392578}
        Batch 1
        available logs: {}
        End of Batch 1
        available logs: {'loss': 2030.1309814453125, 'mean_absolute_error': 33.30256271362305}
    End of Epoch 0 of training
    available logs: {'loss': 2030.1309814453125, 'mean_absolute_error': 33.30256271362305}
    Epoch 1 of training
    available logs: {}
        Batch 0
        available logs: {}
        End of Batch 0
        available logs: {'loss': 1746.2772216796875, 'mean_absolute_error': 30.268001556396484}
        Batch 1
        available logs: {}
        End of Batch 1
        available logs: {'loss': 1467.36376953125, 'mean_absolute_error': 27.10252571105957}
    End of Epoch 1 of training
    available logs: {'loss': 1467.36376953125, 'mean_absolute_error': 27.10252571105957}
Finishing training!
available logs: {'loss': 1467.36376953125, 'mean_absolute_error': 27.10252571105957}

<keras.callbacks.History at 0x7f8bce314c10>

توجه داشته باشید که ما می توانیم در هر مرحله دنبال کنیم که مدل در حال انجام چه کاری است و به کدام معیارها دسترسی داریم. در پایان هر دسته و دوره، ما به تابع ضرر درون نمونه و معیارهای مدل خود دسترسی داریم.

پاسخ به تماس ارزیابی سفارشی

حالا بیایید با آن تماس بگیریم Model.evaluate() روش. می بینیم که در پایان یک دسته به تابع ضرر و معیارهای آن زمان دسترسی داریم و در پایان ارزیابی به تلفات و معیارهای کلی دسترسی داریم:

class TestingCallback(keras.callbacks.Callback):
    def __init__(self):
          self.tabulation = {"test":"", 'batch': " "*8}
      
    def on_test_begin(self, logs=None):
        tab = self.tabulation('test')
        print(f'{tab}Evaluating!')
        print(f'{tab}available logs: {logs}')

    def on_test_end(self, logs=None):
        tab = self.tabulation('test')
        print(f'{tab}Finishing evaluation!')
        print(f'{tab}available logs: {logs}')

    def on_test_batch_begin(self, batch, logs=None):
        tab = self.tabulation('batch')
        print(f"{tab}Batch {batch}")
        print(f"{tab}available logs: {logs}")

    def on_test_batch_end(self, batch, logs=None):
        tab = self.tabulation('batch')
        print(f"{tab}End of batch {batch}")
        print(f"{tab}available logs: {logs}")

res = model.evaluate(
    x_test, y_test, batch_size=100, verbose=0, callbacks=(TestingCallback())
)

Evaluating!
available logs: {}
        Batch 0
        available logs: {}
        End of batch 0
        available logs: {'loss': 382.2723083496094, 'mean_absolute_error': 14.069927215576172}
Finishing evaluation!
available logs: {'loss': 382.2723083496094, 'mean_absolute_error': 14.069927215576172}

پیش‌بینی سفارشی برگشت به تماس

در نهایت، اجازه دهید تماس بگیرید Model.predict() روش. توجه داشته باشید که در پایان هر دسته ما به خروجی های پیش بینی شده مدل خود دسترسی داریم:

class PredictionCallback(keras.callbacks.Callback):
    def __init__(self):
        self.tabulation = {"prediction":"", 'batch': " "*8}

    def on_predict_begin(self, logs=None):
        tab = self.tabulation('prediction')
        print(f"{tab}Predicting!")
        print(f"{tab}available logs: {logs}")

    def on_predict_end(self, logs=None):
        tab = self.tabulation('prediction')
        print(f"{tab}End of Prediction!")
        print(f"{tab}available logs: {logs}")

    def on_predict_batch_begin(self, batch, logs=None):
        tab = self.tabulation('batch')
        print(f"{tab}batch {batch}")
        print(f"{tab}available logs: {logs}")

    def on_predict_batch_end(self, batch, logs=None):
        tab = self.tabulation('batch')
        print(f"{tab}End of batch {batch}")
        print(f"{tab}available logs:\n {logs}")

res = model.predict(x_test(:10),
                    verbose = 0, 
                    callbacks=(PredictionCallback()))

Predicting!
available logs: {}
        batch 0
        available logs: {}
        End of batch 0
        available logs:
 {'outputs': array((( 7.743822),
       (27.748264),
       (33.082104),
       (26.530678),
       (27.939169),
       (18.414223),
       (42.610645),
       (36.69335 ),
       (13.096557),
       (37.120853)), dtype=float32)}
End of Prediction!
available logs: {}

با اینها – می توانید رفتار را سفارشی کنید، نظارت را تنظیم کنید یا فرآیندهای آموزش، ارزیابی یا استنتاج را تغییر دهید. یک جایگزین برای طبقه بندی فرعی استفاده از LambdaCallback.

با استفاده از LambaCallback

یکی از کال بک های داخلی در Keras است LambdaCallback کلاس این فراخوانی تابعی را می پذیرد که روش رفتار و عملکرد آن را مشخص می کند! به یک معنا، به شما این امکان را می دهد که از هر عملکرد دلخواه به عنوان یک تماس استفاده کنید، بنابراین به شما امکان می دهد تماس های سفارشی ایجاد کنید.

کلاس دارای پارامترهای اختیاری است:
–on_epoch_begin

• on_epoch_end
• on_batch_begin
• on_batch_end
• on_train_begin
• on_train_end

هر پارامتر می پذیرد یک تابع که در رویداد مدل مربوطه فراخوانی می شود. به عنوان مثال، اجازه دهید پس از پایان آموزش، یک تماس برای ارسال ایمیل برقرار کنیم:

import smtplib
from email.message import EmailMessage

def send_email(logs): 
    msg = EmailMessage()
    content = f"""The model has finished training."""
    for key, value in logs.items():
      content = content + f"\n{key}:{value:.2f}"
    msg.set_content(content)
    msg('Subject') = f'Training report'
    msg('From') = '(email protected)'
    msg('To') = 'receiver-email'

    s = smtplib.SMTP('smtp.gmail.com', 587)
    s.starttls()
    s.login("(email protected)", "your-gmail-app-password")
    s.send_message(msg)
    s.quit()

lambda_send_email = lambda logs : send_email(logs)

email_callback = keras.callbacks.LambdaCallback(on_train_end = lambda_send_email)

model.fit(
    x_train,
    y_train,
    batch_size=100,
    epochs=1,
    verbose=0,
    callbacks=(email_callback),
)

برای برقراری تماس سفارشی ما با استفاده از LambdaCallback، فقط باید تابعی را که می خواهیم فراخوانی شود پیاده سازی کنیم، آن را به صورت a بپیچانیم lambda تابع و ارسال آن به
LambdaCallback کلاس به عنوان یک پارامتر

پاسخ به تماس برای آموزش تصویرسازی مدل

در این بخش، مثالی از یک callback سفارشی ارائه می دهیم که باعث می شود انیمیشنی از عملکرد مدل ما در طول آموزش بهبود یابد. برای انجام این کار، مقادیر لاگ ها را در انتهای هر دسته ذخیره می کنیم. سپس در انتهای حلقه آموزشی با استفاده از انیمیشن یک انیمیشن ایجاد می کنیم matplotlib.

به منظور افزایش تجسم، تلفات و معیارها در مقیاس گزارش رسم می شوند:

import matplotlib.pyplot as plt
import numpy as np
from matplotlib.animation import FuncAnimation
from IPython import display

class TrainingAnimationCallback(keras.callbacks.Callback):
    def __init__(self, duration = 40, fps = 1000/25):
        self.duration = duration
        self.fps = fps
        self.logs_history = ()

    def set_plot(self):   
        self.figure = plt.figure()
        
        plt.xticks(
            range(0,self.params('steps')*self.params('epochs'), self.params('steps')),
            range(0,self.params('epochs')))
        plt.xlabel('Epoch')
        plt.ylabel('Loss & Metrics ($Log_{10}$ scale)')

        self.plot = {}
        for metric in self.model.metrics_names:
          self.plot(metric), = plt.plot((),(), label = metric)
          
        max_y = (max(log.values()) for log in self.logs_history)
        
        self.title = plt.title(f'batches:0')
        plt.xlim(0,len(self.logs_history)) 
        plt.ylim(0,max(max_y))

           
        plt.legend(loc='upper right')
  
    def animation_function(self,frame):
        batch = frame % self.params('steps')
        self.title.set_text(f'batch:{batch}')
        x = list(range(frame))
        
        for metric in self.model.metrics_names:
            y = (log(metric) for log in self.logs_history(:frame))
            self.plot(metric).set_data(x,y)
        
    def on_train_batch_end(self, batch, logs=None):
        logarithm_transform = lambda item: (item(0), np.log(item(1)))
        logs = dict(map(logarithm_transform,logs.items()))
        self.logs_history.append(logs)
       
    def on_train_end(self, logs=None):
        self.set_plot()
        num_frames = int(self.duration*self.fps)
        num_batches = self.params('steps')*self.params('epochs')
        selected_batches = range(0, num_batches , num_batches//num_frames )
        interval = 1000*(1/self.fps)
        anim_created = FuncAnimation(self.figure, 
                                     self.animation_function,
                                     frames=selected_batches,
                                     interval=interval)
        video = anim_created.to_html5_video()
        
        html = display.HTML(video)
        display.display(html)
        plt.close()

ما از همان مدل قبلی استفاده خواهیم کرد، اما با نمونه های آموزشی بیشتر:

import tensorflow as tf
from tensorflow import keras
import numpy as np

model = keras.Sequential()
model.add(keras.layers.Dense(10, input_dim = 1, activation='relu'))
model.add(keras.layers.Dense(10, activation='relu'))
model.add(keras.layers.Dense(1))
model.compile(
    optimizer=keras.optimizers.RMSprop(learning_rate=0.1),
    loss = "mean_squared_error",
    metrics = ("mean_absolute_error")
)

def create_sample(sample_size, train_test_proportion = 0.9):
    x = np.random.uniform(low = 0, high = 10, size = sample_size)
    y = x**2
    train_test_split = int(sample_size*train_test_proportion)
    x_train, x_test = (x(:train_test_split),x(train_test_split:))
    y_train, y_test = (y(:train_test_split),y(train_test_split:))
    return (x_train,x_test,y_train,y_test)

x_train,x_test,y_train,y_test = create_sample(35200)


model.fit(
    x_train,
    y_train,
    batch_size=32,
    epochs=2,
    verbose=0,
    callbacks=(TrainingAnimationCallback()),
)

خروجی ما انیمیشنی از معیارها و تابع ضرر است که در طی آموزش تغییر می کنند process:

نتیجه

در این راهنما، نگاهی به اجرای callback های سفارشی در Keras انداخته ایم.
دو گزینه برای اجرای تماس های سفارشی وجود دارد – از طریق زیر کلاس بندی keras.callbacks.Callback کلاس یا با استفاده از keras.callbacks.LambdaCallback کلاس

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

اگرچه Keras دارای بسیاری از تماس‌های داخلی است، اما دانستن روش اجرای یک callback سفارشی می‌تواند برای برنامه‌های خاص‌تر مفید باشد.