ד.12 — בנייה ואימון של DQN באיקס עיגול

המצגת: DQN באיקס עיגול · השיעור באתר הקורס · קוד: מאגר Tic_Tac_Toe_DQN · ReplayBuffer.py · DQN.py · DQN_Agent.py · DQN_Trainer.py · Tester.py

בפרק הקודם בנינו את DQN בשלושה שלבים: רשת נוירונים במקום טבלת Q, מאגר מעברים שממנו דוגמים אצוות, ורשת מטרה שמייצבת את היעדים. בפרק הזה נהפוך את הפסאודו־קוד לקוד עובד. השאלה המעשית היא איך כל אחד מהרכיבים נראה ב־PyTorch, ואיך מחברים אותם ללולאת אימון אחת בלי לערבב בין המשחק החי לבין הדגימות מהמאגר.

ניישם את DQN על איקס עיגול: הסוכן X ישחק מול יריב אקראי, ישמור את המעברים שחווה ויאמן רשת שמעריכה את כדאיות המהלכים. המשחק עצמו, הלוח, הסביבה, היריב והגרפיקה, הוא אותו משחק ממאגר Tic_Tac_Toe_2 של פרק ד.8. החדש הוא ארבעה קבצים: מאגר המעברים, הרשת, הסוכן והמאמן. נעבור עליהם בסדר שבו הם נבנים, קובץ אחר קובץ, ובסוף נאמן, נבדוק ונשחק.

כדאי לשים לב כמה מהמבנה כבר מוכר. לולאת המשחק, שני הצעדים של X ושל היריב ובדיקת הסיום הם אותם רכיבים מהפרקים הטבלאיים; מחלקת הרשת, האופטימייזר, פונקציית ההפסד וצמד הקריאות backward ו־step הם בדיוק מה שלמדנו בחלק ג. החדש הוא רק החיבור: הנתונים לאימון מגיעים מהמשחק, והיעדים מחושבים לפי נוסחת DQN.

המשחק הנתון — מאגר Tic_Tac_Toe_DQN

במאגר Tic_Tac_Toe_DQN בגיטהב נמצא הפרויקט המלא, בענף main. זו הדגמה ולא תרגיל: כל הקוד נתון, ותפקידנו להבין אותו ולהריץ אותו. אלה הקבצים:

מלוח לטנסור. הרשת מקבלת מספרים בלבד, ולכן State קיבל שתי מתודות חדשות. toTensor פורשת את הלוח 3×3 לטנסור של תשעה מספרים (1 ל־X, ‏−1 ל־O ו־0 לתא ריק), ו־tensorToState עושה את ההפך ומחזירה אובייקט State מתוך טנסור, כדי שאפשר יהיה לשאול את הסביבה אילו פעולות חוקיות בלוח שנדגם מהמאגר:

def toTensor (self, device = torch.device('cpu')):
    array = self.board.reshape(-1)
    tensor = torch.tensor(array, dtype=torch.float32, device=device)
    return tensor

[staticmethod]
def tensorToState (state_tensor, player = 1):
    board = state_tensor.reshape([3,3]).cpu().numpy()
    return State(board, player)

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

המחלקה ReplayBuffer

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

from collections import deque
import random
import torch
import numpy as np
from State import State


class ReplayBuffer:
    def __init__(self, capacity= 10000) -> None:
        self.buffer = deque(maxlen=capacity)

    def push (self, state : State, action, reward, next_state : State, done):
        self.buffer.append((state.toTensor(), torch.from_numpy(np.array(action)), 
                            torch.tensor(reward), next_state.toTensor(), torch.tensor(done)))

    def push_tensors (self, state_tensor, action_tensor, reward_tensor, next_state_tensor, done):
        self.buffer.append((state_tensor, action_tensor, reward_tensor, next_state_tensor, done))
            
    def sample (self, batch_size):
        if (batch_size > self.__len__()):
            batch_size = self.__len__()
        state_tensors, action_tensor, reward_tensors, \
                   next_state_tensors, dones = zip(*random.sample(self.buffer, batch_size))
        states = torch.vstack(state_tensors)
        actions = torch.vstack(action_tensor)
        rewards = torch.vstack(reward_tensors)
        next_states = torch.vstack(next_state_tensors)
        done_tensor = torch.tensor(dones).long().reshape(-1,1)
        return states, actions, rewards, next_states, done_tensor

    def __len__(self):
        return len(self.buffer)

כל איבר בתור הוא tuple של חמישה שדות: (state, action, reward, next_state, done). חשוב לשמור גם את הפעולה, משום שהרשת שלנו מקבלת מצב ופעולה כדי לחשב Q. done הוא ערך בוליאני: האם המצב הבא סופי. המתודה push ממירה כל שדה לטנסור כבר בזמן השמירה, כך שבזמן הדגימה לא נצטרך המרות; push_tensors מאפשרת לשמור שדות שכבר הם טנסורים.

sample בוחרת batch_size רשומות אקראיות מהתור ומסדרת אותן בטנסורים של עמודות, בצורה המתאימה לאימון רשת. zip(*...) מפרידה את הרשומות לפי שדות: כל המצבים יחד, כל הפעולות יחד וכן הלאה, ו־vstack מערים את הטנסורים של הדוגמאות זה על זה לטנסור אחד. דגלי הסיום הופכים לעמודה של 0 ו־1 (long), כי בהמשך נכפיל בהם. אם מבקשים אצווה גדולה מהמאגר, sample מקטינה אותה בשקט לגודל המאגר. עבור אצווה בגודל B יתקבלו הצורות הבאות:

המחלקה DQN — רשת הנוירונים

הרכיב השני הוא הרשת עצמה, שמחליפה את טבלת Q. היא מקבלת מצב ופעולה ומחזירה מספר אחד: אומדן ל־Q של אותו זוג. תשעת תאי הלוח ושתי קואורדינטות הפעולה יוצרים קלט בן 11 מספרים. שתי שכבות חבויות של 128 ו־64 יחידות, ופלט יחיד:

import math
import random
from typing import Any
import torch
import torch.nn as nn
import torch.nn.functional as F

# Parameters
input_size = 11 # state: board = 3 * 3 + action 1 * 2
layer1 = 128
layer2 = 64
output_size = 1 # Q(s,a)
gamma = 0.99 
MSELoss = nn.MSELoss()

class DQN (nn.Module):
    def __init__(self) -> None:
        super().__init__()
        if torch.cuda.is_available:
            self.device = torch.device('cpu') # 'cuda'
        else:
            self.device = torch.device('cpu')
        
        self.linear1 = nn.Linear(input_size, layer1)
        self.linear2 = nn.Linear(layer1, layer2)
        self.output = nn.Linear(layer2, output_size)
        
    def forward (self, x):
        x = self.linear1(x)
        x = F.relu(x)
        x = self.linear2(x)
        x = F.relu(x)
        x = self.output(x)
        return x
    
    def load_params(self, path):
        self.load_state_dict(torch.load(path))

    def save_params(self, path):
        torch.save(self.state_dict(), path)

    def copy (self):
        new_DQN = DQN()
        new_DQN.load_state_dict(self.state_dict())
        return new_DQN
    
    def loss (self, Q_value, rewards, Q_next_Values, Dones ):
        Q_new = rewards + gamma * Q_next_Values * (1- Dones)
        return MSELoss(Q_value, Q_new)

    def __call__(self, states, actions):
        state_action = torch.cat((states,actions), dim=1)
        return self.forward(state_action)

זהו אותו מבנה של רשת Fully Connected שבנינו בחלק ג, רק עם קלט ופלט אחרים. אחרי כל שכבה חבויה מופעלת ReLU; בשכבת הפלט אין אקטיבציה, כי ערך Q אינו הסתברות ויכול להיות שלילי, למשל כשהמהלך מוביל להפסד. ההתקן נקבע ל־CPU בשני הענפים של התנאי, ולכן כל הדוגמה רצה על המעבד; מי שרוצה GPU מחליף את המחרוזת ל־'cuda'.

ארבע המתודות שאחרי forward הן מה שהופך את הרשת לרשת של DQN:

• __call__ מקבל את המצבים ואת הפעולות כשני טנסורים ומחבר אותם לאורך ממד העמודות: B×9 ו־B×2 הופכים ל־B×11, והפלט הוא B×1. כך קוראים לרשת בצורה Q(states, actions) בכל מקום בקוד.
• loss היא נוסחת היעד של DQN בשורה אחת: Q_new הוא היעד y, והמכפלה ב־(1 - Dones) מאפסת את חלק העתיד במעברים סופיים, כך שהיעד שם הוא התגמול בלבד. Q_next_Values הם ערכי ההמשך שרשת המטרה חישבה, ו־MSELoss משווה בין התחזית Q_value ליעד. שימו לב שהפונקציה אינה מחשבת את ערכי ההמשך בעצמה; היא מקבלת אותם מוכנים מהמאמן.
• copy יוצרת רשת חדשה באותו מבנה ומעתיקה אליה את המשקלים דרך state_dict, מילון של כל משקלי הרשת. כך ניצור את רשת המטרה. השמה כגון Q_hat = Q לא הייתה יוצרת רשת עצמאית אלא שם נוסף לאותו אובייקט.
• save_params ו־load_params שומרות וטוענות את המשקלים לקובץ, בשביל הבודק ובשביל המשחק בחלון.

המחלקה DQN_Agent — הסוכן

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

import math
import random
import torch
import torch.nn as nn
import numpy as np
from DQN import DQN
from State import State

# epsilon Greedy
epsilon_start = 1
epsilon_final = 0.01
epsiln_decay = 5000

# epochs = 1000
# batch_size = 64
gamma = 0.99 
MSELoss = nn.MSELoss()

class DQN_Agent:
    def __init__(self, player = 1, parametes_path = None, train = True, env= None) -> None:
        self.DQN = DQN()
        if parametes_path:
            self.DQN.load_params(parametes_path)
        self.train(train)
        # self.train = train
        self.player = player
        self.env = env

    def train (self, train):
          self.train = train
          if train:
              self.DQN.train()
          else:
              self.DQN.eval()

    def get_action (self, state: State, epoch = 0, events= None, train = True):
        epsilon = self.epsilon_greedy(epoch)
        rnd = random.random()
        actions = self.env.legal_actions(state)
        if train and rnd < epsilon:
            return random.choice(actions)
        
        state_tensor = state.toTensor()
        action_np = np.array(actions)
        action_tensor = torch.from_numpy(action_np)
        expand_state_tensor = state_tensor.unsqueeze(0).repeat((len(action_tensor),1))
        # state_action = torch.cat((expand_state_tensor, action_tensor ), dim=1)
        with torch.no_grad():
            Q_values = self.DQN(expand_state_tensor, action_tensor)
        max_index = torch.argmax(Q_values)
        return actions[max_index]

    def get_actions (self, states, dones):
        actions = []
        for i, state in enumerate(states):
            if dones[i].item():
                actions.append((0,0))
            else:
                actions.append(self.get_action(State.tensorToState(state), train=False)) # book: greedy, as in the slides
        return torch.tensor(actions)

    def epsilon_greedy(self,epoch, start = epsilon_start, final=epsilon_final, decay=epsiln_decay):
        res = final + (start - final) * math.exp(-1 * epoch/decay)
        return res
    
    def save_param (self, path):
        self.DQN.save_params(path)

    def load_params (self, path):
        self.DQN.load_params(path)

    def __call__(self, events= None, state=None, train=True, env=None):
        return self.get_action(state=state, train=train)

ε-greedy עם דעיכה. epsilon_greedy היא הדעיכה המעריכית מ־פרק מונטה קרלו: ε מתחיל ב־1, ועם decay=5000 הוא יורד לכ־0.14 אחרי 10,000 משחקים ומתקרב ל־0.01 בהמשך. get_action מגריל מספר, ואם הוא קטן מ־ε ואנחנו באימון (train), מחזירה פעולה חוקית אקראית.

הבחירה החמדנית. אחרת, הסוכן שואל את הרשת על כל הפעולות החוקיות בבת אחת: unsqueeze ו־repeat משכפלים את הלוח לשורה אחת לכל פעולה, כך שאם יש ארבע משבצות פנויות הרשת מקבלת ארבע שורות ומחזירה ארבעה ערכי Q. argmax נותנת את האינדקס של הערך הגדול ביותר, ובעזרתו מחזירים פעולה מתוך רשימת הפעולות החוקיות. no_grad מונע בניית גרף גרדיאנטים: כאן רק שואלים את הרשת, לא מאמנים אותה. המתודה train מעבירה את הרשת למצב אימון או הערכה; שימו לב שההשמה self.train = train מחליפה את המתודה בערך בוליאני, וזה עובד רק משום שקוראים לה פעם אחת, באתחול.

get_actions — פעולה חמדנית לכל לוח באצווה. המאמן יצטרך, לכל מצב הבא שנדגם מהמאגר, את הפעולה הטובה ביותר בו. get_actions עוברת על האצווה, ממירה כל שורה בחזרה ל־State בעזרת tensorToState, ומפעילה get_action בלי חקירה. ללוח סופי אין פעולה חוקית, ולכן היא מחזירה בו פעולה מדומה (0,0); ערך ההמשך של מעבר כזה יתאפס ממילא ב־loss דרך (1 - Dones).

DQN_Trainer — לולאת האימון

עכשיו כל הרכיבים מוכנים, ונחבר אותם ללולאה אחת לפי הפסאודו־קוד המלא של DQN מהפרק הקודם: בכל צעד של המשחק החי בוחרים פעולה, שומרים את המעבר, ואם המאגר מלא מספיק מאמנים על אצווה; אחת ל־C משחקים מסנכרנים את רשת המטרה. תחילה הפרמטרים והאתחול:

from DQN import DQN
from DQN_Agent import DQN_Agent
from Random_Agent import Random_Agent
from Random_Agent_Advanced import Random_Agent_Advanced
from TicTacToe import TicTacToe
from ReplayBuffer import ReplayBuffer
from State import State

import torch 

epochs = 50000
C = 300
batch = 64
learning_rate = 0.1
path = "Data\DQN_PARAM_Advanced.pth"

def main ():
    env = TicTacToe()
    player1 = DQN_Agent(1, env=env)
    # player2 = Random_Agent(-1, env=env)
    player2 = Random_Agent_Advanced(player=-1, env=env)
    replay = ReplayBuffer()
    Q = player1.DQN
    Q_hat :DQN = Q.copy()
    Q_hat.train = False
    player1_hat = DQN_Agent(1, env=env, train=False) # book: as in DQN_Trainer_2
    player1_hat.DQN = Q_hat
    optim = torch.optim.SGD(Q.parameters(), lr=learning_rate)

epochs הוא מספר המשחקים: המונח "אפוק" מקבל כאן משמעות שונה מזו שבחלק ג, לא מעבר על כל מערך הנתונים אלא משחק אחד. C הוא מספר המשחקים בין סנכרוני רשת המטרה, ו־batch הוא מספר המעברים בכל עדכון. learning_rate הוא קצב הלמידה של SGD, ו־γ=0.99 נקבע בקובץ הרשת: הסוכן כמעט אינו מקטין את משקלם של תגמולים רחוקים, וזה הגיוני במשחק שבו כל התגמול מגיע בסוף.

הרשת הראשית Q היא הרשת של הסוכן player1, ורשת המטרה Q_hat נוצרת ממנה ב־copy. האופטימייזר מקבל רק את פרמטרי Q; רשת המטרה לא תלמד דרך גרדיאנטים. היריב הוא Random_Agent_Advanced, היריב המשופר שהכרנו בפרק ד.8; השורה שבהערה מאפשרת לאמן מול היריב האקראי הבסיסי. הסוכן player1_hat הוא סוכן שני שמחזיק את רשת המטרה, בלי חקירה; נשתמש בו כדי לבחור את פעולת ההמשך בעזרת רשת המטרה, ומיד נסביר למה.

    for epoch in range(epochs):
        print (epoch, end="\r")
        state = State()
        while not env.end_of_game(state):
            action = player1.get_action(state, epoch=epoch)
            after_state, reward = env.next_state(state, action)
            if env.end_of_game(after_state):
                replay.push(state, action, reward, after_state, env.end_of_game(after_state))
                break
            after_action = player2.get_action(state=after_state)
            next_state, reward = env.next_state(after_state, after_action)
            replay.push(state, action, reward, next_state, env.end_of_game(next_state))
            state = next_state

            if epoch < 5000:
                continue
            states, actions, rewards, next_states, dones = replay.sample(batch)
            Q_values = Q(states, actions)
            next_actions = player1_hat.get_actions(next_states, dones) # book: target selects
            with torch.no_grad():
                Q_hat_Values = Q_hat(next_states, next_actions)
            
            loss = Q.loss(Q_values, rewards, Q_hat_Values, dones)
            loss.backward()
            optim.step()
            optim.zero_grad()
        if epoch % C == 0:
            Q_hat.load_state_dict(Q.state_dict())

    player1.save_param(path)

if __name__ == '__main__':
    main()

צעד במשחק החי. כמו בפרקים הטבלאיים, צעד למידה אחד הוא מהלך של X ותגובת היריב. הסוכן בוחר פעולה ב־ε-greedy לפי מספר המשחק, והסביבה מחזירה את after_state והתגמול. אם המהלך סיים את המשחק, שומרים את המעבר עם done אמת ויוצאים מהלולאה. אחרת היריב משחק, ומתקבל next_state, המצב שבו X שוב בתור; שומרים את המעבר השלם (state, action, reward, next_state, done) וממשיכים את המשחק החי ממנו. שימו לב ש־state = next_state נעשה לפני האימון: מכאן והלאה states ו־next_states הם אצוות מהמאגר, ולא המצב החי, בדיוק ההבחנה מהפרק הקודם.

חימום. ב־5,000 המשחקים הראשונים continue מדלג על האימון: הסוכן רק משחק וממלא את המאגר. במצגת התנאי הוא epoch < batch, כלומר ממתינים רק עד שיש מספיק מעברים לאצווה אחת; שני התנאים ממלאים את אותו תפקיד.

צעד האימון. דוגמים אצווה, ומחשבים ברשת הראשית את התחזית Q_values לזוגות שבוצעו. את ערכי ההמשך מחשבים בשני שלבים: get_actions בוחרת לכל מצב הבא את הפעולה החמדנית, ו־Q_hat נותנת את הערך שלה, בתוך no_grad, כי דרך היעד לא מעבירים גרדיאנט. Q.loss בונה מהם את היעד ומחשבת MSE, ואז backward, ‏step ואיפוס הגרדיאנטים. סדר האיפוס (אחרי step במקום לפני backward) שקול, כל עוד מאפסים פעם אחת בכל עדכון.

סנכרון. if epoch % C == 0 נמצא מחוץ ללולאת הצעדים, ולכן ההעתקה של המשקלים לרשת המטרה מתבצעת פעם אחת בכל C משחקים, כמו בפסאודו־קוד. בסיום נשמרים משקלי הרשת הראשית לקובץ.

מי בוחר את פעולת ההמשך. זו הנקודה היחידה שבה קוד הספר שונה מהקובץ שבענף main, ושתי השורות המסומנות # book: הן השינוי. במאגר get_actions נקראת דרך player1, כלומר הרשת הראשית בוחרת את פעולת ההמשך, ובענף main היא אף בוחרת אותה ב־ε-greedy. בספר, כמו ב־DQN_Trainer_2.py שבאותו מאגר, הבחירה נעשית דרך player1_hat, שמחזיק את רשת המטרה, ובלי חקירה: רשת המטרה גם בוחרת את הפעולה וגם מעריכה אותה. זה בדיוק ה־max Q(S′,a;w⁻) מנוסחת DQN של הפרק הקודם. הגרסה שבמצגת, שבה הראשית בוחרת בחמדנות והמטרה מעריכה, היא הפיצול של DDQN, ונחזור אליה בפרק הבא: כל מה שצריך כדי לעבור בין השתיים הוא להחליף player1_hat ב־player1 בשורה אחת.

Tester — בדיקת המודל

האימון נמשך זמן, ואיננו רוצים לחזור עליו בכל פעם שרוצים לשחק נגד הסוכן. כל מה שהסוכן למד נמצא במשקלי הרשת, ולכן די לטעון אותם. הבודק יוצר סוכן עם train=False, טוען את המשקלים ומשחק 1,000 משחקים מול היריב, בלי גרפיקה ובלי אימון:

from TicTacToe import TicTacToe
from State import State
from Human_Agent import Human_Agent
from Random_Agent import Random_Agent
from Random_Agent_Advanced import Random_Agent_Advanced
from DQN_Agent import DQN_Agent

PATH = "Data\DQN_PARAM_Advanced.pth" # book: the file DQN_Trainer.py saves
# PATH=None
env = TicTacToe(State())
player1 = DQN_Agent(1, env=env, parametes_path=PATH, train=False)
# player1 = Random_Agent(1, env,graphics=None)
player2 = Random_Agent_Advanced(-1, env,graphics=None)
# player2 = Random_Agent(-1, env,graphics=None)
num = 1000

def main ():

    x_win = 0
    o_win = 0
    tie = 0
        
    for n in range(num):
        state = State()
        player = player1
        while not env.end_of_game(state):
            action = player.get_action(state=state, train=False)
            state, _ = env.next_state(state,action)
            player = switch_players(player)
        if state.end_of_game == 1:
            x_win +=1
        elif state.end_of_game == -1:
            o_win += 1
        else:
            tie +=1
        state.reset()    
        print(n, end = "\r")
    print()
    print(x_win, o_win, tie) 

def switch_players(player):
    if player == player1:
        return player2
    else:
        return player1

if __name__ == '__main__':
    main()

זו אותה לולאת בדיקה מ־Tester.py של פרק ד.8: השחקנים מתחלפים אחרי כל מהלך, ובסוף כל משחק סופרים לפי end_of_game: 1 לניצחון X, ‏−1 לניצחון O ו־2 לתיקו. get_action נקראת עם train=False, ולכן אין חקירה. בקובץ שבמאגר PATH מצביע על קובץ משקלים ששמו שונה מזה שהמאמן יוצר; בקוד המובא בספר הוא הותאם לקובץ ש־DQN_Trainer.py שומר.

הרצה ובדיקה

שכפלו את מאגר Tic_Tac_Toe_DQN מגיטהב, ובסביבת Python עם PyTorch ו־pygame הפעילו תחילה DQN_Trainer.py ולאחריו Tester.py. הראשון מריץ את מונה המשחקים בטרמינל ושומר בסיום את המשקלים ב־Data/; השני טוען אותם ומדפיס את ספירת התוצאות: ניצחונות X, ניצחונות O ותיקו.

@@RUN_RESULTS@@

אחרי האימון הריצו את Game.py: הוא טוען את המשקלים השמורים (עדכנו בו את PATH לקובץ שנוצר), הסוכן משחק X ואתם O בלחיצת עכבר.

כאן יש הבדל חשוב מחלק ג. שם, הפסד נמוך על נתוני הבדיקה היה מדד טוב לאיכות המודל, כי היעדים היו תוויות אמת. ב־DQN ה־loss מודד התאמה ליעדים שנבנו מהדגימות ומהרשת עצמה, ורשת יכולה להתאים היטב ליעדים שגויים שהיא בעצמה ייצרה. לכן מודדים את הסוכן בספירת תוצאות מול היריב, ולא ב־loss. גם תוצאה טובה מול יריב אקראי אינה מוכיחה משחק מיטבי מול כל יריב: הסוכן למד לשחק מול היריב שהתאמן מולו.

בפרק הבא נבחן את השינוי הקטן שמציע DDQN לחישוב היעד.