কাস্টম টেনসরফ্লো/কেরাস কলব্যাক লেখার জন্য গাইড

ভূমিকা

ধরুন আপনি প্রশিক্ষণ, মূল্যায়ন বা ভবিষ্যদ্বাণীর সময় আপনার কেরাস মডেলের কিছু নির্দিষ্ট আচরণ করতে চান। উদাহরণস্বরূপ, আপনি প্রতিটি প্রশিক্ষণ যুগে আপনার মডেল সংরক্ষণ করতে চাইতে পারেন। এটি করার একটি উপায় হল কলব্যাক ব্যবহার করা।

সাধারণভাবে, কলব্যাক হল এমন ফাংশন যা কিছু ঘটনা ঘটলে কল করা হয় এবং অন্য ফাংশনে আর্গুমেন্ট হিসেবে পাস করা হয়। কেরাসের ক্ষেত্রে, এগুলি আপনার মডেলের আচরণ কাস্টমাইজ করার একটি হাতিয়ার - তা প্রশিক্ষণ, মূল্যায়ন বা অনুমানের সময়ই হোক। কিছু অ্যাপ্লিকেশন হল লগিং, মডেল অধ্যবসায়, তাড়াতাড়ি বন্ধ করা বা শেখার হার পরিবর্তন করা। এটি আর্গুমেন্ট হিসাবে কলব্যাকের একটি তালিকা পাস করে করা হয় keras.Model.fit(),keras.Model.evaluate() or keras.Model.predict().

কলব্যাকের জন্য কিছু সাধারণ ব্যবহারের ক্ষেত্রে শেখার হার পরিবর্তন করা, লগিং করা, মনিটরিং করা এবং প্রশিক্ষণের তাড়াতাড়ি বন্ধ করা। কেরাসে বিল্ট-ইন কলব্যাকের একটি সংখ্যা রয়েছে, বিস্তারিত
ডকুমেন্টেশনে.

যাইহোক, আরও কিছু নির্দিষ্ট অ্যাপ্লিকেশনের জন্য কাস্টম কলব্যাকের প্রয়োজন হতে পারে। এই ক্ষেত্রে, একটি হোল্ডিং পিরিয়ডের পরে কোসাইন ক্ষয় সহ লার্নিং রেট ওয়ার্মআপ বাস্তবায়ন করা বর্তমানে অন্তর্নির্মিত নয়, তবে একটি সময়সূচী হিসাবে ব্যাপকভাবে ব্যবহৃত এবং গৃহীত হয়।

কলব্যাক ক্লাস এবং এর পদ্ধতি

কেরাসের একটি নির্দিষ্ট কলব্যাক ক্লাস আছে, keras.callbacks.Callback, পদ্ধতি সহ যা প্রশিক্ষণের সময় বলা যেতে পারে, পরীক্ষা এবং বৈশ্বিক, ব্যাচ বা যুগের স্তরে অনুমান। যাতে কাস্টম কলব্যাক তৈরি করুন, আমাদের একটি সাবক্লাস তৈরি করতে হবে এবং এই পদ্ধতিগুলিকে ওভাররাইড করতে হবে।

সার্জারির keras.callbacks.Callback ক্লাসের তিন ধরণের পদ্ধতি রয়েছে:

• বিশ্বব্যাপী পদ্ধতি: শুরুতে বা শেষে বলা হয় fit(), evaluate() এবং predict().
• ব্যাচ-স্তরের পদ্ধতি: ব্যাচ প্রক্রিয়াকরণের শুরুতে বা শেষে বলা হয়।
• যুগ-স্তরের পদ্ধতি: প্রশিক্ষণ ব্যাচের শুরুতে বা শেষে বলা হয়।

বিঃদ্রঃ: প্রতিটি পদ্ধতি নামক একটি dict অ্যাক্সেস আছে 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()])

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 ব্যবহার করে

কেরাসের অন্তর্নির্মিত কলব্যাকগুলির মধ্যে একটি হল 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 একটি পরামিতি হিসাবে ক্লাস।

মডেল প্রশিক্ষণ ভিজ্যুয়ালাইজ করার জন্য একটি কলব্যাক

এই বিভাগে, আমরা একটি কাস্টম কলব্যাকের একটি উদাহরণ দেব যা প্রশিক্ষণের সময় আমাদের মডেলের কর্মক্ষমতা উন্নত করার একটি অ্যানিমেশন তৈরি করে। এটি করার জন্য, আমরা প্রতিটি ব্যাচের শেষে লগের মান সংরক্ষণ করি। তারপর, প্রশিক্ষণ লুপ শেষে, আমরা ব্যবহার করে একটি অ্যানিমেশন তৈরি করি 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()],
)

আমাদের আউটপুট হল মেট্রিক্সের একটি অ্যানিমেশন এবং লস ফাংশন যখন সেগুলি প্রশিক্ষণ প্রক্রিয়ার মাধ্যমে পরিবর্তিত হয়:

উপসংহার

এই নির্দেশিকায়, আমরা কেরাসে কাস্টম কলব্যাক বাস্তবায়নের দিকে নজর দিয়েছি।
কাস্টম কলব্যাক বাস্তবায়নের জন্য দুটি বিকল্প রয়েছে - সাবক্লাসিংয়ের মাধ্যমে keras.callbacks.Callback ক্লাস, বা ব্যবহার করে keras.callbacks.LambdaCallback বর্গ.

আমরা ব্যবহার করে একটি ব্যবহারিক উদাহরণ দেখেছি LambdaCallbackপ্রশিক্ষণ লুপের শেষে একটি ইমেল পাঠানোর জন্য, এবং একটি উদাহরণ সাবক্লাসিং Callback ক্লাস যা প্রশিক্ষণ লুপের একটি অ্যানিমেশন তৈরি করে।

যদিও কেরাসের অনেকগুলি অন্তর্নির্মিত কলব্যাক রয়েছে, কীভাবে একটি কাস্টম কলব্যাক প্রয়োগ করতে হয় তা জানা আরও নির্দিষ্ট অ্যাপ্লিকেশনের জন্য কার্যকর হতে পারে।