11-DQN באיקס עיגול | למידת מכונה ולמידת חיזוק

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

המצגת: DQN באיקס עיגול (עותק מקומי) · קוד: ReplayBuffer · הרשת · הסוכן · המאמן · הבודק

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

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

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

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

הממשק שהאימון מקבל מהמשחק

נשתמש באותו ממשק איקס עיגול. המצב שומר לוח ותור, והפעולה היא (row, col). הפונקציה sample_step מבצעת את פעולת X ואת תגובת היריב, אם נדרשת, ומחזירה (next_state, reward, done).

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

ReplayBuffer — שומרים חמישה שדות ודוגמים אצווה

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

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

from collections import deque
import random
import torch

class ReplayBuffer:
    def __init__(self, capacity=10000, seed=2):
        self.buffer = deque(maxlen=capacity)
        self.rng = random.Random(seed)

    def push(self, state, action, reward, next_state, done):
        self.buffer.append((
            torch.tensor(state.board, dtype=torch.float32).reshape(1, 9),
            torch.tensor(action, dtype=torch.float32).reshape(1, 2),
            torch.tensor([[reward]], dtype=torch.float32),
            torch.tensor(next_state.board, dtype=torch.float32).reshape(1, 9),
            torch.tensor([[done]], dtype=torch.bool)))

    def sample(self, batch_size):
        batch = self.rng.sample(list(self.buffer), batch_size)
        states, actions, rewards, next_states, dones = zip(*batch)
        return (torch.vstack(states), torch.vstack(actions),
                torch.vstack(rewards), torch.vstack(next_states),
                torch.vstack(dones))

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

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

טנסור	צורה	תוכן
`states`	B×9	לוחות לפני פעולות X
`actions`	B×2	קואורדינטות הפעולות
`rewards`	B×1	תגמולים מנקודת מבט X
`next_states`	B×9	לוחות אחרי צעד הלמידה
`dones`	B×1	האם כל מעבר הסתיים

המאמן יקרא ל־sample רק כאשר יש לפחות B רשומות. לכן אין צורך לצמצם בשקט את גודל האצווה, ואין ניסיון לפרק מאגר ריק. קלט הרשת והתגמולים הם float32; דגלי הסיום הם בוליאניים.

רשת DQN — מצב ופעולה נכנסים, Q יוצא

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

11 → 128 → 64 → 1

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

import torch
from torch import nn

class DQN(nn.Module):
    def __init__(self):
        super().__init__()
        self.linear1 = nn.Linear(11, 128)
        self.linear2 = nn.Linear(128, 64)
        self.output = nn.Linear(64, 1)

    def forward(self, states, actions):
        x = torch.cat((states, actions), dim=1)
        x = torch.relu(self.linear1(x))
        x = torch.relu(self.linear2(x))
        return self.output(x)

חיבור הטנסורים מתבצע לאורך ממד העמודות: B×9 ו־B×2 הופכים ל־B×11. הפלט הוא B×1. נכתוב Q(states, actions) כדי להפעיל את הרשת דרך המעטפת של nn.Module; החיבור הוגדר ב־forward, ואין צורך להחליף את __call__.

DQNAgent — פעולה חוקרת או חמדנית

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

class DQNAgent:
    def __init__(self, env, model, seed=0, training=True):
        self.env = env
        self.model = model
        self.rng = random.Random(seed)
        self.set_training(training)

    def set_training(self, enabled):
        self.training = enabled
        self.model.train(enabled)

    def get_action(self, state, epoch=0):
        actions = self.env.get_actions(state)
        if not actions:
            raise ValueError('No action in a terminal state')
        epsilon = epsilon_at(epoch, decay=5000) if self.training else 0.0
        if self.rng.random() < epsilon:
            return self.rng.choice(actions)
        states = torch.tensor(state.board, dtype=torch.float32)
        states = states.reshape(1, 9).repeat(len(actions), 1)
        action_tensor = torch.tensor(actions, dtype=torch.float32)
        with torch.no_grad():
            values = self.model(states, action_tensor).flatten()
        return actions[values.argmax().item()]

epsilon_at היא פונקציית הדעיכה המעריכית מ־פרק מונטה קרלו. כאן נשתמש ב־decay=5000, לפי קובץ הסוכן של DQN. בתחילת האימון ε=1 ובהמשך הוא מתקרב ל־0.01. בזמן בדיקה ε=0 בלי תלות במספר האפיזודה.

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

no_grad מונע בניית גרף גרדיאנטים בזמן בחירת פעולה: כאן רק שואלים את הרשת, לא מאמנים אותה, ולכן אין טעם לשמור מידע לחישוב נגזרות. train(False) מעביר את הרשת למצב הערכה; הוא אינו תחליף ל־no_grad. שני המנגנונים משמשים למטרות שונות: הראשון חוסך חישוב וזיכרון של גרדיאנטים, והשני מסמן לרשת שהיא במצב בדיקה, מה שמשפיע על סוגי שכבות שמתנהגים אחרת באימון ובבדיקה. ברשת הפשוטה שלנו אין שכבות כאלה, אבל נכון להרגיל את עצמנו לקרוא לו.

ערכי ההמשך של אצווה

כדי לבנות את היעד y = R + γ·max Q(S′,a′;w⁻) לכל דוגמה באצווה, צריך את החלק "max Q(S′,a′;w⁻)": הערך הטוב ביותר שאפשר להשיג במצב הבא, לפי רשת המטרה. לפני לולאת האימון נכתוב פונקציה לחישוב ערכי ההמשך. כל שורה ב־next_states מתארת לוח אחר. במעבר סופי נשאיר 0; במעבר שאינו סופי נמצא את המשבצות הפנויות וניקח את הערך המרבי של רשת המטרה בלבד.

def next_q_values(target, next_states, dones):
    values = torch.zeros((len(next_states), 1), dtype=torch.float32)
    with torch.no_grad():
        for i, board in enumerate(next_states):
            if dones[i].item():
                continue
            actions = (board.reshape(3, 3) == 0).nonzero().float()
            states = board.reshape(1, 9).repeat(len(actions), 1)
            values[i, 0] = target(states, actions).max()
    return values

הביטוי board.reshape(3, 3) == 0 יוצר לוח בוליאני שבו אמת מסמנת תא ריק, ו־nonzero מחזירה כאן את קואורדינטות התאים הריקים, כלומר את הפעולות החוקיות. איננו צריכים ליצור מחדש את כל אובייקט המשחק כדי לקבל אותן. אחר כך, כמו בסוכן, משכפלים את הלוח לשורה אחת לכל פעולה, ורשת המטרה מחזירה ערך לכל שורה. בדיקת done מקדימה את החיפוש, כך שלוח סופי אינו דורש פעולה מדומה או חישוב מקסימום על רשימה ריקה.

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

הכנת המאמן ורשת המטרה

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

EPOCHS = 30000
C = 1000
BATCH_SIZE = 64
LEARNING_RATE = 0.1
GAMMA = 0.99

EPOCHS הוא מספר המשחקים, C הוא מספר האפיזודות שהסתיימו בין סנכרוני המטרה, ו־BATCH_SIZE הוא מספר המעברים בכל עדכון. LEARNING_RATE הוא קצב הלמידה של האופטימייזר, ו־GAMMA הוא מקדם ההיוון; ערך 0.99 אומר שהסוכן כמעט אינו מקטין את משקלם של תגמולים רחוקים, וזה הגיוני במשחק שבו כל התגמול מגיע בסוף. נשתמש ב־SGD כדי לעדכן את הרשת הראשית. שימו לב שהמונח "אפוק" מקבל כאן משמעות שונה מזו שבחלק ג: לא מעבר על כל מערך הנתונים, אלא משחק אחד.

באתחול ניצור שני אובייקטים נפרדים ונעתיק את משקלי הראשית למטרה. השמה כגון target = Q לא הייתה יוצרת רשת עצמאית, אלא שם נוסף לאותו אובייקט, וכל עדכון של Q היה משנה גם את "המטרה". state_dict הוא מילון של כל המשקלים ברשת, ו־load_state_dict מעתיק אותם לרשת אחרת באותו מבנה. האופטימייזר יקבל רק את פרמטרי Q; רשת המטרה לא תלמד דרך גרדיאנטים.

Q = DQN()
target = DQN()
target.load_state_dict(Q.state_dict())
target.eval()
target.requires_grad_(False)
optimizer = torch.optim.SGD(Q.parameters(), lr=LEARNING_RATE)
loss_function = torch.nn.MSELoss()

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

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

def train(epochs=EPOCHS, seed=0):
    torch.manual_seed(seed)
    env = TicTacToe()
    Q = DQN()
    agent = DQNAgent(env, Q, seed=seed)
    target = DQN()
    target.load_state_dict(Q.state_dict())
    target.eval()
    target.requires_grad_(False)
    replay = ReplayBuffer(seed=seed + 2)
    opponent_rng = random.Random(seed + 1)
    optimizer = torch.optim.SGD(Q.parameters(), lr=LEARNING_RATE)
    loss_function = torch.nn.MSELoss()
    losses = []
    for epoch in range(epochs):
        state = State()
        while not env.end_of_game(state):
            action = agent.get_action(state, epoch)
            next_state, reward, done = sample_step(
                env, state, action, opponent_rng)
            replay.push(state, action, reward, next_state, done)
            if len(replay) >= BATCH_SIZE:
                states, actions, rewards, next_states, dones = replay.sample(BATCH_SIZE)
                predictions = Q(states, actions)
                future_values = next_q_values(target, next_states, dones)
                targets = rewards + GAMMA * future_values * (~dones).float()
                loss = loss_function(predictions, targets)
                optimizer.zero_grad()
                loss.backward()
                optimizer.step()
                losses.append(loss.item())
            state = next_state
        if (epoch + 1) % C == 0:
            target.load_state_dict(Q.state_dict())
    return env, agent, losses

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

המסכה (~dones).float() היא 0 במעברים סופיים ו־1 באחרים. כך היעד במעבר סופי נשאר בדיוק התגמול. כל אחד מ־predictions, ‏rewards, ‏future_values ו־targets הוא B×1, ולכן השוואת ה־MSE נעשית בין איברים מתאימים באצווה.

שלוש השורות zero_grad, ‏backward ו־step הן צעד האימון המוכר מחלק ג. איפוס הגרדיאנטים מקדים את backward, ואחריו step משנה את משקלי Q. ערכי ההמשך חושבו ללא גרדיאנט, ולכן אין עדכון עקיף של רשת המטרה, וה־backpropagation זורם רק דרך predictions, בדיוק כפי שדרשנו בפרק הקודם. losses שומרת מספר אחד לכל עדכון, לצורך עיון במהלך האימון.

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

שמירה ו־Tester

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

env, agent, losses = train()
torch.save(agent.model.state_dict(), WEIGHTS_PATH)

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

def test(path=WEIGHTS_PATH, games=1000, seed=100):
    env = TicTacToe()
    model = DQN()
    model.load_state_dict(torch.load(path, map_location='cpu', weights_only=True))
    agent = DQNAgent(env, model, training=False)
    return evaluate(env, agent.get_action, games=games, seed=seed)

evaluate היא פונקציית הבדיקה שכבר השתמשנו בה בפרקי האימון הטבלאי. היא משחקת 1,000 משחקים מול O אקראי וסופרת ניצחונות, הפסדים ותיקו; אין בה אימון או שינוי משקלים. כך נבדקת המדיניות של הסוכן לאחר הלמידה.

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

הרצת הדוגמה וקוד המקור

שמרו את חמשת קובצי DQN המקושרים בראש הפרק לצד ttt_env.py, ‏ttt_training.py ו־tabular_agent.py מהפרקים הקודמים. בסביבת Python עם PyTorch הפעילו תחילה dqn_train.py ולאחריו dqn_test.py. הראשון יוצר את קובץ המשקלים, והשני טוען אותו ומדפיס את ספירות המשחקים.

קוד המקור המלא נמצא ב־מאגר Tic_Tac_Toe_DQN בגרסה ששימשה להתאמה. עותק הספר שומר את רשת 11→128→64→1 ואת סדר רכיבי ההדגמה, עם התאמות לממשק המשותף וליעד DQN רגיל. לחומרי בניית משחקים ראו את חלק Pygame ואת אתר הקורס.

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