Ты сможешь! Введение в машинное обучение с подкреплением для программистов и не только. ai.. ai. codding.. ai. codding. machinelearning.. ai. codding. machinelearning. ml.. ai. codding. machinelearning. ml. python.. ai. codding. machinelearning. ml. python. reinforcement-learning.

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

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

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

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

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

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

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

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

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

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

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

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

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

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

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

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

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

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

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

Состояние (State)
Вознаграждение (Reward)

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

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

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

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

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

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

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

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

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

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

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

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

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

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

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

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

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

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

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

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

Для примера я выбрал довольно простое окружение: GridWorld.

Задача игры: дойти из стартовой точки в финишную.

Само окружение разрабатывать не будем, а воспользуемся уже готовым, которая предоставляет библиотека gymnasium.

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

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

Для примера потребуется 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

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

Назначение	Название библиотеки
Окружение для действий агента	`gymnasium`
Обучение политик агента	`torch, numpy`
Отрисовка графиков	`pandas, matplotlib`
Отображение прогресса обучения в терминале	`tqdm`
Графическое отображение игры	`pygame`

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

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

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

import gymnasium as gym

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

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

имя окружения;
размер игрового поля.

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

Класс Агент

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

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

Выбрать действие в окружении.
Сделать действие в окружении (взаимодействие с окружением).
Запомнить полученный опыт (результат взаимодействия с окружением).
Обучиться на полученном опыте.

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

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). Обратите внимание, что 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() реализую чуть позже.

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

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

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

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

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). Это данные можно использовать для тренировки Агента, поэтому ему необходимо сохранить их в своей памяти. Эту операцию выполняет функция 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() завершается возвращением количественного значения ошибки, которую вычислила функция потерь.

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

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

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

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

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

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

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

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

метрики;
тест.

Метрики

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

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

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

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

Пример графика метрик после обучения алгоритма DQN

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

Номера графиков (сверху-вниз)	Заключение
1, 2	Когда Агент сталкивается с новой для себя ситуацией в Окружении, то он делает много действий, а процент позитивных действий довольно мал. Когда Агент находится в знакомой для него ситуации в Окружении, то общее число действий мало и большая часть, а зачастую все из них позитивные.
3	Большую часть игровых сессий Агент завершает успехом во время обучения.
4	Количественное значение ошибки уменьшается со временем обучения.

Тест

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

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

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

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

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.

Заключение

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

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

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

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

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

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

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

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

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

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

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

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

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

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

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

Детализированная схема взаимодействия Агента с Окружением

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

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

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

Автор: VladimirPolukeev

Источник

Запись добавлена: 01.07.2026 в 12:12
Оставлено в

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

Меню навигации

На главную

Главное

Рубрики

Методики

Информация

Из архивов

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

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

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

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

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

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

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

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

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

Класс Агент

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

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

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

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

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

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

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

Метрики

Тест

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

Заключение

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

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

Меню навигации

Рекомендуем

На главную

Главное

Рубрики

Методики

Информация

Из архивов

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

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

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

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

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

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

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

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

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

Класс Агент

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

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

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

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

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

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

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

Метрики

Тест

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

Заключение

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