پیاده سازی یادگیری Q عمیق یا Deep Q-Learning در پایتون – راهنمای گام به گام

در مطلب گذشته به پیاده‌سازی یادگیری Q یا Q-Learning در پایتون پرداختیم. اگر با این مبحث آشنا نیستید، توصیه می‌کنیم ابتدا مطلب پیاده سازی الگوریتم Q-Learning در پایتون – راهنمای گام به گام را مطالعه کنید. در این مطلب قصد داریم نحوه پیاده سازی Deep Q Learning را در پایتون یاد بگیریم و کدنویسی آن را به طور کامل بررسی کنیم.

در مطلب گذشته متوجه شدیم که اگر به تعداد «عمل» (Action) و m «شرایط» (State) داشته باشیم، Q-Table نهایی به شکل $$m\times n$$ خواهد بود. حال اگر تعداد اعمال یا شرایط خیلی زیاد باشد، این شیوه از تخمین مقادیر Q (Quality یا ارزش) بهینه نخواهد بود. برای مثال اگر بخواهیم برای یک ماشین خودران که می‌تواند هزاران حالت از محیط را مشاهده کند و ده‌ها تصمیم بگیرد، به این شیوه عمل کنیم، به مشکل خواهیم خورد.

فیلم آموزش پیاده سازی بازی مار در پایتون با یادگیری تقویتی عمیق در فرادرس

کلیک کنید

یکی از راه حل‌های این مشکل، استفاده از شبکه‌های عصبی مصنوعی (Artificial Neural Network یا ANN) است. شبکه‌های عصبی عمیق (Deep Neural Network یا DNN) در صورتی که به درستی آموزش ببیند، می‌توان نتایج آن را تعمیم داد. در این شرایط به جای ذخیره مقادیر Q در یک جدول، می‌توانیم پیش‌بینی مدل برای Q را استفاده کنیم.

در مطلب گذشته، به شکل زیر یک جدولی ایجاد کردیم که با دریافت شرایط و عمل، Q آن را برمی‌گرداند:

اما برای Deep Q-Learning به شکل زیر عمل می‌کنیم:

دانلود کد آماده برای پیاده سازی Deep Q Learning

با توجه به پیچیدگی کدهای مربوط به پیاده سازی Deep Q Learning، کدهای آماده این مطلب در فایل زیر قابل دانلود است. اما از آنجا که هدف این مطلب، آموزش پیاده سازی Deep Q Learning در پایتون است،‌ بخش‌های مختلف این کدها توضیح داده شده است.

• برای دانلود کد آماده در پیاده سازی Deep Q Learning + اینجا کلیک کنید.

پیاده سازی Deep Q Learning در پایتون

در این روش یک شبکه عصبی عمیق ایجاد می‌شود و به نحوی آموزش می‌بیند که با دریافت شرایط، Q مربوط به هر عمل را در خروجی برگرداند.

فیلم آموزش پیاده سازی بازی مار در پایتون با یادگیری تقویتی عمیق در فرادرس

کلیک کنید

این شبکه عصبی باید به اندازه ابعاد شرایط ورودی داشته باشید و به تعداد اعمال خروجی داشته باشد. به این ترتیب حتی در صورتی که شرایط پیوسته نیز باشد، باز هم می‌توان از شبکه عصبی کمک گرفت.

در کنار شباهت‌های زیادی که بین ­Q-Learning و Deep Q-Learning وجود دارد، تفاوت‌های بسیار مهم نیز بین این دو روش وجود دارد که در ادامه به آن‌ها پرداخته می‌شود:

در روش Q-Learning برای یادگیری مقادیر Q از فرمول زیر استفاده می‌کردیم:

$$Q(s, a)^{\text {new }} \leftarrow Q(s, a)^{\text {old }}+\alpha\left[r+\gamma \max _{a^{\prime}} Q\left(s^{\prime}, a^{\prime}\right)^{\text {old }}-Q(s, a)^{\text {old }}\right]$$

با به‌روزرسانی این مقادیر، یادگیری عامل به اتمام می‌رسید، اما در Deep Q-Learning پس از به‌روزرسانی مقادیر Q، باید مدل عمیق نیز بر روی این داده‌ها آموزش ببیند.

در روش Q-Learning اغلب مقادیر کوچک برای نرخ یادگیری (Learning Rate) $$\alpha$$ مناسب است، اما در Deep Q-Learning اغلب مقادیر بالای $$\alpha$$ مناسب است. در اغلب موارد مقدار نرخ یادگیری برابر با 1 در نظر گرفته می‌شود که در این صورت رابطه به‌روزرسانی مقادیر Q به شکل زیر ساده می‌شود:

$$Q(s, a)^{n e w} \leftarrow r+\gamma \max _{a^{\prime}} Q\left(s^{\prime}, a^{\prime}\right)^{\text {old }}$$

نکات زیر را نیز باید در زمان پیاده سازی Deep Q Learning در نظر داشته باشید:‌

• در روش Q-Learning تمامی مقادیر Q به صورت ذخیره شده آماده بودند، اما در Deep Q-Learning باید شرایط فعلی را وارد مدل کنیم تا در خروجی مقادیر Q را تولید کند.
• آموزش عامل در Deep Q-Learning زمان‌برتر از Q-Learning است.
• در آموزش Deep Q-Network، با به‌روزرسانی وزن‌ها نسبت به یک داده (شامل شرایط اولیه، عمل انجام شده، پاداش دریافتی و شرایط بعدی)، مقادیر Q برای سایر شرایط نیز تغییر می‌کند، درحالیکه در Q-Table این اتفاق رخ نمی‌داد. این اتفاق باعث می‌شود که مدل مرتباً یک مقداری از Q را تعقیب کند و هرگز به آن نرسد.
  این اتفاق یک مشکل برای آموزش مدل است که در ادامه مطلب به روش مقابله با آن خواهیم پرداخت. اگر محیط را یک تور در نظر بگیریم، هر گره از آن را یک شرایط در نظر بگیریم و این تور را از محلی آویزان کنیم، ارتفاع آن برابر با Qها خواهد بود. اگر پس از یادگیری مدل، Q یکی از این شرایط افزایش پیدا کند (ارتفاع آن زیاد شود)، نقاط همسایه نیز به سمت بالا حرکت خواهند کرد. در این شرایط عبارت می‌تواند تحت یک فرآیند خودتنظیمی مثبت (Positive Feedback) مرتباً افزایش می‌یابد و باعث افزایش $$Q(s, a)^{n e w}$$ و برعکس آن نیز رخ می‌دهد. در این شرایط مقادیر Q واگرا خواهد شد.
• در Q-Learning مقادیر اپسیلون یا $$\epsilon$$ را به صورت خطی در طول اپیزودها (Episode) کاهش می‌دادیم، اما برای Deep Q-Learning اغلب دو فاز وجود دارد، در فاز اول کاهش (Decay) اپسیلون را به صورت خطی یا نمایی شاهد هستیم اما در فاز دوم مقادیر اپسیلون ثابت شده و برابر با مقدار نهایی می‌شود.
• در ­Q-Learning چون مدلی برای آموزش وجود ندارد، از الگوریتم‌های بهینه‌ساز (Optimizer) استفاده نمی‌شود، اما برای آموزش Deep Q-Network از الگوریتم‌های بهینه‌ساز استفاده می‌کنیم، بنابراین باید ورودی‌ها و خروجی‌های مدل باید در بازه مشخصی (اغلب در بازه $$[-1,+1]$$ یا $$[0,+1]$$ ) قرار گیرند یا میانگین و واریانس آن‌ها به ترتیب برابر با 0 و 1 باشد. به این دلیل، باید مقادیر State و Q در صورت نیاز تغییر مقیاس (Scaling) داده شوند.

فیلم آموزش یادگیری عمیق با پایتون – تنسورفلو و کراس TensorFlow و Keras در فرادرس

کلیک کنید

حال کدنویسی را شروع می‌کنیم. کد نهایی حاصل، یک کلاس (Class) به نام WORLD خواهد بود که به شکل زیر استفاده خواهد شد:

1World = WORLD()
2
3World.CreateModel()
4World.CompileModel()
5World.ModelSummary()
6
7World.PlotEpsilons()
8World.PlotState2Reward()
9
10World.Train()
11
12World.SaveModel()
13
14World.PlotActionLog()
15World.PlotEpisodeLog()
16World.PlotModelPrediction('Model Prediction For Q Against Real Values (Final)')
17
18for _ in range(10):
19    World.Test()

حال برای شروع، کتابخانه‌های مورد نیاز را فراخوانی می‌کنیم:

1import os as os
2import numpy as np
3import typing as typ
4import random as ran
5import colorama as col
6import tensorflow as tf
7import collections as co
8import matplotlib.pyplot as plt
9import tensorflow.keras.layers as lay
10import tensorflow.keras.models as mod
11import tensorflow.keras.losses as los
12import tensorflow.keras.optimizers as opt
13import tensorflow.keras.activations as act
14import tensorflow.keras.initializers as ini

کاربرد این کتابخانه‌هادر پیاده سازی Deep Q Learning به ترتیب زیر است:

• ماژول os برای استفاده از امکانات سیستم عامل (Operating System) است.
• کتابخانه Numpy برای محاسبات برداری، ذخیره داده و برخی تنظیمات برنامه استفاده خواهد شد.
• ماژول typing برای تعیین جنس متغیرها استفاده خواهد شد. تعیین جنس متغیرها برای استفاده‌های بعدی و افزایش خوانایی کد بسیار مهم است.
• ماژول random برای تولید اعداد تصادفی استفاده می‌شود. هرچند که برخی کتابخانه‌ها نیز خود برای تولید اعداد تصادفی امکاناتی دارند.
• کتابخانه Colorama برای نوشتن متون رنگی کاربرد دارد. رنگی کردن نوشته‌ها در شرایط که مقدار زیادی متن در خروجی کد ایجاد می‌شود، می‌تواند مفید باشد.
• کتابخانه Tensorflow برای ایجاد و آموزش مدل عمیق استفاده خواهد شد.
• ماژول collections انواعی از ساختمان‌های داده (Data Structure) مختلف را دارد که از نوع Deque برای ایجاد حافظه برای مدل استفاده خواهد شد.
• کتابخانه Matplotlib نیز برای رسم نمودار و مصورسازی (Visualization) عملکرد مدل استفاده خواهد شد.
• شش مورد بعدی، بخش‌های مختلفی از کتابخانه Keras هستند که برای سهولت در کدنویسی هریک به صورت جداگانه به اختصار فراخوانی شده‌اند.

حال تمامی کتابخانه‌های مورد نیاز را فراخوانی کردیم. در اولین قدم کلاس مورد نظر را ایجاد می‌کنیم:

1class WORLD:

این کلاس محیطی را شبیه‌سازی خواهد کرد که عامل بتواند دمای یک فضاپیما را تنظیم کنم. دمای این فضاپیما متغیری بین $$-100$$ و $$+100$$ خواهد بود و عامل با تنظیم آن بر روی 0، به بیشترین پاداش ممکن خواهد رسید. عامل نیز در هر «گام» (Step)، 2 عمل برای انجام خواهد داشت:

• کاهش 1 واحدی دما
• افزایش 1 واحدی دما

در ابتدا «متد» (Method) سازنده را ایجاد می‌کنیم. این متد 20 ورودی به شکل زیر خواهد داشت:

1    def __init__(self,
2                 MinT:typ.Union[int, float]=-100,
3                 MaxT:typ.Union[int, float]=+100,
4                 nEpisode:int=300,
5                 mStep:int=70,
6                 qLR:float=9e-1,
7                 Gamma:float=9e-1,
8                 Epsilon0:float=1,
9                 Epsilon1:float=2e-2,
10                 nDecayEpisode:int=200,
11                 nDense:list[int]=[256, 256],
12                 Activation:str='elu',
13                 Optimizer:str='Adam',
14                 Loss:str='huber',
15                 mLR:float=3e-3,
16                 nEpoch:int=1,
17                 TrainOn:int=16,
18                 sBatch:int=32,
19                 sMemory:int=1024,
20                 Verbose:int=0,
21                 RandomState:typ.Union[int, None]=None):

این ورودی‌ها به ترتیب موارد زیر را تنظیم می‌کنند:

• MinT: کمترین دمای ممکن که می‌تواند ایجاد شود. این ورودی به صورت پیش‌فرض بر روی $$-100$$ تنظیم شده است.
• MaxT: بیشترین دمای ممکن که می‌تواند ایجاد شود. این ورودی به صورت پیش فرض بر روی $$+100$$ تنظیم شده است.

دو متغیر فوق برای جلوگیری از افزایش یا کاهش بیش از اندازه دما ایجاد شده‌اند. توجه داشته باشید شبکه‌های عصبی مصنوعی اغلب در بازه مشخصی «تعمیم‌پذیری» (Generalizability) خوبی دارند و نمی‌توان بازه‌های بزرگی را به این منظور استفاده کرد. داده‌های استفاده شده برای آموزش مدل همواره باید توزیع مناسبی در کل بازه ورودی داشته باشند.

سایر ورودی‌ها به ترتیب در زیر آورده شده‌اند.

• nEpisode: تعداد اپیزود‌های آموزش مدل را نشان می‌دهد. مقدار پیشفرض این ورودی بر روی 300 تنظیم می‌شود. تعیین دقیق این اعداد ممکن نیست و اغلب به صورت حدودی و با سعی و خطا است.
• mStep: بیشترین تعداد ممکن گام برای عامل را نشان می‌دهد. با توجه به اینکه بهترین دمای ممکن برابر با 0 است و بیشترین فاصله از این نقطه از این نقطه برابر با 100 است، عامل می‌تواند در 100 گام خود را به این نقطه برساند. بنابراین 70 گام به عنوان پیشفرض می‌تواند مناسب باشد.
• qLR: این ورودی نرخ یادگیری برای Q را نشان می‌دهد. مقدار پیشفرض آن برابر با 0.9 تنظیم شده است، درحالیکه برای Q-Learning از مقدار 0.1 استفاده کردیم.
• Gamma: این ورودی نرخ تخفیف را نشان می‌دهد.
• Epsilon0: مقدار اولیه اپسیلون است که باید بیشتر از 0.9 باشد. به صورت پیشفرض مقدار آن برابر با 1 در نظر گرفته شده است.
• Epsilon1: مقدار نهایی اپسیلون است که باید کمتر از 0.1 باشد. به صورت پیشفرض مقدار آن برابر با 0.02 در نظر گرفته شده است. با توجه به اینکه قصد داریم در فاز دوم، مقدار اپسیلون را تا انتها ثابت نگه داریم، بهتر است Epsilon1 بزرگ‌تر از 0 باشد.
• nDecayEpisode: این ورودی، تعداد اپیزودهای فاز اول را نشان می‌دهد. در طول این تعداد اپیزود اول، مقدار اپسیلون مرتباً کاهش خواهد یافت تا به مقدار نهایی برسد. در ادامه اپیزودها، مقدار اپسیلون ثابت خواهد ماند. مقدار این متغیر همواره باید کمتر از nEpisode باشد. این مقدار به صورت پیش‌فرض برابر با 200 تنظیم شده است که ۶۶ درصد کل اپیزودها را شامل می‌شود.
• nDense: این ورودی یک لیست است که مقادیر داخل آن باید اعدادی صحیح باشند. تعداد این اعداد نشان‌دهنده تعداد لایه‌های پنهان شبکه عصبی مورد استفاده است و مقدار آن‌ها نشان‌دهنده تعداد نورون‌های هر لایه است. با آزمون و خطا در مورد این مسئله، مقدار برای این ورودی انتخاب شده است.
• Activation: این ورودی تابع فعال‌سازی (Activation Function) نورون‌های لایه پنهان شبکه را تعیین می‌کند. مقدار آن به صورت پیشفرض ELU (Exponential Linear Unit) تعیین شده است. این تابع در مقایسه با ReLU (Rectified Linear Unit) عملکرد بهتری نشان می‌دهد و به دلیل داشتن گرادیان غیرصفر برای ورودی‌های منفی، آموزش آن‌ها بهتر و سریع‌تر انجام می‌شود. تابع Sigmoid نیز می‌تواند گزینه مناسب دیگری باشد، اما در صورت استفاده از دو لایه پنهان، می‌تواند بر روی سرعت همگرایی مدل اثر منفی بگزارد. در اغلب موارد، تابع ReLU به عنوان تابع فعال‌سازی لایه‌های پنهان استفاده می‌شود.
• Optimizer: این ورودی الگوریتم بهینه‌ساز شبکه عصبی را نشان می‌دهد. برای بهینه‌سازی مدل‌های Tensorflow الگوریتم‌های متفاوتی وجود دارد که SGD (Stochastic Gradient Descent) و Adam دو مورد از پرکاربردترین آن‌ها است. برای این مسئله، الگوریتم Adam به صورت پیشفرض استفاده خواهد شد.
• Loss: این ورودی تابع هزینه (Loss Function) مورد استفاده برای آموزش شبکه عصبی را نشان می‌دهد. در اغلب مسائل رگرسیون از خطای MSE یا Mean Squared Error استفاده می‌شود. برای مسائل رگرسیونی که داده‌های پرت داریم و قصد داریم اثر آن‌ها را کاهش دهیم نیز از MAE یا Mean Absolute Error استفاده می‌کنیم. اما برای Deep Q-Networkها اغلب Huber Loss توصیه می‌شود.
• mLR: این ورودی نرخ یادگیری شبکه عصبی را تعیین می‌کند. در اغلب موارد، نرخ یادگیری 0.001 استفاده می‌شود اما در این مسئله مقدار 0.003 مناسب است.
• nEpoch: این ورودی تعداد مراحل یا Epoch آموزش مدل بر روی هر Batch را نشان می‌دهد. با توجه به اینکه در این مسئله مجموعه داده‌ای وجود ندارد و عامل با برقرار تعامل با محیط، اطلاعاتی جمع‌آوری می‌کند، باید مدل به جای چندین Batch، هر بار روی تنها 1 Batch آموزش ببیند. بنابراین عدد 1 مناسب است.
• TrainOn: این ورودی تعیین می‌کند که بعد از انجام چند گام توسط عامل، مدل بر روی یک Batch آموزش ببیند. مقدار پیشفرض این متغیر 16 است، بنابراین پس از هر 16 گام، مدل یک مرحله بر روی یک Batch آموزش دیده و وزن‌های آن به‌روزرسانی خواهد شد.
• sBatch: این ورودی سایز Batchها را نشان می‌دهد. این مقدار اغلب از توان‌های 2 انتخاب می‌شود. مقادیر 32، 64 و 128 در اغلب موارد نتایج مطلوبی دارند.
• sMemory: این ورودی سایز حافظه مدل را تعیین می‌کند. یکی از راه‌های جلوگیری از مشکل واگرایی مقادیر Q استفاده از حافظه و آموزش مدل بر روی تجارب قدیمی‌تر است. این ورودی تعیین می‌کند که حداکثر چند تجربه گذشته می‌تواند در آموزش مدل استفاده شود. مقدار 1024 می‌تواند داده‌هایی از 12 اپیزود قبل را در خود جای دهد، بنابراین مناسب خواهد بود. برای مسائل پیچیده‌تر، باید مقدار این متغیر حداقل 20 برابر شود.
• Verbose: شبکه‌های عصبی Tensorflow در طول آموزش و پیشبینی پیام‌هایی مبنی بر میزان پیشرفت و نتایج نمایش می‌دهد. به دلیل استفاده زیاد از این توابع، مقدار بسیار زیادی پیام در خروجی نمایش داده می‌شود که مناسب نیست. به همین دلیل ورودی دیگری به نام Verbose برای این توابع وجود دارد که میزان پرحرف بودن آن‌ها را تنظیم می‌کند. برای اضافه کردن امکان تنظیم این موضوع، ورودی Verbose را ایجاد می‌کنیم. مقدار پیش‌فرض این ورودی 0 است که باعث می‌شود هیچ پیامی از سمت این توابع ایجاد نشود.
• RandomState: تعدادی از فرآیند‌های مربوط به Tensorflow و کدی که می‌خواهیم بنویسیم، از اعداد تصادفی استفاده می‌کنند. به همین دلیل ورودی دیگری نیز ایجاد می‌کنیم تا این فرآیند‌ها را در صورت نیاز کنترل کنیم. مقدار این ورودی می‌تواند یک عدد صحیح یا None باشد. در صورتی که مقدار آن None باشد، فرآیندهای تصادفی در هر بار اجرا به شکل متفاوتی رخ می‌دهند و نتایج قابل بازتولید نخواهد (Not Reproducible) بود. در مقابل با تعیین یک عدد صحیح، نتایج یکسانی تولید خواهد شد. توجه داشته باشید که با تغییر این عدد به عددی دیگر، اعداد تصادفی تولید شده متفاوت با حالت قبلی خواهد بود اما همچنان نتایج قابل بازتولید خواهد بود.

فیلم آموزش پایتون – کار با کتابخانه PyAutoGUI برای خودکارسازی GUI در فرادرس

کلیک کنید

به این ترتیب تمامی ورودی‌های متد سازنده بررسی و توضیح داده شد. حال ورودی‌های دریافت شده را در شیء (Object) ذخیره می‌کنیم تا در متدهای بعدی استفاده کنیم:

1    def __init__(self,
2                 MinT:typ.Union[int, float]=-100,
3                 MaxT:typ.Union[int, float]=+100,
4                 nEpisode:int=300,
5                 mStep:int=70,
6                 qLR:float=9e-1,
7                 Gamma:float=9e-1,
8                 Epsilon0:float=1,
9                 Epsilon1:float=2e-2,
10                 nDecayEpisode:int=200,
11                 nDense:list[int]=[256, 256],
12                 Activation:str='elu',
13                 Optimizer:str='Adam',
14                 Loss:str='huber',
15                 mLR:float=3e-3,
16                 nEpoch:int=1,
17                 TrainOn:int=16,
18                 sBatch:int=32,
19                 sMemory:int=1024,
20                 Verbose:int=0,
21                 RandomState:typ.Union[int, None]=None):
22        self.MinT = MinT
23        self.MaxT = MaxT
24        self.nEpisode = nEpisode
25        self.mStep = mStep
26        self.qLR = qLR
27        self.Gamma = Gamma
28        self.Epsilon0 = Epsilon0
29        self.Epsilon1 = Epsilon1
30        self.nDecayEpisode = nDecayEpisode
31        self.nDense = nDense
32        self.Activation = Activation
33        self.Optimizer = Optimizer
34        self.Loss = Loss
35        self.mLR = mLR
36        self.nEpoch = nEpoch
37        self.TrainOn = TrainOn
38        self.sBatch = sBatch
39        self.sMemory = sMemory
40        self.Verbose = Verbose
41        self.RandomState = RandomState

متد سازنده در انتها دو متد دیگر را نیز فراخوانی می‌کند که اولی Random State را در صورت نیاز تعیین می‌کند و دومی برخی تنظیمات دیگر را:

1    def __init__(self,
2                 MinT:typ.Union[int, float]=-100,
3                 MaxT:typ.Union[int, float]=+100,
4                 nEpisode:int=300,
5                 mStep:int=70,
6                 qLR:float=9e-1,
7                 Gamma:float=9e-1,
8                 Epsilon0:float=1,
9                 Epsilon1:float=2e-2,
10                 nDecayEpisode:int=200,
11                 nDense:list[int]=[256, 256],
12                 Activation:str='elu',
13                 Optimizer:str='Adam',
14                 Loss:str='huber',
15                 mLR:float=3e-3,
16                 nEpoch:int=1,
17                 TrainOn:int=16,
18                 sBatch:int=32,
19                 sMemory:int=1024,
20                 Verbose:int=0,
21                 RandomState:typ.Union[int, None]=None):
22        self.MinT = MinT
23        self.MaxT = MaxT
24        self.nEpisode = nEpisode
25        self.mStep = mStep
26        self.qLR = qLR
27        self.Gamma = Gamma
28        self.Epsilon0 = Epsilon0
29        self.Epsilon1 = Epsilon1
30        self.nDecayEpisode = nDecayEpisode
31        self.nDense = nDense
32        self.Activation = Activation
33        self.Optimizer = Optimizer
34        self.Loss = Loss
35        self.mLR = mLR
36        self.nEpoch = nEpoch
37        self.TrainOn = TrainOn
38        self.sBatch = sBatch
39        self.sMemory = sMemory
40        self.Verbose = Verbose
41        self.RandomState = RandomState
42        self.SetRandomState()
43        self.ApplySettings()

برای پیاده‌سازی متد SetRandomState به شکل زیر عمل می‌کنیم:

1    def SetRandomState(self):
2        if self.RandomState is not None:
3            ran.seed(self.RandomState)
4            np.random.seed(self.RandomState)
5            tf.random.set_seed(self.RandomState)
6            os.environ['PYTHONHASHSEED'] = str(self.RandomState)

توجه داشته باشید که تنها در صورتی Random State را تعیین می‌کنیم که مقدار آن None نباشد. نکته مهم دیگری که باید به آن توجه کرد، وجود Random Seed در کتابخانه‌های مختلف است. بنابراین باید همه آن‌ها در صورت نیاز تنظیم شوند. دو ماژول os و random تنها در این بخش از کد استفاده شده‌اند، اما با توجه به اینکه ممکن است بعداً برای پروژه‌های بزرگ‌تر کدهای بیشتری داشته باشیم، بهتر است ابتدای کد Fix شوند.

حال متد ApplySettings را پیاده‌سازی می‌کنیم. در این متد ابتدا Style مربوط به نمودارها را تنظیم می‌کنیم، سپس برخی قراردادهای مربوط به محیط و عامل را اضافه می‌کنیم:

1    def ApplySettings(self):
2        plt.style.use('ggplot')
3        self.Code2Actions = {0: 'Down', 1: 'Up'}
4        self.Action2Change = {0: -1, 1: +1}
5        self.nAction = len(self.Code2Actions)

حال شماره اپیزود را نیز بر روی $$-1$$ تنظیم می‌کنیم. با توجه به اینکه در ابتدای هر اپیزود، مقدار آن یک واحد افزایش خواهد یافت، اولین مقدار آن برابر با 0 خواهد بود. حال مقادیر اپسیلون را تعیین می‌کنیم. برای فاز 1 از روند کاهش خطی استفاده می‌کنیم و برای فاز 2 مقدار ثابت نهایی را استفاده می‌کنیم:

1        nConstantEpisode = self.nEpisode - self.nDecayEpisode
2        epsA = np.linspace(start=self.Epsilon0,
3                           stop=self.Epsilon1,
4                           num=self.nDecayEpisode)
5        epsB = self.Epsilon1 * np.ones(nConstantEpisode)
6        self.Epsilons = np.hstack((epsA, epsB))

توجه داشته باشید که تابع numpy.ones آرایه‌ای با ابعاد وارد شده ایجاد می‌کند که تمامی درایه‌های آن برابر با 1 است. ضرب کردن یک عدد اعشاری در آن، باعث تغییر تمامی درایه‌ها به آن عدد خواهد شد.

برای دو فاز، مقادیر اپسیلون در آرایه‌های epsA و epsB ذخیره می‌شود. برای جمع کردن آن‌ها پشت سر هم، می‌توانیم از numpt.hstack استفاده کنیم.

همانند کد قبلی دو آرایه نیز برای ذخیره پاداش هر عمل و مجموع پاداش هر اپیزود ایجاد می‌کنیم:

1    def ApplySettings(self):
2        plt.style.use('ggplot')
3        self.Code2Actions = {0: 'Down', 1: 'Up'}
4        self.Action2Change = {0: -1, 1: +1}
5        self.nAction = len(self.Code2Actions)
6        self.Episode = -1
7        nConstantEpisode = self.nEpisode - self.nDecayEpisode
8        epsA = np.linspace(start=self.Epsilon0,
9                           stop=self.Epsilon1,
10                           num=self.nDecayEpisode)
11        epsB = self.Epsilon1 * np.ones(nConstantEpisode)
12        self.Epsilons = np.hstack((epsA, epsB))
13        self.ActionLog = np.zeros((self.nEpisode, self.mStep))
14        self.EpisodeLog = np.zeros(self.nEpisode)

حال تابع فعال‌سازی مربوط به لایه خروجی را نیز تعیین می‌کنیم. با توجه به اینکه خروجی شبکه پیوسته است، تنها مقدار مناسب برای تابع فعال‌سازی لایه خروجی تابع خطی (Linear) است، به همین دلیل این متغیر در ورودی دریافت نمی‌شود:

1    def ApplySettings(self):
2        plt.style.use('ggplot')
3        self.Code2Actions = {0: 'Down', 1: 'Up'}
4        self.Action2Change = {0: -1, 1: +1}
5        self.nAction = len(self.Code2Actions)
6        self.Episode = -1
7        nConstantEpisode = self.nEpisode - self.nDecayEpisode
8        epsA = np.linspace(start=self.Epsilon0,
9                           stop=self.Epsilon1,
10                           num=self.nDecayEpisode)
11        epsB = self.Epsilon1 * np.ones(nConstantEpisode)
12        self.Epsilons = np.hstack((epsA, epsB))
13        self.ActionLog = np.zeros((self.nEpisode, self.mStep))
14        self.EpisodeLog = np.zeros(self.nEpisode)
15        self.oActivaiton = 'linear'

حال در این بخش قصد داریم روش دیگری نیز به کار گیریم تا مشکل مربوط به واگرایی مقادیر Q را رفع کنیم. در این روش مقادیر هدف شبکه عصبی را بین یک بیشینه و کمینه مقدار محدود می‌کنیم، به این ترتیب، امکان واگرایی مدل نخواهد بود. به این منظور باید کمترین و بیشترین مقدار پاداش را محاسبه کنیم. ابتدا یک آرایه از 201 شرایط در محدوده مشخص شده انتخاب می‌کنیم. سپس مقادیر پاداش را برای هر کدام محاسبه می‌کنیم:

1    def ApplySettings(self):
2        plt.style.use('ggplot')
3        self.Code2Actions = {0: 'Down', 1: 'Up'}
4        self.Action2Change = {0: -1, 1: +1}
5        self.nAction = len(self.Code2Actions)
6        self.Episode = -1
7        nConstantEpisode = self.nEpisode - self.nDecayEpisode
8        epsA = np.linspace(start=self.Epsilon0,
9                           stop=self.Epsilon1,
10                           num=self.nDecayEpisode)
11        epsB = self.Epsilon1 * np.ones(nConstantEpisode)
12        self.Epsilons = np.hstack((epsA, epsB))
13        self.ActionLog = np.zeros((self.nEpisode, self.mStep))
14        self.EpisodeLog = np.zeros(self.nEpisode)
15        self.oActivaiton = 'linear'
16        States = np.linspace(start=self.MinT,
17                             stop=self.MaxT,
18                             num=201)
19        Rewards = self.State2Reward(States)

در این بخش از متد State2Reward استفاده شده است که در ادامه به آن خواهیم پرداخت.

حال کمترین و بیشترین مقدار پاداش را محاسبه می‌کنیم:

1    def ApplySettings(self):
2        plt.style.use('ggplot')
3        self.Code2Actions = {0: 'Down', 1: 'Up'}
4        self.Action2Change = {0: -1, 1: +1}
5        self.nAction = len(self.Code2Actions)
6        self.Episode = -1
7        nConstantEpisode = self.nEpisode - self.nDecayEpisode
8        epsA = np.linspace(start=self.Epsilon0,
9                           stop=self.Epsilon1,
10                           num=self.nDecayEpisode)
11        epsB = self.Epsilon1 * np.ones(nConstantEpisode)
12        self.Epsilons = np.hstack((epsA, epsB))
13        self.ActionLog = np.zeros((self.nEpisode, self.mStep))
14        self.EpisodeLog = np.zeros(self.nEpisode)
15        self.oActivaiton = 'linear'
16        States = np.linspace(start=self.MinT,
17                             stop=self.MaxT,
18                             num=201)
19        Rewards = self.State2Reward(States)
20        self.MinReward = Rewards.min()
21        self.MaxReward = Rewards.max()

با توجه به فرمول مربوط به به‌روزرسانی Qها، می‌توان به صورت حدودی، مقادیر زیر را به عنوان بیشینه و کمینه مقدار Q در نظر گرفت:

1    def ApplySettings(self):
2        plt.style.use('ggplot')
3        self.Code2Actions = {0: 'Down', 1: 'Up'}
4        self.Action2Change = {0: -1, 1: +1}
5        self.nAction = len(self.Code2Actions)
6        self.Episode = -1
7        nConstantEpisode = self.nEpisode - self.nDecayEpisode
8        epsA = np.linspace(start=self.Epsilon0,
9                           stop=self.Epsilon1,
10                           num=self.nDecayEpisode)
11        epsB = self.Epsilon1 * np.ones(nConstantEpisode)
12        self.Epsilons = np.hstack((epsA, epsB))
13        self.ActionLog = np.zeros((self.nEpisode, self.mStep))
14        self.EpisodeLog = np.zeros(self.nEpisode)
15        self.oActivaiton = 'linear'
16        States = np.linspace(start=self.MinT,
17                             stop=self.MaxT,
18                             num=201)
19        Rewards = self.State2Reward(States)
20        self.MinReward = Rewards.min()
21        self.MaxReward = Rewards.max()
22        self.MinQ = (1 + self.qLR * self.Gamma) * self.MinReward
23        self.MaxQ = (1 + self.qLR * self.Gamma) * self.MaxReward

توجه داشته باشید که این روش تخمینی از مقادیر مرزی است و ممکن است برای برخی مسائل مناسب نباشد.

حال یک حافظه و «شمارنده» (Counter) برای عامل ایجاد می‌کنیم:

1    def ApplySettings(self):
2        plt.style.use('ggplot')
3        self.Code2Actions = {0: 'Down', 1: 'Up'}
4        self.Action2Change = {0: -1, 1: +1}
5        self.nAction = len(self.Code2Actions)
6        self.Episode = -1
7        nConstantEpisode = self.nEpisode - self.nDecayEpisode
8        epsA = np.linspace(start=self.Epsilon0,
9                           stop=self.Epsilon1,
10                           num=self.nDecayEpisode)
11        epsB = self.Epsilon1 * np.ones(nConstantEpisode)
12        self.Epsilons = np.hstack((epsA, epsB))
13        self.ActionLog = np.zeros((self.nEpisode, self.mStep))
14        self.EpisodeLog = np.zeros(self.nEpisode)
15        self.oActivaiton = 'linear'
16        States = np.linspace(start=self.MinT,
17                             stop=self.MaxT,
18                             num=201)
19        Rewards = self.State2Reward(States)
20        self.MinReward = Rewards.min()
21        self.MaxReward = Rewards.max()
22        self.MinQ = (1 + self.qLR * self.Gamma) * self.MinReward
23        self.MaxQ = (1 + self.qLR * self.Gamma) * self.MaxReward
24        self.Memory = co.deque(maxlen=self.sMemory)
25        self.Counter = 0

توجه داشته باشید که collections.deque می‌تواند یک بیشینه طول نیز داشته باشد. در صورتی که به بیشینه طول رسیده باشد و مقدار جدیدی به آن اضافه (Append) شود، اولین مقدار موجود (عضو شماره 0 یا سمت چپ) حذف خواهد شد. به این ترتیب حافظه عامل از یک مقدار مشخص بزرگ‌تر نخواهد شد و داده‌های بدون کاربرد به مرور زمان جای خود را به داده‌های جدید خواهد داد. وجود حافظه برای جلوگیری از واگرایی مقادیر Q از اهمیت بالایی برخوردار است. متغیر Counter نیز تعداد داده‌های جدید را نشان می‌دهد که پس از آخرین به‌روزرسانی وزن‌ها اضافه شده‌اند.

فیلم آموزش پیاده سازی اندیکاتورهای تکنیکال با پایتون – بخش دوم در فرادرس

کلیک کنید

مقادیر دما یا همان شرایط عددی در بازه $$[-100,+100]$$ است. این مقیاس می‌تواند در آموزش مدل مشکل‌ساز شود. به همین دلیل مقیاس آن را به بازه $$[-1,+1]$$ تغییر می‌دهیم. به این منظور می‌توانیم تبدیل خطی زیر را انجام دهیم:

$$\begin{gathered}
x^{\prime}=2 \times\left(\frac{x-\operatorname{Min}}{\operatorname{Max}-\operatorname{Min}}\right)-1=\frac{2 \times x-2 \times \operatorname{Min}}{\operatorname{Max}-\operatorname{Min}}-1=\frac{2 \times x-\operatorname{Min}-\operatorname{Max}}{\operatorname{Max}-\operatorname{Min}} \\
=x \times\left(\frac{2}{\operatorname{Max}-\operatorname{Min}}\right)+\left(\frac{\operatorname{Min}+\operatorname{Max}}{\operatorname{Min}-\operatorname{Max}}\right)
\end{gathered}$$

در رابطه فوق، تنها دو عدد ضریب $$x$$ و عدد ثابت مهم است. به همین دلیل این دو عدد را محاسبه و ذخیره می‌کنیم:

1    def ApplySettings(self):
2        plt.style.use('ggplot')
3        self.Code2Actions = {0: 'Down', 1: 'Up'}
4        self.Action2Change = {0: -1, 1: +1}
5        self.nAction = len(self.Code2Actions)
6        self.Episode = -1
7        nConstantEpisode = self.nEpisode - self.nDecayEpisode
8        epsA = np.linspace(start=self.Epsilon0,
9                           stop=self.Epsilon1,
10                           num=self.nDecayEpisode)
11        epsB = self.Epsilon1 * np.ones(nConstantEpisode)
12        self.Epsilons = np.hstack((epsA, epsB))
13        self.ActionLog = np.zeros((self.nEpisode, self.mStep))
14        self.EpisodeLog = np.zeros(self.nEpisode)
15        self.oActivaiton = 'linear'
16        States = np.linspace(start=self.MinT,
17                             stop=self.MaxT,
18                             num=201)
19        Rewards = self.State2Reward(States)
20        self.MinReward = Rewards.min()
21        self.MaxReward = Rewards.max()
22        self.MinQ = (1 + self.qLR * self.Gamma) * self.MinReward
23        self.MaxQ = (1 + self.qLR * self.Gamma) * self.MaxReward
24        self.Memory = co.deque(maxlen=self.sMemory)
25        self.Counter = 0
26        self.a = 2 / (self.MaxT - self.MinT)
27        self.b = (self.MinT + self.MaxT) / (self.MinT - self.MaxT)

به این ترتیب دو ضریب مورد نیاز برای تغییر مقیاس نیز محاسبه می‌شود.

الگوریتم بهینه‌ساز، تابع هزینه و تابع فعال‌ساز هر سه به شکل «رشته» (String) دریافت شده‌اند. باید این موارد را به شکلی که برای Tensorflow قابل استفاده باشد درآوریم. به این منظور خواهیم داشت:

1        self.Optimizer = self.Optimizer.upper()
2        self.Loss = self.Loss.upper()
3        self.Activation = self.Activation.upper()
4        if self.Optimizer == 'ADAM':
5            self.Optimizer = opt.Adam(learning_rate=self.mLR)
6        elif self.Optimizer == 'SGD':
7            self.Optimizer = opt.SGD(learning_rate=self.mLR)
8        elif self.Optimizer == 'RMSPROP':
9            self.Optimizer = opt.RMSprop(learning_rate=self.mLR)
10        if self.Loss == 'MSE':
11            self.Loss = los.MeanSquaredError()
12        elif self.Loss == 'MAE':
13            self.Loss = los.MeanAbsoluteError()
14        elif self.Loss == 'HUBER':
15            self.Loss = los.Huber(delta=1)
16        if self.Activation == 'RELU':
17            self.Activation = act.relu
18        elif self.Activation == 'ELU':
19            self.Activation = act.elu
20        elif self.Activation == 'TANH':
21            self.Activation = act.tanh

توجه داشته باشید که ممکن است هریک از این موارد به شکل سلیقه‌ای با حروف کوچک، بزرگ یا ترکیبی از آن دو باشد. به همین دلیل در ابتدا ‌آن‌ها را با استفاده از متد upper به حروف بزرگ تبدیل می‌کنیم تا مشکلی از این جهت پیش نیاید. به این ترتیب دو متد مورد استفاده در متد سازنده همگی پیاده‌سازی شدند.

تابع پاداش در پیاده سازی Deep Q Learning

متدی به نام State2Reward نیز در حین کدنویسی استفاده شد که تعریف نشده است. در کد مربوط به Q-Learning به دلیل محدود بودن شرایط، پاداش حالات را به صورت قراردادی مقداردهی می‌کردیم و مشکلی از این جهت نبود. اما در این مسئله به دلیل پیوسته بودن شرایط محیط و دید عامل از محیط (Observation)، نمی‌توان چنین عمل کرد و باید از یک تابع استفاده کنیم. به این منظور از تابع زیر استفاده می‌کنیم:

$$F(x) = 1 -0.002 \times x^2$$

این تابع درجه دوم بوده و در نقطه $$x = 0$$ به بیشینه مقدار خود می‌رسد. توجه داشته باشید که در مسائل دنیای واقعی، ما از این تابع مطلع نیستیم و به دنبال تخمین آن هستیم.

در انتهای برنامه نموداری برای این تابع نیز رسم خواهد شد که به شکل زیر خواهد بود:

این متد را به شکل زیر تعریف می‌کنیم:

1    def State2Reward(self,
2                     State:typ.Union[int, float, np.ndarray]) -> typ.Union[float, np.ndarray]:
3        Reward = 1 - 0.0002 * State ** 2
4        return Reward

توجه داشته باشید که ممکن است این تابع در طول کد به ازای یک State یا چندین State فراخوانی شود. به این دلیل ورودی آن می‌تواند صحیح، غیرصحیح و آرایه باشد. بنابراین باید تمامی این سه حالت در تابع مدنظر گیرد. خوشبختانه به دلیل سازگاری آرایه‌های Numpy با اعمال ریاضی عادی، نیازی به تفکیک ورودی نیست و یک عبارت تمامی حالات را پوشش خواهد داد.

حال می‌توانیم یک متد دیگر رسم کنیم تا نمودار مربوط به State2Reward را در بازه کمترین دما تا بیشترین دما رسم کند:

1    def PlotState2Reward(self):
2        States = np.linspace(start=self.MinT,
3                             stop=self.MaxT,
4                             num=201)
5        Rewards = self.State2Reward(States)
6        plt.plot(States,
7                 Rewards,
8                 ls='-',
9                 lw=1.2,
10                 c='teal')
11        plt.title('Reward-State Plot')
12        plt.xlabel('State')
13        plt.ylabel('Reward')
14        plt.show()

توجه داشته باشید که در اینجا States یک آرایه است و اگر تابع State2Reward به درست پیاده‌سازی نشده بود، باید یک حلقه ایجاد می‌شد. در سطر داخل بازه معین، 201 نقطه با فاصله یکسان انتخاب می‌شود. در سطر بعدی مقادیر Reward برای هریک محاسبه می‌شود. در نهایت با استفاده از matplotlib.pyplot.plot یک نمودار خطی برای آن رسم می‌شود. نمودار اخیر حاصل این متد است که در انتهای برنامه رسم خواهد شد.

رسم نمودار اپسیلون در پیاده سازی Deep Q Learning

می‌توانیم یک نمودار دیگر نیز برای نمایش مقادیر اپسیلون داشته باشیم. به این منظور متد زیر را پیاده‌سازی می‌کنیم:

1    def PlotEpsilons(self):
2        T = np.arange(start=1,
3                      stop=self.nEpisode + 1,
4                      step=1)
5        plt.plot(T,
6                 self.Epsilons,
7                 ls='-',
8                 lw=1.2,
9                 c='teal')
10        plt.title('Epsilon Over Episodes')
11        plt.xlabel('Episode')
12        plt.ylabel('Epsilon')
13        plt.show()

در انتهای کد این متد فراخوانی شده و نمودار زیر را ایجاد خواهد کرد:

مشاهده می‌کنیم که اپسیلون از Epsilon0 شروع شده و در اپیزود 200 به Epsilon1 می‌رسد. از اپیزود 200 تا 300 نیز مقدار آن ثابت می‌ماند. می‌توان از کاهش نمایی (Exponential Decay) نیز استفاده کرد که اغلب این حالت مورد استفاده قرار می‌گیرد.

در نمودار فوق شاید مقادیر به صورت دقیق قابل تشخیص نباشد. برای رفع این مشکل می‌توان از نمودار «نیمه‌لگاریتمی» (Semi-Logarith) استفاده کرد.

تغییر مقیاس شرایط

حال باید متد دیگری نیز پیاده‌سازی کنیم تا با دریافت State در مقیاس اصلی، آن را به مقیاس $$[-1 , +1]$$ ببرد. به این منظور یک متد ایجاد می‌کنیم و در ورودی شرایط را دریافت می‌کنیم:

1    def StateScaler(self,
2                    State:typ.Union[int, float, np.ndarray]) -> typ.Union[float, np.ndarray]:

توجه داشته باشید که این متد نیز می‌تواند تنها یک State دریافت کند یا مجموعه‌ای از Stateها را دریافت کند. حال باید State ورودی را در self.a ضرب کرده و با self.b جمع کنیم و مقدار حاصل را در خروجی برگردانیم:

1    def StateScaler(self,
2                    State:typ.Union[int, float, np.ndarray]) -> typ.Union[float, np.ndarray]:
3        ScaledState = self.a * State + self.b
4        return ScaledState

به این ترتیب این متد نیز کامل خواهد شد. در مواقعی که بخواهیم خروجی شبکه را دریافت کنیم یا شبکه را بر روی مجموعه داده‌ای آموزش دهیم، باید مقادیر ورودی را با استفاده از این متد Scale کنیم.

ایجاد مدل عمیق

حال می‌توانیم توابع مربوط به مدل عمیق را پیاده‌سازی کنیم. اولین تابع CreateModel خواهد بود که وظیفه آن ایجاد مدل عمیق خواهد بود:

1    def CreateModel(self):

با توجه به اینکه قصد داریم یک پرسپترون چندلایه (Multilayer Perceptron یا MLP) ایجاد کنیم، یک مدل Sequential خواهیم داشت. پس یک شی از این کلاس ایجاد می‌کنیم:

1    def CreateModel(self):
2        self.Model = mod.Sequential()

حال یک لایه ورودی به مدل اضافه می‌کنیم. با توجه به اینکه State تنها شامل یک عدد است، ابعاد ورودی به شکل $$(1 , )$$ خواهد بود:

1    def CreateModel(self):
2        self.Model = mod.Sequential()
3        self.Model.add(lay.InputLayer(input_shape=(1, )))

حال می‌توانیم لایه‌های Dense را به مدل اضافه کنیم. تعداد نورون‌های هر لایه در self.nDense موجود است. بنابراین یک حلقه بر روی این لیست ایجاد می‌کنیم و با استفاده از متد add مربوط به کلاس Sequential، هر لایه را اضافه می‌کنیم:

1    def CreateModel(self):
2        self.Model = mod.Sequential()
3        self.Model.add(lay.InputLayer(input_shape=(1, )))
4        for n in self.nDense:
5            BI = ini.RandomUniform(minval=-1,
6                                   maxval=+1,
7                                   seed=self.RandomState)
8            self.Model.add(lay.Dense(units=n,
9                                     activation=self.Activation,
10                                     bias_initializer=BI))

توجه داشته باشید که لایه Dense تنظیمات متعددی دارد که یکی از آن‌ها، bias_initializer است. این ورودی شیوه مقداردهی اولیه (Initialization) بایاس (Bias) را نشان می‌دهد. نورون‌های لایه Dense به صورت پیشفرض بایاس 0 را به عنوان مقدار اولیه به خود می‌گیرند. برای این مسئله می‌توانیم مقادیر اولیه آن‌ها را به صورت یکنواخت در بازه $$[-1 , +1]$$ انتخاب کنیم. توجه داشته باشید که Initializerها نیز یک ورودی seed دارند که این مورد نیز باید با self.RandomState تنظیم شود.

به این ترتیب تا به اینجا لایه ورودی و لایه‌های پنهان اضافه می‌شود. حال باید یک لایه دیگر به عنوان خروجی شبکه ایجاد کنیم و به مدل اضافه کنیم:

1    def CreateModel(self):
2        self.Model = mod.Sequential()
3        self.Model.add(lay.InputLayer(input_shape=(1, )))
4        for n in self.nDense:
5            BI = ini.RandomUniform(minval=-1,
6                                   maxval=+1,
7                                   seed=self.RandomState)
8            self.Model.add(lay.Dense(units=n,
9                                     activation=self.Activation,
10                                     bias_initializer=BI))
11        BI = ini.RandomUniform(minval=-0.1,
12                               maxval=+0.1,
13                               seed=self.RandomState)
14        self.Model.add(lay.Dense(units=self.nAction,
15                                 activation=self.oActivaiton,
16                                 bias_initializer=BI))

توجه داشته باشید که تابع فعال‌سازی لایه خروجی متفاوت با لایه‌های پنهان است. از طرفی با توجه به اینکه مقادیر Return در بازه $$[-1 , +1]$$ است، مقادیر بایاس لایه خروجی به احتمال زیاد در بازه‌ای بسیار محدودتر قرار خواهد گرفت، به همین دلیل مقداردهی اولیه آن‌ها بهتر است در بازه‌ای کوچک‌تر و متمرکزتر بر روی میانگین انجام گیرد.

کامپایل کردن مدل

بنابراین با فراخوانی متد قبلی، مدل عمیق ایجاد خواهد شد. مدل‌های Tensorflow علاوه بر ایجاد، نیاز به Compile شدن نیز دارند. به همین دلیل متد دیگری به نام CompileModel ایجاد می‌کنیم:

1    def CompileModel(self):
2        self.Model.compile(optimizer=self.Optimizer,
3                           loss=self.Loss)

در این متد، الگوریتم بهینه‌ساز و تابع هزینه مورد استفاده در آموزش شبکه را تعیین می‌کند.

گزارش ساختار مدل در پیاده سازی Deep Q Learning

حال می‌توانیم متد دیگری ایجاد کنیم تا گزارشی از شبکه را در خروجی چاپ کند:

1    def ModelSummary(self):
2        self.HL()
3        print('Model Summary:')
4        self.Model.summary()
5        self.HL()

این متد در انتهای کد فراخوانی خواهد شد و نتیجه‌ای به شکل زیر خواهد داشت:

1______________________________________________________________________________________________
2Model Summary:
3Model: "sequential"
4_________________________________________________________________
5 Layer (type)                Output Shape              Param #
6=================================================================
7 dense (Dense)               (None, 256)               512
8
9 dense_1 (Dense)             (None, 256)               65792
10
11 dense_2 (Dense)             (None, 2)                 514
12
13=================================================================
14Total params: 66,818
15Trainable params: 66,818
16Non-trainable params: 0
17_________________________________________________________________

به این ترتیب می‌توان سایز خروجی لایه‌ها و تعداد پارامترهای آن‌ها را مشاهده کرد. این مدل دارای 66818 پارامتر است.

در متد ModelSummary از متد دیگری به نام HL استفاده شد که وظیفه ایجاد یک خط افقی در خروجی را بر عهده دارد. این متد مشابه کد قبل به شکل زیر تعریف خواهد شد:

1    def HL(self, s:str='_', n:int=120):
2        print(s * n)

پیش‌بینی با مدل

حال باید متد دیگری نیز پیاده‌سازی کنیم تا با گرفتن یک State یا آرایه‌ای از Stateها، مقادیر آن را نرمال‌سازی (Normalizing) کرده و وارد شبکه کند. در خروجی مقادیر پیشبینی شده برای Q را خواهیم داشت که باید برگردانده شود:

1    def PredictQ(self,
2                 States:typ.Union[int, float, np.ndarray]) -> np.ndarray:

حال باید مقادیر State را تغییر مقیاس دهیم:

1    def PredictQ(self,
2                 States:typ.Union[int, float, np.ndarray]) -> np.ndarray:
3        ScaledStates = self.StateScaler(States)

با توجه به اینکه در ورودی یک آرایه داشته باشیم یا یک عدد، روش محاسبه مقادیر متفاوت خواهد بود.

بنابراین خواهیم داشت:

1    def PredictQ(self,
2                 States:typ.Union[int, float, np.ndarray]) -> np.ndarray:
3        ScaledStates = self.StateScaler(States)
4        if isinstance(States, np.ndarray):
5            Q = self.Model.predict(ScaledStates.reshape(-1, 1),
6                                   verbose=self.Verbose)
7        else:
8            Q = self.Model.predict(np.array([[ScaledStates]]),
9                                   verbose=self.Verbose)[0]
10        return Q

توجه داشته باشید که اگر مقادیر ورودی یک آرایه باشد، در ابعاد $$(n,)$$ خواهد بود بنابراین مقادیر تغییر مقیاس یافته نیز در ابعاد $$(n,)$$ خواهد بود. شبکه عصبی این ورودی‌ها را در ابعاد $$(n,1)$$ قبول می‌کند بنابراین در شرط اول از عبارت reshape(-1, 1) استفاده می‌کنیم.

در شرط دوم مقادیر State و ScaledState تنها یک عدد خواهد بود، بنابراین باید آن را در لیست قرار داده و تبدیل به آرایه کنیم. در این حالت چون تنها یک خروجی داریم، می‌توانیم خروجی برا برای داده اول برگردانیم. عبارتی که در انتهای کد مربوط به else آمده، به همین منظور است.

فیلم آموزش مبانی یادگیری عمیق یا Deep Learning در فرادرس

کلیک کنید

ذخیره و فراخوانی مدل

آموزش مدل‌های عمیق فرآیندی زمان‌بر است. از طرفی حل مسائل یادگیری تقویتی (Reinforcement Learning) نیز دشواری‌های خود را دارد. بنابراین ذخیره مدل در طول آموزش و فراخوانی مدل‌های آموزش دیده گذشته امری معمول و منطقی است. به همین منظور دو متد دیگر نیز برای ذخیره مدل آموزش دیده (Save) و بارگزاری آن (Load) پیاده‌سازی می‌کنیم:

1    def SaveModel(self, Path:str='Model'):
2        mod.save_model(self.Model, Path)
3        print(col.Fore.MAGENTA + 'Model Saved Successfully.' + col.Fore.RESET)
4    def LoadModel(self, Path:str='Model'):
5        self.Model = mod.load_model(Path)
6        print(col.Fore.MAGENTA + 'Model Loaded Successfully.' + col.Fore.RESET)

به این ترتیب کلاس ایجاد شده، قابلیت استفاده از مدل‌های آموزش دیده قبلی و ذخیره مدل فعلی را نیز خواهد داشت.

ذخیره داده در حافظه

با رخ دادن هر گام در محیط، یک داده جدید ایجاد می‌شود. این داده‌ها باید به ترتیب ایجاد شدن در حافظه عامل ذخیره شوند. به این منظور یک متد ایجاد می‌کنیم که با گرفتن شرایط اولیه، عمل انجام شده، پاداش دریافتی و شرایط نهایی، آن‌ها را به حافظه افزوده و مقدار self.Counter را یک واحد افزایش دهد:

1    def Save2Memory(self,
2                    State:float,
3                    Action:int,
4                    Reward:float,
5                    State2:float):
6        self.Memory.append([State, Action, Reward, State2])
7        self.Counter += 1

به این ترتیب مدل دارای حافظه بوده و تعداد داده‌های جدید را به خاطر خواهد سپرد.

آموزش مدل در پیاده سازی Deep Q Learning

پس از ایجاد و کامپایل کردن مدل، می‌توانیم آن را آموزش دهیم. به این منظور نیز یک متد جدید تعریف می‌کنیم.

فیلم آموزش پیاده سازی بازی مار در پایتون با یادگیری تقویتی عمیق در فرادرس

کلیک کنید

در این متد فرآیند استفاده از حافظه و معادله Q-Learning استفاده خواهد شد:

1    def TrainModel(self):

عامل، به جز 12 اپیزود اول، در سایز اپیزودها، 1024 رکورد (Record) دارد. از بین این 1024 داده، تنها 32 مورد برای آموزش انتخاب خواهد شد. این فرآیند به صورت تصادفی انجام خواهد شد. به این منظور از numpy.random.chice استفاده می‌کنیم:

1    def TrainModel(self):
2        Selecteds = np.random.choice(len(self.Memory),
3                                     size=self.sBatch)

توجه داشته باشید که چون ممکن است طول self.Memory برابر با self.sMemory باشد، باید بیشترین مقدار برابر با طول self.Memory تنظیم شود. در خروجی این کد، 32 داده انتخاب خواهد شد. حال دو آرایه خالی برای ذخیره داده‌ها ایجاد می‌کنیم:

1    def TrainModel(self):
2        Selecteds = np.random.choice(len(self.Memory),
3                                     size=self.sBatch)
4        X0 = np.zeros((self.sBatch, 1))
5        Y = np.zeros((self.sBatch, self.nAction))

یک حلقه بر روی شماره (Index) داده‌های انتخاب شده ایجاد می‌کنیم:

1    def TrainModel(self):
2        Selecteds = np.random.choice(len(self.Memory),
3                                     size=self.sBatch)
4        X0 = np.zeros((self.sBatch, 1))
5        Y = np.zeros((self.sBatch, self.nAction))
6        for i, j in enumerate(Selecteds):

می‌توان داده مورد نظر را به شکل زیر به 4 متغیر تجزیه کرد:

1    def TrainModel(self):
2        Selecteds = np.random.choice(len(self.Memory),
3                                     size=self.sBatch)
4        X0 = np.zeros((self.sBatch, 1))
5        Y = np.zeros((self.sBatch, self.nAction))
6        for i, j in enumerate(Selecteds):
7            State, Action, Reward, State2 = self.Memory[j]

حال مقادیر Q را برای شرایط فعلی و شرایط پس از انجام عمل محاسبه می‌کنیم:

1    def TrainModel(self):
2        Selecteds = np.random.choice(len(self.Memory),
3                                     size=self.sBatch)
4        X0 = np.zeros((self.sBatch, 1))
5        Y = np.zeros((self.sBatch, self.nAction))
6        for i, j in enumerate(Selecteds):
7            State, Action, Reward, State2 = self.Memory[j]
8            oldOutput1 = self.PredictQ(State)
9            oldOutput2 = self.PredictQ(State2)

با توجه به اینکه در شرایط فعلی، تنها یک عمل انتخاب شده و انجام شده است، باید تنها مقدار این خروجی به‌روزرسانی شود:

1    def TrainModel(self):
2        Selecteds = np.random.choice(len(self.Memory),
3                                     size=self.sBatch)
4        X0 = np.zeros((self.sBatch, 1))
5        Y = np.zeros((self.sBatch, self.nAction))
6        for i, j in enumerate(Selecteds):
7            State, Action, Reward, State2 = self.Memory[j]
8            oldOutput1 = self.PredictQ(State)
9            oldOutput2 = self.PredictQ(State2)
10            oldQ1 = oldOutput1[Action]

حال می‌توانیم مقدار Temporal Difference را محاسبه کنیم و سپس مقدار جدید Q مورد نظر را محاسبه کنیم:

1    def TrainModel(self):
2        Selecteds = np.random.choice(len(self.Memory),
3                                     size=self.sBatch)
4        X0 = np.zeros((self.sBatch, 1))
5        Y = np.zeros((self.sBatch, self.nAction))
6        for i, j in enumerate(Selecteds):
7            State, Action, Reward, State2 = self.Memory[j]
8            oldOutput1 = self.PredictQ(State)
9            oldOutput2 = self.PredictQ(State2)
10            oldQ1 = oldOutput1[Action]
11            TD = Reward + self.Gamma * oldOutput2.max() - oldQ1
12            newQ1 = oldQ1 + self.qLR * TD

قبل از آموزش مدل بر روی این داده، برای عمل Action در شرایط State، مقدار Q برابر با oldQ1 بود.

حال انتظار داریم که در همان شرایط، برای همان عمل، مقدار Q برابر با newQ1 شود. بنابراین باید Q مربوط به سایر اعمال ثابت بماند. به این منظور یک کپی (Copy) از خروجی قدیمی می‌گیریم و تنها یک مقدار از آن را به‌روزرسانی می‌کنیم:

1    def TrainModel(self):
2        Selecteds = np.random.choice(len(self.Memory),
3                                     size=self.sBatch)
4        X0 = np.zeros((self.sBatch, 1))
5        Y = np.zeros((self.sBatch, self.nAction))
6        for i, j in enumerate(Selecteds):
7            State, Action, Reward, State2 = self.Memory[j]
8            oldOutput1 = self.PredictQ(State)
9            oldOutput2 = self.PredictQ(State2)
10            oldQ1 = oldOutput1[Action]
11            TD = Reward + self.Gamma * oldOutput2.max() - oldQ1
12            newQ1 = oldQ1 + self.qLR * TD
13            newOutput1 = oldOutput1.copy()
14            newOutput1[Action] = newQ1

به این ترتیب خروجی مد نظر حاصل می‌شود. نکته مهمی که در اینجا وجود دارد، امکان رخ دادن فرآیند خودتنظیمی مثبت و واگرایی مقادیر Q است. برای جلوگیری از این اتفاق، مقادیر این خروجی را بین self.MinQ و self.MaxQ محدود می‌کنیم. به این منظور تابع numpy.clip مناسب است:

1    def TrainModel(self):
2        Selecteds = np.random.choice(len(self.Memory),
3                                     size=self.sBatch)
4        X0 = np.zeros((self.sBatch, 1))
5        Y = np.zeros((self.sBatch, self.nAction))
6        for i, j in enumerate(Selecteds):
7            State, Action, Reward, State2 = self.Memory[j]
8            oldOutput1 = self.PredictQ(State)
9            oldOutput2 = self.PredictQ(State2)
10            oldQ1 = oldOutput1[Action]
11            TD = Reward + self.Gamma * oldOutput2.max() - oldQ1
12            newQ1 = oldQ1 + self.qLR * TD
13            newOutput1 = oldOutput1.copy()
14            newOutput1[Action] = newQ1
15            newOutput1 = np.clip(newOutput1,
16                                 a_min=self.MinQ,
17                                 a_max=self.MaxQ)

به این ترتیب تا به اینجا دو مکانیسم برای جلوگیری از واگرایی مقادیر Q پیاده‌سازی کردیم. حال داده‌های ایجاد شده را ذخیره می‌کنیم. ورودی شبکه State است و خروجی مورد انتظار newOutput1، بنابراین:

1    def TrainModel(self):
2        Selecteds = np.random.choice(len(self.Memory),
3                                     size=self.sBatch)
4        X0 = np.zeros((self.sBatch, 1))
5        Y = np.zeros((self.sBatch, self.nAction))
6        for i, j in enumerate(Selecteds):
7            State, Action, Reward, State2 = self.Memory[j]
8            oldOutput1 = self.PredictQ(State)
9            oldOutput2 = self.PredictQ(State2)
10            oldQ1 = oldOutput1[Action]
11            TD = Reward + self.Gamma * oldOutput2.max() - oldQ1
12            newQ1 = oldQ1 + self.qLR * TD
13            newOutput1 = oldOutput1.copy()
14            newOutput1[Action] = newQ1
15            newOutput1 = np.clip(newOutput1,
16                                 a_min=self.MinQ,
17                                 a_max=self.MaxQ)
18            X0[i] = State
19            Y[i] = newOutput1

پس از اتمام این حلقه، 32 داده خواهیم داشت. تنها موردی که باید انجام شود، تغییر مقیاس X0 است. این تغییر مقیاس را انجام داده و مدل را آموزش می‌دهیم. در نهایت نیز پیامی مبنی بر آموزش مدل نمایش می‌دهیم:

1    def TrainModel(self):
2        Selecteds = np.random.choice(len(self.Memory),
3                                     size=self.sBatch)
4        X0 = np.zeros((self.sBatch, 1))
5        Y = np.zeros((self.sBatch, self.nAction))
6        for i, j in enumerate(Selecteds):
7            State, Action, Reward, State2 = self.Memory[j]
8            oldOutput1 = self.PredictQ(State)
9            oldOutput2 = self.PredictQ(State2)
10            oldQ1 = oldOutput1[Action]
11            TD = Reward + self.Gamma * oldOutput2.max() - oldQ1
12            newQ1 = oldQ1 + self.qLR * TD
13            newOutput1 = oldOutput1.copy()
14            newOutput1[Action] = newQ1
15            newOutput1 = np.clip(newOutput1,
16                                 a_min=self.MinQ,
17                                 a_max=self.MaxQ)
18            X0[i] = State
19            Y[i] = newOutput1
20        X = self.StateScaler(X0)
21        self.Model.fit(X,
22                       Y,
23                       epochs=self.nEpoch,
24                       batch_size=self.sBatch,
25                       verbose=self.Verbose)
26        print(col.Fore.GREEN + 'Model Trained On Batch.' + col.Fore.RESET)

به این ترتیب این متد خواهد توانست تجارب ذخیره شده در حافظه را به صورت تصادفی انتخاب کرده، داده‌ها را از آن استخراج کند و در نهایت مدل را آموزش دهد.

انتخاب عمل

مشابه مطلب قبلی، یک متد نیز برای انتخاب عمل با توجه به سیاست مورد نظر ایجاد می‌کنیم. این متد در ورودی سیاست مورد نظر را دریافت و در خروجی شماره عمل انتخاب شده را برمی‌گرداند:

1    def Decide(self, Policy:str) -> int:
2        if Policy == 'R':
3            return np.random.randint(low=0, high=self.nAction)
4        elif Policy == 'G':
5            q = self.PredictQ(self.State)
6            return np.argmax(q)
7        elif Policy == 'EG':
8            if np.random.rand() < self.Epsilon:
9                return self.Decide('R')
10            else:
11                return self.Decide('G')

به این ترتیب:

• اگر سیاست ورودی R باشد، «سیاست تصادفی» (Random Policy) در پیش گرفته می‌شود.
• اگر سیاست ورودی G باشد، «سیاست حریصانه» (Greedy Policy) در پیش گرفته می‌شود.
• اگر سیاست ورودی EG باشد، سیاست Epsilon-Greedy در پیش گرفته می‌شود.

ریست کردن شرایط

در ابتدای هر اپیزود، شرایط محیط به حالت اولیه برگردانده می‌شود و گام‌های انجام شده پاک می‌شود. برای انجام این عملیات، متدی به شکل زیر تعریف می‌کنیم:

1    def ResetState(self):
2        self.State = np.random.uniform(low=self.MinT,
3                                       high=self.MaxT)
4        self.Step = -1

توجه داشته باشید که در این مسئله، برخلاف مسئله Frozen Lake، محل شروع عامل یکسان نیست و هر بار از محل متفاوتی شروع می‌شود. بنابراین self.State برابر با یک عدد تصادفی در بازه مشخص شده برای دما است.

تعداد گام‌های انجام شده نیز برابر با $$-1$$ تنظیم می‌شود. این تابع در ابتدای هر اپیزود فراخوانی خواهد شد، بنابراین در ابتدای اپیزود برابر با $$0$$ خواهد بود که صحیح است.

اپیزود بعدی

در ابتدای هر اپیزود، باید مشخص کنیم که تغییر اپیزود رخ داده و فرآیندهای مرتبط با آن رخ دهد. این فرآیندها شامل موارد زیر است:

• افزایش Episode
• به‌روزرسانی مقدار Epsilon
• ریست (Reset) کردن شرایط

به این ترتیب می‌توان نوشت:

1    def NextEpisode(self):
2        self.Episode += 1
3        self.Epsilon = self.Epsilons[self.Episode]
4        self.ResetState()

توجه داشته باشید که چون تابع ResetState می‌تواند به تنهایی نیز مورد استفاده قرار گیرد، نمی‌توان آن را حذف کرده و دستورات را در NextEpisode نوشت.

گام بعدی

مشابه متد NextEpisode، متد دیگری نیز با نام NextStep ایجاد می‌کنیم:

1    def NextStep(self):
2        self.Step += 1

ممکن است این ایده به وجود بیاید که می‌توانیم متد‌های کوچک را حذف کنیم و کد مربوط به آن‌ها را در متن اصلی برنامه بنویسیم. با توجه به اینکه این پروژه قابلیت گسترش در قالب‌های مختلف را دارد، در این مقاله سعی بر آن بوده که تمامی متدها وجود داشته باشند. با گسترش کدها، اغلب متدهای آورده شده پیچیده‌تر می‌شوند و فرآیندهای بیشتری را رقم می‌زنند.

انجام عمل

حال به متدی نیاز داریم که با دریافت عمل انتخاب شده، آن را انجام دهد، پاداش حاصل را محاسبه کند، شرایط نهایی را محاسبه کند و شرط اتمام مسئله را نیز بررسی نماید. به این منظور یک متد ایجاد می‌کنیم که در ورودی عمل مورد نظر را به شکل یک عدد صحیح دریافت کند:

1    def Do(self, Action:int) -> tuple:

با توجه به اینکه 3 مورد در خروجی برگردانده خواهد شد، خروجی این متد یک تاپل (Tuple) خواهد بود.

حال باید تغییرات حاصل از عمل انجام شده را اعمال کنیم:

1    def Do(self, Action:int) -> tuple:
2        self.State += self.Action2Change[Action]

برای این منظور، دیکشنری Action2Change استفاده می‌شود که تغییرات دما برای هر عمل را نشان می‌دهد.

حال باید بررسی کنیم تا دما از محدوده مورد نظر ما خارج نشود:

1    def Do(self, Action:int) -> tuple:
2        self.State += self.Action2Change[Action]
3        if self.State > self.MaxT:
4            self.State = self.MaxT
5        elif self.State < self.MinT:
6            self.State = self.MinT

حال می‌توانیم پاداش حاصل را محاسبه کنیم. به این منظور متد State2Reward مناسب خواهد بود:

1    def Do(self, Action:int) -> tuple:
2        self.State += self.Action2Change[Action]
3        if self.State > self.MaxT:
4            self.State = self.MaxT
5        elif self.State < self.MinT:
6            self.State = self.MinT
7        Reward = self.State2Reward(self.State)

حال شرط اتمام اپیزود را بررسی می‌کنیم و در نهایت خروجی‌های متد را برمی‌گردانیم:

1    def Do(self, Action:int) -> tuple:
2        self.State += self.Action2Change[Action]
3        if self.State > self.MaxT:
4            self.State = self.MaxT
5        elif self.State < self.MinT:
6            self.State = self.MinT
7        Reward = self.State2Reward(self.State)
8        if self.Step == self.mStep - 1:
9            Done = True
10        else:
11            Done = False
12        return Reward, self.State, Done

به این ترتیب متد مربوط به شبیه‌سازی تغییرات حاصل از عامل نیز پیاده‌سازی می‌شود.

رسم خروجی مدل در پیاده سازی Deep Q Learning

برای بررسی شیوه عملکرد مدل و مقادیر خروجی آن، می‌توانیم یک متد دیگر ایجاد کنیم که نمودار متد State2Reward را در کنار خروجی‌های مدل رسم کند:

1    def PlotModelPrediction(self, Title:str):

در ابتدا، Stateهایی در بازه مشخص شده ایجاد می‌کنیم:

1    def PlotModelPrediction(self, Title:str):
2        States = np.linspace(start=self.MinT,
3                             stop=self.MaxT,
4                             num=201)

حال خروجی متد State2Reward و شبکه عصبی را دریافت می‌کنیم:

1    def PlotModelPrediction(self, Title:str):
2        States = np.linspace(start=self.MinT,
3                             stop=self.MaxT,
4                             num=201)
5        Rewards = self.State2Reward(States)
6        Predictions = self.PredictQ(States)

حال می‌توانیم یک نمودار برای تابع اصلی و دو نمودار برای خروجی شبکه رسم کنیم:

1    def PlotModelPrediction(self, Title:str):
2        States = np.linspace(start=self.MinT,
3                             stop=self.MaxT,
4                             num=201)
5        Rewards = self.State2Reward(States)
6        Predictions = self.PredictQ(States)
7        plt.plot(States,
8                 Rewards,
9                 ls='-',
10                 lw=1.4,
11                 c='k',
12                 label='Reward(State)')
13        plt.plot(States,
14                 Predictions[:, 0],
15                 ls='-',
16                 lw=1.2,
17                 c='r',
18                 label='Q(State, Down)')
19        plt.plot(States,
20                 Predictions[:, 1],
21                 ls='-',
22                 lw=1.2,
23                 c='b',
24                 label='Q(State, Up)')
25        plt.title(Title)
26        plt.xlabel('State')
27        plt.ylabel('Reward / Q')
28        plt.legend()
29        plt.show()