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

بازنشر افلاطون

دنبال: 0

معرفی

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

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

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

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

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

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

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

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

توجه داشته باشید: هر متد به دیکتی به نام دسترسی دارد logs. کلیدها و مقادیر logs متنی هستند - به رویدادی که متد را فراخوانی می کند بستگی دارد. علاوه بر این، ما به مدل داخل هر متد از طریق مسیر دسترسی داریم self.model ویژگی.

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

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

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}")

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

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

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}

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

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

حالا بیایید با آن تماس بگیریم 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()])

راهنمای عملی و عملی ما برای یادگیری Git را با بهترین روش ها، استانداردهای پذیرفته شده در صنعت و برگه تقلب شامل بررسی کنید. دستورات Google Git را متوقف کنید و در واقع یاد گرفتن آی تی!

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: {}

با اینها - می توانید رفتار را سفارشی کنید، نظارت را تنظیم کنید یا فرآیندهای آموزش، ارزیابی یا استنتاج را تغییر دهید. یک جایگزین برای sublcassing استفاده از 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()],
)

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

مرورگر شما از فیلم HTML پشتیبانی نمی کند.

نتیجه

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

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

Althoug Keras دارای بسیاری از کال بک های داخلی است، دانستن نحوه اجرای یک کال بک سفارشی می تواند برای برنامه های خاص تر مفید باشد.

تمبر زمان: اکتبر 14، 2022نوامبر 6، 2022