คู่มือการเขียน TensorFlow/Keras Callbacks แบบกำหนดเอง

บทนำ

สมมติว่าคุณต้องการให้โมเดล Keras ของคุณมีพฤติกรรมบางอย่างระหว่างการฝึก การประเมิน หรือการคาดคะเน ตัวอย่างเช่น คุณอาจต้องการบันทึกแบบจำลองของคุณในทุกช่วงเวลาการฝึกอบรม วิธีหนึ่งในการทำเช่นนี้คือการใช้การโทรกลับ

โดยทั่วไป Callbacks เป็นฟังก์ชันที่ถูกเรียกใช้เมื่อมีเหตุการณ์บางอย่างเกิดขึ้น และจะถูกส่งผ่านเป็นอาร์กิวเมนต์ไปยังฟังก์ชันอื่นๆ ในกรณีของ Keras พวกมันเป็นเครื่องมือในการปรับแต่งลักษณะการทำงานของโมเดลของคุณ ไม่ว่าจะเป็นระหว่างการฝึก การประเมิน หรือการอนุมาน แอปพลิเคชั่นบางตัวกำลังบันทึก, การคงอยู่ของโมเดล, การหยุดก่อนกำหนดหรือการเปลี่ยนแปลงอัตราการเรียนรู้ สิ่งนี้ทำได้โดยการส่งรายการโทรกลับเป็นอาร์กิวเมนต์สำหรับ keras.Model.fit(),keras.Model.evaluate() or keras.Model.predict().

บางกรณีการใช้งานทั่วไปสำหรับการเรียกกลับกำลังแก้ไขอัตราการเรียนรู้ การบันทึก การตรวจสอบ และการหยุดการฝึกอบรมก่อนกำหนด Keras มีรายละเอียดการโทรกลับในตัวจำนวนหนึ่ง
ในเอกสาร.

อย่างไรก็ตาม บางแอปพลิเคชันที่เฉพาะเจาะจงอาจต้องมีการเรียกกลับแบบกำหนดเอง ตัวอย่างเช่น ใช้การวอร์มอัพอัตราการเรียนรู้ด้วย Cosine Decay หลังจากช่วงพัก ยังไม่มีในตัว แต่ใช้กันอย่างแพร่หลายและนำมาใช้เป็นตัวกำหนดตารางเวลา

คลาสการโทรกลับและวิธีการ

Keras มีคลาสการโทรกลับเฉพาะ keras.callbacks.Callbackด้วยวิธีการที่สามารถเรียกใช้ระหว่างการฝึกอบรม การทดสอบ และการอนุมานในระดับโลก ชุด หรือระดับยุค เพื่อที่จะ สร้างการโทรกลับที่กำหนดเองเราจำเป็นต้องสร้างคลาสย่อยและแทนที่เมธอดเหล่านี้

พื้นที่ 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()])

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เราเพียงแค่ต้องใช้งานฟังก์ชั่นที่เราต้องการเรียก ห่อมันเป็น 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
มีสองตัวเลือกสำหรับการนำการเรียกกลับแบบกำหนดเองไปใช้ – ผ่านการจัดคลาสย่อย keras.callbacks.Callback คลาสหรือโดยใช้ keras.callbacks.LambdaCallback ชั้นเรียน

เราได้เห็นตัวอย่างหนึ่งที่ใช้ได้จริง LambdaCallbackสำหรับการส่งอีเมลในตอนท้ายของลูปการฝึกอบรม และตัวอย่างหนึ่งของคลาสย่อย Callback คลาสที่สร้างภาพเคลื่อนไหวของลูปการฝึก

แม้ว่า Keras จะมี callback ในตัวหลายตัว การรู้วิธีใช้ callback แบบกำหนดเองอาจมีประโยชน์สำหรับแอปพลิเคชันเฉพาะเจาะจงมากขึ้น