Ты сможешь! Введение в машинное обучение с подкреплением для программистов и не только

Для меня как программиста одна из самых больших сложностей в изучении технологий машинного обучения ^[1] (ML) и искусственного интеллекта ^[2] (AI) — разбор практических примеров.

Почти весь код туториалов, который мне попадался в открытом доступе, с точки зрения ^[3] кодирования, написан на уровне junior-программиста, что вполне закономерно, ведь все Data Science и ML-инженеры, которых я знаю, в большей степени математики ^[4], а не программисты. И сложность их кода не только в языковых конструкциях и отсутствии хороших практик кодирования, но и в том, что он больше похож на математические выкладки и довольно тяжело читаем для людей, имеющих за плечами только институтский и школьный курс математики. И вся это сложность умножается на непростую теоретическую базу, поэтому если ты не знаком с теорией, то догадаться по коду, для чего нужны выполняемые действия, порой бывает просто решительно невозможно.

Я заинтересовался ML и AI в 2019 году, и с тех пор количество статей и примеров кода в Интернете выросло многократно, но одно, к сожалению, так и осталось неизменно — стиль кодирования примеров и их математичность.

Поэтому решил написать данную статью для таких же программистов как я, которые интересуются технологиями машинного обучения и искусственного интеллекта, где совсем не будет математики, а вместо неё — только код. В общем, всё, как мы привыкли.

Без теории никак?

Технологии машинного обучения и искусственного интеллекта сейчас развиваются с невероятной скоростью и количество направлений просто огромно, поэтому я решил остановиться на одном: обучение с подкреплением ^[5].

Как я говорил, упор будет сделан на кодинг, но ознакомиться с основными понятиями и общим концептом всё же необходимо, но, как и обещал, в статье не буду трогать госпожу математику.

Основные понятия обучения с подкреплением

Сейчас выделяют три раздела машинного обучения:

• Обучение с учителем

• Обучение без учителя

• Обучение с подкреплением

Объединяет эти разделы одно требование: необходимы данные для обучения, которые называются тренировочный датасет.

Для обучения с учителем/без учителя данные подготавливаются заранее, а для обучения с подкреплением заранее подготовленных данных – нет, поэтому:

добыча (майнинг) данных — одна из задач обучения с подкреплением.

Добытчик данных

В обучении с учителем/без учителя подготовкой тренировочного датасета, как правило, занимается человек. Данные, обычно, формируются после взаимодействия с реальным миром. Например, фотографии кошек готовятся для обучения классификатора. Затем они оцифровываются, могут проводиться дополнительные обработки данных и затем этап подготовки завершается приведением оцифрованных данных к формату, который принимает на вход алгоритм.

В обучении с подкреплением сборкой данных занимается программа, которую называют Агент (Agent).

Поставщик данных для обучения

Агент взаимодействует с внешним миром, которое называется Окружение (Environment).

Данные, которые получает Агент после взаимодействия с Окружением принято делить на две категории:

• Состояние (State)

• Вознаграждение (Reward)

Формат данных категории Состояние является произвольным и зависит от окружения и решаемой задачи. Формат данных категории Вознаграждение, как правило, — число, которое количественно показывает, насколько удачное действие совершил Агент в Окружении.

Основная задача обучения

В обучении с подкреплением объектом обучения является Агент, который взаимодействует с Окружением, поэтому основная задача обучения с подкреплением:

обучить Агента взаимодействовать с Окружением, чтобы получаемое вознаграждение было как можно большим, в идеале — максимальным.

Pipeline машинного обучения с подкреплением

Основной pipeline машинного обучения с подкреплением следующий:

1. Агент взаимодействует с Окружением.

2. Окружение возвращает результат взаимодействия в виде Состояния окружения и Вознаграждение.

3. Агент обучается на основе полученных данных от Окружения.

4. Агент переходит к пункту 1.

Политика взаимодействия с Окружением

Алгоритм, который определяет действие Агента в Окружении, называют Политика (Policy). И обучение Агента, на самом деле, сводится к обучению алгоритма Политики.

Итоги. Коротко

Основные понятия:

• Окружение

• Состояние

• Вознаграждение

• Агент

• Действие

• Политика взаимодействия

Основная задача обучения с подкреплением:

• Обучить Агента взаимодействовать с Окружением, чтобы получаемое вознаграждение было как можно большим, в идеале — максимальным.

Подзадача обучения с подкреплением:

• Добыча (майнинг) тренировочных данных для обучения путём взаимодействия с Окружением.

От теории к практике

Обучение с подкреплением похоже на то, как человек обучается, поэтому теория концептуально выглядит понятной и довольно несложной, но трудности начинаются тогда, когда возникает вопрос: как от теории перейти к практике?

И первое, что нужно сделать —

определиться с каким окружением будет взаимодействовать Агент.

Установка необходимых библиотек

Перед разбором практической части нужно установить все необходимые зависимости.

Для примера потребуется Python 3.12 и следующие библиотеки:

• numpy==1.26.4

• gymnasium==1.2.2

• pygame==2.1.3

• pandas==2.2.3

• matplotlib==3.9.2

• tqdm==4.67.1

• torch==2.10.0

Подробно описывать каждую библиотеку не буду, для этого можно обратиться к официальной документации, но вкратце опишу, где будет использоваться каждая библиотека.

Создание окружения

Первое, что необходимо сделать — создать окружение, с которым будет взаимодействовать агент. Для туториала я выбрал окружение GridWorld, реализацию которого предоставляет библиотека gymnasium.

Вот код создания окружения:

import gymnasium as gym

def train():
    env = gym.make('gymnasium_env/GridWorld-v0', size=7)

Функция make() принимает два параметра:

• имя окружения;

• размер игрового поля.

Всё, окружение готово, теперь можно переходить к созданию Агента.

Класс Агент

Агент — это программа, которая взаимодействует с окружением и обучается получать в нём наибольшую награду.

Алгоритм работы Агента следующий:

1. Выбрать действие в окружении.

2. Сделать действие в окружении (взаимодействие с окружением).

3. Запомнить полученный опыт ^[6] (результат взаимодействия с окружением).

4. Обучиться на полученном опыте.

Исходя из этого алгоритма, я создал класс с таким интерфейсом:

import numpy as np

class GridWorldAgent():
   def get_action(self, obs: dict[str, np.ndarray]) -> int:
       pass

   def memorize_exp(self, action: int, reward: float, obs: dict[str, np.ndarray]):
       pass

   def learn(self) -> float:
       pass

где

• obs — словарь, который описывает состояние окружения (obs — сокр. от observation). Словарь состоит из 2-х ключей: agent, target. Здесь agent описывает текущее положение агента на игровом поле, а target описывает направление движения (вверх, вниз, влево, вправо) к цели.

• action — действие, сделанное агентом (измеряется от 0 до 3, где каждая цифра обозначает движение в одну из 4-х доступных сторон)

• reward — награда за действие (насколько близко агент приблизился к цели, измеряется от 0 до 1)

Политика выбора действия Агента

Агент будет выбирать действие исходя из текущего состояния окружения, поэтому функция get_action() принимает состояние окружения в виде аргумента.

Алгоритм, который определяет действие Агента в Окружении называют Политика (Policy). И обучение Агента, на самом деле, сводится к обучению алгоритма Политики.

Выделяют два вида Политики:

• исследование окружения (exploration);

• достижения наибольшей награды (explotation).

Политика Исследования окружения (exploration) не гарантирует получение наибольшей награды, но эта политика приносит опыт, тот самый тренировочный датасет, на котором обучается Агент. Поэтому в качестве Политики Исследования окружения чаще всего используют стратегию Случайно выбранного действия.

Политика Достижения наибольшей награды (explotation) — это тот алгоритм Политики, который необходимо обучить выбирать действие, приводящее к получению наибольшей награды в заданном Окружении.

По этой причине я создал два интерфейса политики (т. к. в Python нет понятия интерфейса, его роль в примере играют абстрактные классы):

import numpy as np

from abc import ABC, abstractmethod

class AbstractPolicy(ABC):


    @abstractmethod
    def get_action(self, obs: dict[str, np.ndarray]) -> int:
        pass



class AbstractPolicyLearnable(AbstractPolicy):

    @abstractmethod
    def learn(self, train_samples: list[tuple]) -> float:
        pass

Выбор действия по-прежнему зависит от текущего состояния окружения, поэтому функция get_action() принимает на вход соответствующий аргумент. Обучение же происходит на тренировочном датасете и соответствующий аргумент объявлен в сигнатуре функции learn().

Использование Агентом Политики

Каждый вид политики нужен Агенту, ведь с помощью одной он будет собирать новый опыт, а с помощью другой — обучаться получать наибольшую награду в Окружении. Добавлю обе политики в код класса Агента:

import numpy as np

class GridWorldAgent():

   def __init__(self, policy_explotate: AbstractPolicyLearnable, policy_explorate: AbstractPolicy):
        self.policy_explotate: AbstractPolicyLearnable = policy_explotate
        self.policy_explorate: AbstractPolicy = policy_explorate


   def get_action(self, obs: dict[str, np.ndarray]) -> int:
       pass

   def memorize_exp(self, action: int, reward: float, obs: dict[str, np.ndarray]):
       pass

   def learn(self) -> float:
       pass

При создании Агента, объект получит две политики: для набора опыта (policy_explorate) и для достижения наибольшей награды (policy_explotate). Обратите внимание ^[7], что policy_explotate — политика, которая имеют функцию обучения; у policy_explorate такой функции нет.

Далее осталось написать реализацию функции get_action() и самая большая сложность в его реализации — это определить в каком случае использовать одну политику, а в каком — другую.

Самая простое и популярное решение — высчитывать случайное число и сравнивать его с пороговым значением. Результат сравнения будет определять вид используемой политики.

import numpy as np
import random 

class GridWorldAgent():

   def __init__(self, policy_explotate: AbstractPolicyLearnable, policy_explorate: AbstractPolicy):
        self.policy_explotate: AbstractPolicyLearnable = policy_explotate
        self.policy_explorate: AbstractPolicy = policy_explorate


   def get_action(self, obs: dict[str, np.ndarray]) -> int:
        sample = random.random()
        eps_threshold = 0.9
        if sample < eps_threshold:
            return self.policy_explotate.get_action(obs)
        else:
            return self.policy_explorate.get_action(obs)


   def memorize_exp(self, action: int, reward: float, obs: dict[str, np.ndarray]):
       pass

   def learn(self) -> float:
       pass

Переменная eps_threshold — то самое пороговое значение. Для простоты примера в статье я указал константу, но в финальном примере (ссылка на репозиторий в конце статьи) этот параметр динамически вычисляется по определённому алгоритму, который учитывает количество сделанных Агентом шагов. И чем больше Агент сделает шагов, тем чаще он будет использовать политику Достижения наибольшей награды.

Реализация Политик

Агент умеет взаимодействовать с Политиками для выбора действия, но сейчас не реализована ни одна Политика. Пора этим заняться.

Напомню, что у Агента в наличии две политики:

• исследование окружения (exploration);

• достижения наибольшей награды (explotation).

В качестве Политики Исследования окружения чаще всего используют стратегию Случайно выбранного действия. Вот код этой политики:

import numpy as np
import random

class PolicyRandom(AbstractPolicy):

    def __init__(self, env_action_num: int):
        super().__init__()
        self.env_action_num: int = env_action_num


    def get_action(self, obs: dict[str, np.ndarray]) -> int:
        num_actions = self.env_action_num
        next_action = random.randrange(num_actions)

        return next_action

При создании объекта PolicyRandom передаётся аргумент env_action_num — он указывает число доступных действий в окружении. При вызове функции get_action() просто берётся произвольное число в промежутке от 0 до env_action_num, независимо от того в каком состоянии находится сейчас окружение. Обратите внимание, что AbstractPolicy — необучаемая политика, у неё отсутствует функция learn().

Зато функция learn() должна присутствовать у политики Достижения наибольшей награды.

Выбор этого вида политик — огромен, но самые простейшие это Q-table и Deep Q-Network (DQN). Обе политики реализованы в финальном примере (ссылка на репозиторий в конце статьи), но в статье я рассмотрю всего один: DQN.

Вот код этой политики:

import torch
import torch.nn as nn
import numpy as np

class PolicyDQN(AbstractPolicyLearnable):

    def __init__(self, policy_net: nn.Module):
        self.policy_net = policy_net

    
    def get_action(self, obs: dict[str, np.ndarray]) -> int:
        pass


    def learn(self, train_samples: list[tuple]) -> float:
        pass

При создании объекта PolicyDQN передаётся, как параметр, простая полносвязная нейронная сеть. Вот её код:

import torch.nn as nn
import torch.nn.functional as F
import torch

class DQN(nn.Module):

    def __init__(self):
        super(DQN, self).__init__()
        self.fc1 = nn.Linear(in_features=2, out_features=128)
        self.fc2 = nn.Linear(in_features=128, out_features=128)
        self.fc3 = nn.Linear(in_features=128, out_features=4)

   
    def forward(self, x: torch.Tensor) -> torch.Tensor:
        x = F.relu(self.fc1(x))
        x = F.relu(self.fc2(x))
        x = self.fc3(x)
        return x

Входящий тензор x будет состоять всего из двух значений: направление до цели по X, направление по цели по Y. Исходящий тензор будет состоять из четырёх значений — по количеству доступных действий в окружении.

Эту нейронную сеть будет использовать класс PolicyDQN для выбора действия, которое позволяет получить наибольшую награду. Функция get_action() служит для этой цели:

import torch
import torch.nn as nn
import numpy as np

class PolicyDQN(AbstractPolicyLearnable):

    def __init__(self, policy_net: nn.Module):
        self.policy_net = policy_net

    
    def get_action(self, obs: dict[str, np.ndarray]) -> int:
        with torch.no_grad():
            policy_input = self.__to_policy_input_obs(obs)
            policy_output = self.policy_net(policy_input)
            next_action = policy_output.argmax()
                         
            return next_action.item()


    def learn(self, train_samples: list[tuple]) -> float:
        pass

Функция __to_policy_input_obs() конвертирует словарь, который представляет состояние окружения в тензор torch, который принимается на вход в нейронную сеть. Код конвертации чисто технический, поэтому для краткости не буду его приводить в статье, с ним можно ознакомиться в финальном примере (ссылка на репозиторий в конце статьи).

Результатом работы нейронной сети policy_output является вектор чисел, и с помощью функции argmax() определяется порядковый индекс наибольшего значения в этом векторе. Этот индекс и является номером действия, который позволит получить наибольшую награду, правда только после того, как обучим нейронную сеть, но функцию learn() реализую чуть позже.

Обучение Агента

Напомню, что алгоритм работы Агента следующий:

1. выбрать действие в окружении;

2. сделать действие в окружении (взаимодействие с окружением);

3. запомнить полученный опыт (результат взаимодействия с окружением);

4. обучиться на полученном опыте.

Все компоненты готовы для реализации этого алгоритма:

import gymnasium as gym

def train():
    env = gym.make('gymnasium_env/GridWorld-v0', size=7)
    
    policy_net = DQN()
    policy_explotation = PolicyDQN(policy_net)
    policy_exploration = PolicyRandom(env.action_space.n)

    agent = GridWorldAgent(policy_explotation, policy_exploration)

    for episode in tqdm(range(100)):
        # Создать новую игровую сессию
        obs, info = env.reset()

        for step in range(300):
            # 1. Выбрать действие в окружении
            action = agent.get_action(obs)

            # 2. Сделать действие в окружении (повзаимодействовать с окружением)
            obs, reward, terminated, truncated, info = env.step(action)

            # 3. Запомнить полученный опыт (результат взаимодействия с окружением)
            agent.memorize_exp(action, reward, obs)

            # 4. Обучиться на полученном опыте
            agent.learn()

Каждая игровая сессия — это эпизод. Агенту для обучение даётся 100 эпизодов по 300 шагов в каждом. На каждом шаге Агент выполняет свой алгоритм состоящий из четырёх пунктов.

Цикл обучения Агента реализован. Осталось наполнить реализацией пункты 3 и 4 алгоритма работы Агента.

Память Агента

После того, как Агент повзаимодействовал с окружением, он получает текущее состояние окружения (obs) и награду за выполненное действие (reward). Это данные можно использовать для тренировки Агента, поэтому ему необходимо сохранить их в своей памяти ^[8]. Эту операцию выполняет функция memorize_exp():

import numpy as np
import random 

class GridWorldAgent():

   def __init__(self, policy_explotate: AbstractPolicyLearnable, policy_explorate: AbstractPolicy):
        self.policy_explotate: AbstractPolicyLearnable = policy_explotate
        self.policy_explorate: AbstractPolicy = policy_explorate
        self.memory = AgentMemory()


   def get_action(self, obs: dict[str, np.ndarray]) -> int:
        sample = random.random()
        eps_threshold = 0.9
        if sample < eps_threshold:
            return self.policy_explotate.get_action(obs)
        else:
            return self.policy_explorate.get_action(obs)


   def memorize_exp(self, action: int, reward: float, obs: dict[str, np.ndarray]):
       self.memory.save(action, reward, obs)


   def learn(self) -> float:
       pass

Функция memorize_exp() очень проста: она передаёт все входящие переменные объекту, который хранится в переменной memory. Я создал отдельный класс AgentMemory, который хранит опыт взаимодействия Агента с Окружением и предоставляет методы по сохранению нового опыта и извлечению ранее накопленного. Вот код класса:

import numpy as np
import random

from collections import deque, namedtuple


class AgentMemory():
    # объявление текстовых констант не несёт смысловой нагрузки, поэтому опущено для краткости

    Sample = namedtuple(
        "Sample",
        (SAMPLE_ACTION, SAMPLE_REWARD, SAMPLE_IS_POSITIVE_EXP,  SAMPLE_AGENT_ROW_N_BEFORE, 
         SAMPLE_AGENT_COL_N_BEFORE, SAMPLE_TARGET_ROW_D_BEFORE, SAMPLE_TARGET_COL_D_BEFORE, 
         SAMPLE_AGENT_ROW_N_AFTER, SAMPLE_AGENT_COL_N_AFTER, SAMPLE_TARGET_ROW_D_AFTER,
         SAMPLE_TARGET_COL_D_AFTER))
    
    def __init__(self):
        self.capacity = 5000
        self.memory_new_exp = deque([], maxlen=capacity)
        self.memory_positive = deque([], maxlen=capacity)


    def save(self, action: int, reward: float, obs: dict[str, np.ndarray]):
        pass


    def load_sample_random_positive(self) -> list[tuple]:
        pass

Память Агента поделена на два раздела:

• любой новый опыт, который храниться в переменной memory_new_exp;

• позитивный опыт, который храниться в переменной memory_positive.

Такое деление сделано по следующей причине:

Говорят: нужно учиться на ошибках. Но учиться на ошибках и на ошибочных примерах — это немного разные вещи, поэтому необходимо выделять позитивный опыт, на котором и будет происходить обучение.

Внутри класса объявлен namedtuple Sample — объект, который представляет ячейку памяти Агента.

Теперь реализую сохранение опыта Агента:

import numpy as np
import random

from collections import deque, namedtuple

class AgentMemory():
    # объявление текстовых констант (SAMPLE_ACTION, и т.п.) не несёт смысловой нагрузки, поэтому опущено для краткости

    Sample = namedtuple(
        "Sample",
        (SAMPLE_ACTION, SAMPLE_REWARD, SAMPLE_IS_POSITIVE_EXP,  SAMPLE_AGENT_ROW_N_BEFORE, 
         SAMPLE_AGENT_COL_N_BEFORE, SAMPLE_TARGET_ROW_D_BEFORE, SAMPLE_TARGET_COL_D_BEFORE, 
         SAMPLE_AGENT_ROW_N_AFTER, SAMPLE_AGENT_COL_N_AFTER, SAMPLE_TARGET_ROW_D_AFTER,
         SAMPLE_TARGET_COL_D_AFTER))
    
    def __init__(self):
        self.capacity = 5000
        self.memory_new_exp = deque([], maxlen=capacity)
        self.memory_positive = deque([], maxlen=capacity)


    def save(self, action: int, reward: float, obs: dict[str, np.ndarray]):
        sample = self.__convert_to_sample(action, reward, obs)

        self.memory_new_exp.append(sample)

        is_positive_exp = getattr(sample, self.SAMPLE_IS_POSITIVE_EXP)
        if (is_positive_exp):
            self.memory_positive.append(sample)


    def load_sample_random_positive(self) -> list[tuple]:
        pass

Функция __convert_to_sample() перекладывает данные из входных переменных в новый namedtuple Sample. Его реализация чисто техническая, поэтому приводить код в примере не буду. Код можно посмотреть в финальном примере (ссылка на репозиторий в конце статьи).

Хотя отдельно стоит проговорить как определяется значение флага SAMPLE_IS_POSITIVE_EXP. Оно определяется по награде (reward): если награда текущего sample больше предыдущего, то опыт считается позитивным.

Обучение Агента

Теперь Агент умеет сохранять в памяти свой опыт взаимодействия с Окружением. Более того, он умеет выделять из него позитивный. Это очень важно для обучения, т. к. обучаться агент будет только на позитивном опыте.

Реализую последнюю функцию агента learn():

import numpy as np
import random 

class GridWorldAgent():

   def __init__(self, policy_explotate: AbstractPolicyLearnable, policy_explorate: AbstractPolicy):
        self.policy_explotate: AbstractPolicyLearnable = policy_explotate
        self.policy_explorate: AbstractPolicy = policy_explorate
        self.memory = AgentMemory()


   def get_action(self, obs: dict[str, np.ndarray]) -> int:
        sample = random.random()
        eps_threshold = 0.9
        if sample < eps_threshold:
            return self.policy_explotate.get_action(obs)
        else:
            return self.policy_explorate.get_action(obs)


   def memorize_exp(self, action: int, reward: float, obs: dict[str, np.ndarray]):
       self.memory.save(action, reward, obs)


   def learn(self) -> float:
        memory_sample_tuples = self.memory.load_sample_random_positive()
        loss_output = self.policy_explotate.learn(memory_sample_tuples)      

        return loss_output

Реализация функции learn() довольная простая:

• вытаскиваем из памяти позитивный опыт;

• передаём позитивный опыт для обучения Политике Достижения наибольшей награды.

Реализую оставшиеся 2 функции: load_sample_random_positive() и policy_explotate.learn()

Реализация функции load_sample_random_positive():

import numpy as np
import random

from collections import deque, namedtuple

class AgentMemory():
    # объявление текстовых констант (SAMPLE_ACTION, и т.п.) не несёт смысловой нагрузки, поэтому опущено для краткости

    Sample = namedtuple(
        "Sample",
        (SAMPLE_ACTION, SAMPLE_REWARD, SAMPLE_IS_POSITIVE_EXP,  SAMPLE_AGENT_ROW_N_BEFORE, 
         SAMPLE_AGENT_COL_N_BEFORE, SAMPLE_TARGET_ROW_D_BEFORE, SAMPLE_TARGET_COL_D_BEFORE, 
         SAMPLE_AGENT_ROW_N_AFTER, SAMPLE_AGENT_COL_N_AFTER, SAMPLE_TARGET_ROW_D_AFTER,
         SAMPLE_TARGET_COL_D_AFTER))
    
    def __init__(self):
        self.capacity = 5000
        self.memory_new_exp = deque([], maxlen=capacity)
        self.memory_positive = deque([], maxlen=capacity)


    def save(self, action: int, reward: float, obs: dict[str, np.ndarray]):
        sample = self.__convert_to_sample(action, reward, obs)

        self.memory_new_exp.append(sample)

        is_positive_exp = getattr(sample, self.SAMPLE_IS_POSITIVE_EXP)
        if (is_positive_exp):
            self.memory_positive.append(sample)


    def load_sample_random_positive(self) -> list[tuple]:
        batch_size = 200 if len(self.memory_positive) > 200 else len(self.memory_positive)

        return random.sample(self.memory_positive, batch_size)

Задача функции load_sample_random_positive() вернуть произвольно выбранный опыт, но не больше 200 sample за один раз. Это число подобранно экспериментально. Если количество опыта меньше заданного порога, то выбирается весь существующий опыт.

Остался последний шаг: обучить Политику Достижения наибольшей выгоды на позитивном опыте взаимодействии Агента с Окружением. Вот код обучения Политики:

class PolicyDQN(AbstractPolicyLearnable):

    def __init__(self, policy_net: nn.Module):
        self.policy_net = policy_net
        self.optimizer = torch.optim.Adam(self.policy_net.parameters(), lr=0.01)
        self.loss_func = nn.MSELoss()

    
    def get_action(self, obs: dict[str, np.ndarray]) -> int:
        with torch.no_grad():
            policy_input = self.__to_policy_input_obs(obs)
            policy_output = self.policy_net(policy_input)
            next_action = policy_output.argmax()
                         
            return next_action.item()


    def learn(self, train_samples: list[tuple]) -> float:
        input_tensor_stack_before, action_tensor_stack = self.__get_input_and_reward_tensors(train_samples)

        self.optimizer.zero_grad()

        actions_predict = self.policy_net(input_tensor_stack_before)
        actions_target = self.__to_target_actions(actions_predict, action_tensor_stack)

        loss_output = self.loss_func(actions_predict, actions_target)
        loss_output.backward()
        self.optimizer.step()

        return loss_output.item()

Для обучения потребуются два дополнительных объекта:

• оптимизатор (optimizer) Adam;

• функция потерь (loss_func), квадратичная MSELoss.

Далее необходимо конвертировать train_samples в torch тензоры, которые будут переданы на вход нейронной сети (policy_net). Конвертацией занимается функция __get_input_and_reward_tensors(). Код функции __get_input_and_reward_tensors() чисто технический, поэтому в примере его не буду показывать. Ознакомиться с кодом функции можно в финальном примере (ссылка на репозиторий в конце статьи).

Дополнительно функция __get_input_and_reward_tensors() возвращает action_tensor_stack, который потребуется для формирования ожидаемого результата (actions_target).

Затем вычисляем действия с помощью нейронной сети и сохраняем результат в переменную actions_predict. Далее готовим ожидаемые действия с помощью функции __to_target_actions(). Код функции __to_target_actions() чисто технический, поэтому в примере его не буду показывать. Ознакомиться с кодом функции можно в финальном примере (ссылка на репозиторий в конце статьи).

А дальше классика машинного обучения с учителем:

• передаём actions_predict и actions_target в функцию потерь loss_func;

• выполняем backward propagation;

• запускаем работу оптимизатора

Работа функции learn() завершается возвращением количественного значения ошибки ^[9], которую вычислила функция потерь.

На этом реализация обучения закончена.

Финиш обучения. Что дальше?

На текущий момент полностью реализован алгоритм работы Агента:

1. выбрать действие в окружении.;

2. сделать действие в окружении (взаимодействие с окружением);

3. запомнить полученный опыт (результат взаимодействия с окружением);

4. обучиться на полученном опыте.

Более того алгоритм выполняется в множестве игровых сессий, которые называются эпизоды. Агенту достаточно 100 эпизодов по 300 шагов, чтобы научиться играть в игровом окружении GridWorld.

Но как узнать:

обучился ли агент?

Тут нам помогут два инструмента:

1. метрики;

2. тест.

Метрики

Существуют стандартные метрики, которые используются в машинном обучении, но в данном примере я решил отойти от канона и добавил специфичные (скажу так: продуктовые) метрики, т. к. они показались мне наиболее информативными и понятными для конечного пользователя. Их всего четыре:

1. отношение общего количества действий в эпизоде к позитивным действиям, сделанными с помощью политики Исследования окружения;

2. отношение общего количества действий в эпизоде к позитивным действиям, сделанными с помощью политики Достижения наибольшей наград;

3. количество успешно завершённых игровых сессий;

4. размер ошибки Агента во время обучения (значения функции потерь).

Так как метрики специфичные под эту задачу, код реализации разбирать не буду. С ней можно ознакомиться самостоятельно в финальном примере (ссылка на репозиторий в конце статьи).

Графики метрик выглядят вот так:

По данным метрикам можно посмотреть динамику успешности действий Агента во время обучения.

Тест

Метрики раскрывают детали процесса обучения и показывают учился ли вообще Агент или нет, но насколько правильно он обучился, покажет только тестирование.

Тест отличается от обучения тем, что в нём выполняются первые два шага алгоритма Агента вместо четырёх:

1. выбрать действие в окружении;

2. сделать действие в окружении (повзаимодействовать с окружением).

Код теста выглядит так:

import gymnasium as gym

# код обучения опущен для краткости

def play(env: gym.Env, agent: GridWorldAgent):
    terminated = False
    step_count = 0
    # Создать новую игровую сессию 
    obs, info = env.reset()
    while not terminated and step_count < 30:
        step_count += 1
        
        # 1. Выбрать действие в окружении
        action = agent.get_action(obs)
        # 2. Сделать действие в окружении (повзаимодействовать с окружением)
        obs, reward, terminated, truncated, info = env.step(action)


# пересоздаём окружение с render_mode="human", которое активирует графическое отображение игрового поля с помощью библиотеки pygame
env = gym.make(gym_env_id, size=7, render_mode="human")

# agent был создан ранее на этапе обучения
play(env, agent)

Для примера я выбрал ручной тест, который с помощью библиотеки pygame отобразит игровое поле и действия Агента на нём. Если модель обучилась, то сколько бы не запускался тест, Агент безошибочно успешно завершит игровую сессию.

Сравнение работы алгоритмов

В конце статьи будет ссылка на финальный пример, в котором реализована ещё одна политика Достижения наибольшей награды. Эта политика называется Q-table. Она значительно проще DQN и по-хорошему стоило начать изучение с неё, но я подумал, что тебе, дорогой читатель, пример с нейронной сетью будет гораздо интересней. Надеюсь, я не ошибся.

Здесь сравнительное видео Агента играющего в GridWorld до обучения, после обучения по политике Q-table и политике DQN. Обратите внимание, как по-разному ходит Агент при использовании политик Q-table и DQN.

Заключение

Основные понятия:

• Окружение

• Состояние

• Вознаграждение

• Агент

• Действие

• Политика взаимодействия

Основная задача обучения с подкреплением:

• Обучить Агента взаимодействовать с Окружением, чтобы получаемое вознаграждение было как можно большим, в идеале — максимальным.

Подзадача обучения с подкреплением:

• Добыча (майнинг) тренировочных данных для обучения путём взаимодействия с Окружением.

Алгоритм работы Агента:

1. выбрать действие в окружении;

2. сделать действие в окружении (взаимодействие с окружением);

3. запомнить полученный опыт (результат взаимодействия с окружением);

4. обучиться на полученном опыте.

Интерфейс Агента:

• определить действие для Окружения;

• запомнить опыт взаимодействия с Окружением;

• обучиться.

Интерфейс Политики Агента:

• вычислить действие по текущему состоянию Окружения;

• обучиться на полученном опыте.

Интерфейс Памяти Агента:

• сохранить опыт взаимодействия с Окружением;

• взять из памяти только позитивные сэмплы опыта.

Общая детализированная схема:

Финальный пример

В своём репозитории я выложил финальный пример. В нём значительно больше деталей, которые я намеренно опустил в статье для того, чтобы сосредоточится на самом главном и не утонуть в коде.

• Ссылка на репозиторий: https://github.com/DarrMirr/reinforcement-learning-introduction ^[10]