מדריך לכתיבת התקשרויות מותאמות אישית של TensorFlow/Keras

חותמת זמן: 14 באוקטובר 2022 6:30

מבוא

נניח שאתה רוצה שמודל Keras שלך יקבל התנהגות מסוימת במהלך אימון, הערכה או חיזוי. לדוגמה, ייתכן שתרצה לשמור את הדגם שלך בכל עידן אימון. אחת הדרכים לעשות זאת היא באמצעות Callbacks.

באופן כללי, Callbacks הן פונקציות שנקראות כאשר מתרחש אירוע כלשהו, ​​ומועברות כארגומנטים לפונקציות אחרות. במקרה של Keras, הם מהווים כלי להתאמה אישית של ההתנהגות של המודל שלך - בין אם זה במהלך אימון, הערכה או מסקנות. יישומים מסוימים הם רישום, התמדה במודל, עצירה מוקדמת או שינוי קצב הלמידה. זה נעשה על ידי העברת רשימה של Callbacks כארגומנטים עבור keras.Model.fit(),keras.Model.evaluate() or keras.Model.predict().

כמה מקרי שימוש נפוצים להתקשרות חוזרת הם שינוי קצב הלמידה, רישום, ניטור והפסקה מוקדמת של ההדרכה. ל-Keras יש מספר התקשרויות מובנות, מפורטות
בתיעוד.

עם זאת, כמה יישומים ספציפיים יותר עשויים לדרוש התקשרות חוזרת מותאמת אישית. לדוגמה, יישום חימום קצב למידה עם קוסינוס דיקיי לאחר תקופת החזקה אינו מובנה כרגע, אך נמצא בשימוש נרחב ומאומץ כמתזמן.

מחלקת התקשרות חוזרת ושיטותיה

ל-Keras יש מחלקת התקשרות חוזרת ספציפית, 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}")

אם כל אחת מהשיטות הללו לא תוחלף - התנהגות ברירת המחדל תימשך כפי שהייתה בעבר. בדוגמה שלנו - אנו פשוט מדפיסים את היומנים הזמינים ואת הרמה שבה ה-callback מוחל, עם הזחה מתאימה.

בואו נסתכל על הפלטים:

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, עם שיטות עבודה מומלצות, סטנדרטים מקובלים בתעשייה ודף רמאות כלול. תפסיק לגוגל פקודות 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 מחלקה כפרמטר.

התקשרות לאחור להדמיית אימון מודל

בחלק זה, ניתן דוגמה להתקשרות חוזרת מותאמת אישית שעושה אנימציה של שיפור הביצועים של המודל שלנו במהלך האימון. על מנת לעשות זאת, אנו מאחסנים את ערכי היומנים בסוף כל אצווה. לאחר מכן, בסוף לולאת האימון, אנו יוצרים אנימציה באמצעות 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.
ישנן שתי אפשרויות ליישום התקשרויות חוזרות מותאמות אישית - באמצעות סיווג משנה של keras.callbacks.Callback בכיתה, או על ידי שימוש ב- keras.callbacks.LambdaCallback מעמד.

ראינו דוגמה מעשית אחת לשימוש LambdaCallbackעבור שליחת אימייל בסוף לולאת האימון, ודוגמה אחת לסיווג המשנה של Callback שיעור שיוצר אנימציה של לולאת האימון.

למרות של-Keras יש הרבה התקשרויות מובנות, לדעת איך ליישם התקשרות חוזרת מותאמת אישית יכולה להיות שימושית עבור יישומים ספציפיים יותר.

בול זמן:

עוד מ Stackabuse