المُقدّمة
لنفترض أنك تريد أن يكون لنموذج Keras الخاص بك بعض السلوك المحدد أثناء التدريب أو التقييم أو التنبؤ. على سبيل المثال ، قد ترغب في حفظ النموذج الخاص بك في كل فترة تدريب. طريقة واحدة للقيام بذلك هي استخدام Callbacks.
بشكل عام ، عمليات الاسترجاعات هي وظائف يتم استدعاؤها عند وقوع حدث ما ، ويتم تمريرها كوسيطات إلى دوال أخرى. في حالة Keras ، فهي أداة لتخصيص سلوك نموذجك - سواء كان ذلك أثناء التدريب أو التقييم أو الاستدلال. بعض التطبيقات تقوم بالتسجيل ، أو نموذج المثابرة ، أو التوقف المبكر أو تغيير معدل التعلم. يتم ذلك عن طريق تمرير قائمة عمليات الاسترجاعات كوسيطات لـ keras.Model.fit(),keras.Model.evaluate() or keras.Model.predict().
بعض حالات الاستخدام الشائعة لعمليات الاسترجاعات هي تعديل معدل التعلم والتسجيل والمراقبة والإيقاف المبكر للتدريب. يحتوي Keras على عدد من عمليات الاسترجاعات المضمنة والمفصلة
في الوثائق.
ومع ذلك ، قد تتطلب بعض التطبيقات الأكثر تحديدًا رد اتصال مخصص. على سبيل المثال، تنفيذ إحماء معدل التعلم مع تسوس جيب التمام بعد فترة الانتظار ليس مضمنًا حاليًا ، ولكنه مستخدم على نطاق واسع ومعتمد كمجدول.
فئة رد الاتصال وطرقها
لدى 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()])
تحقق من دليلنا العملي العملي لتعلم 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.
باستخدام LambaCallback
أحد عمليات الاسترجاعات المضمنة في Keras هو ملف LambdaCallback صف دراسي. يقبل رد النداء هذا وظيفة تحدد كيف يتصرف وماذا يفعل! بمعنى ما ، يسمح لك باستخدام أي وظيفة تعسفية بمثابة رد اتصال ، مما يسمح لك بإنشاء عمليات رد نداء مخصصة.
يحتوي الفصل على المعلمات الاختيارية:
-on_epoch_begin
on_epoch_endon_batch_beginon_batch_endon_train_beginon_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()],
)
مخرجاتنا عبارة عن رسم متحرك للمقاييس ووظيفة الخسارة أثناء تغيرها خلال عملية التدريب:
متصفحك لا يدعم فيديو HTML.
وفي الختام
في هذا الدليل ، ألقينا نظرة على تنفيذ عمليات الاسترجاعات المخصصة في Keras.
هناك خياران لتنفيذ عمليات الاسترجاعات المخصصة - من خلال التصنيف الفرعي لملف keras.callbacks.Callback فئة ، أو باستخدام keras.callbacks.LambdaCallback فئة.
لقد رأينا أحد الأمثلة العملية على استخدام LambdaCallbackلإرسال بريد إلكتروني في نهاية حلقة التدريب ، ومثال واحد على تصنيف فرعي لـ Callback الفصل الذي يُنشئ رسمًا متحركًا لحلقة التدريب.
يحتوي Althoug Keras على العديد من عمليات الاسترجاعات المضمنة ، ومعرفة كيفية تنفيذ رد نداء مخصص يمكن أن يكون مفيدًا لتطبيقات أكثر تحديدًا.
