कस्टम TensorFlow/Keras कॉलबैक लिखने के लिए मार्गदर्शिका

समय टिकट: 14 अक्टूबर, 2022 6:30 पूर्वाह्न

परिचय

मान लीजिए कि आप चाहते हैं कि आपके केरस मॉडल का प्रशिक्षण, मूल्यांकन या भविष्यवाणी के दौरान कुछ विशिष्ट व्यवहार हो। उदाहरण के लिए, आप प्रत्येक प्रशिक्षण युग में अपने मॉडल को सहेजना चाह सकते हैं। ऐसा करने का एक तरीका कॉलबैक का उपयोग करना है।

सामान्य तौर पर, कॉलबैक ऐसे कार्य होते हैं जिन्हें कुछ घटना होने पर कहा जाता है, और अन्य कार्यों के लिए तर्क के रूप में पारित किया जाता है। केरस के मामले में, वे आपके मॉडल के व्यवहार को अनुकूलित करने के लिए एक उपकरण हैं - चाहे वह प्रशिक्षण, मूल्यांकन या अनुमान के दौरान हो। कुछ एप्लिकेशन लॉगिंग, मॉडल दृढ़ता, प्रारंभिक रोक या सीखने की दर को बदल रहे हैं। यह कॉलबैक की सूची को तर्क के रूप में पारित करके किया जाता है keras.Model.fit(),keras.Model.evaluate() or keras.Model.predict().

कॉलबैक के लिए कुछ सामान्य उपयोग के मामले सीखने की दर को संशोधित करना, लॉगिंग, निगरानी और प्रशिक्षण को जल्दी रोकना है। केरस में कई अंतर्निहित कॉलबैक हैं, विस्तृत
प्रलेखन में.

हालांकि, कुछ और विशिष्ट अनुप्रयोगों के लिए कस्टम कॉलबैक की आवश्यकता हो सकती है। उदाहरण के लिए, होल्डिंग अवधि के बाद कोसाइन क्षय के साथ लर्निंग रेट वार्मअप लागू करना वर्तमान में अंतर्निहित नहीं है, लेकिन व्यापक रूप से उपयोग किया जाता है और शेड्यूलर के रूप में अपनाया जाता है।

कॉलबैक क्लास और उसके तरीके

केरस की एक विशिष्ट कॉलबैक क्लास है, keras.callbacks.Callback, उन विधियों के साथ जिन्हें वैश्विक, बैच या युग स्तर पर प्रशिक्षण, परीक्षण और अनुमान के दौरान बुलाया जा सकता है। के लिए कस्टम कॉलबैक बनाएं, हमें एक उपवर्ग बनाने और इन विधियों को ओवरराइड करने की आवश्यकता है।

RSI 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 सीखने के लिए व्यावहारिक मार्गदर्शिका देखें। Googling 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: {}

इनके साथ - आप व्यवहार को अनुकूलित कर सकते हैं, निगरानी स्थापित कर सकते हैं या अन्यथा प्रशिक्षण, मूल्यांकन या अनुमान की प्रक्रियाओं को बदल सकते हैं। उपवर्गीकरण का एक विकल्प का उपयोग करना है LambdaCallback.

लैंबा कॉलबैक का उपयोग करना

केरस में अंतर्निहित कॉलबैक में से एक है 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()],
)

हमारा आउटपुट मेट्रिक्स और हानि फ़ंक्शन का एक एनीमेशन है क्योंकि वे प्रशिक्षण प्रक्रिया के माध्यम से बदलते हैं:

आपका ब्राउज़र HTML वीडियो का समर्थन नहीं करता है।

निष्कर्ष

इस गाइड में, हमने केरस में कस्टम कॉलबैक के कार्यान्वयन पर एक नज़र डाली है।
कस्टम कॉलबैक को लागू करने के लिए दो विकल्प हैं - उपवर्ग के माध्यम से keras.callbacks.Callback कक्षा, या का उपयोग करके keras.callbacks.LambdaCallback वर्ग.

हमने इसका उपयोग करते हुए एक व्यावहारिक उदाहरण देखा है LambdaCallbackप्रशिक्षण लूप के अंत में एक ईमेल भेजने के लिए, और एक उदाहरण उप-वर्गीकरण करने के लिए Callback क्लास जो ट्रेनिंग लूप का एनिमेशन बनाती है।

अल्थौग केरस में कई अंतर्निहित कॉलबैक हैं, यह जानना कि कस्टम कॉलबैक कैसे लागू किया जाए, यह अधिक विशिष्ट अनुप्रयोगों के लिए उपयोगी हो सकता है।

समय टिकट:

से अधिक स्टैकब्यूज