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

Для многих программистов одна из главных сложностей в изучении машинного обучения (ML) и искусственного интеллекта (AI) — разбор практических примеров. Код в большинстве доступных туториалов написан на уровне junior-разработчика и больше напоминает математические выкладки, из‑за чего его трудно понять тем, кто не углублялся в теорию. Автор статьи, заинтересовавшийся ML и AI в 2019 году, решил написать материал без математики — только с кодом, понятный коллегам-программистам. В центре внимания — обучение с подкреплением.

В ML выделяют три раздела: обучение с учителем, обучение без учителя и обучение с подкреплением. Во всех случаях нужны данные для обучения — тренировочный датасет. В обучении с учителем и без учителя данные подготавливают заранее, а в обучении с подкреплением их нужно добывать (майнить) — и это одна из ключевых задач. В первых двух случаях подготовкой датасета обычно занимается человек, а в обучении с подкреплением — программа, которую называют Агент (Agent). Агент взаимодействует с внешним миром (Окружением, Environment). Данные, которые Агент получает после взаимодействия с Окружением, делят на две категории: Состояние (State) и Вознаграждение (Reward). Задача обучения с подкреплением — научить Агента взаимодействовать с Окружением так, чтобы получать максимально возможное вознаграждение.

Pipeline обучения с подкреплением выглядит так: Агент взаимодействует с Окружением; Окружение возвращает результат взаимодействия в виде Состояния и Вознаграждения; Агент обучается на основе полученных данных; затем процесс повторяется. Алгоритм, который определяет действие Агента в Окружении, называют Политикой (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.

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

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

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

В качестве Политики Исследования окружения можно использовать стратегию случайно выбранного действия (PolicyRandom). При создании объекта PolicyRandom передаётся аргумент envactionnum — число доступных действий в окружении. При вызове функции getaction() берётся произвольное число в диапазоне от 0 до envaction_num.

Для Политики Достижения наибольшей награды есть много вариантов, например Q-table и Deep Q-Network (DQN). В статье рассматривается DQN. При создании объекта PolicyDQN передаётся полносвязная нейронная сеть. Входящий тензор x будет состоять из двух значений: направление до цели по X и по Y. Исходящий тензор — из четырёх значений, по количеству доступных действий в окружении. Функция getaction() служит для выбора действия, которое позволит получить наибольшую награду. Функция _topolicyinput_obs() конвертирует словарь, представляющий состояние окружения, в тензор torch, который принимается на вход нейронной сети. Результат работы нейронной сети — вектор чисел; с помощью функции argmax() определяется индекс наибольшего значения в этом векторе — он и будет номером действия, которое позволит получить максимальную награду (после обучения нейронной сети).

Алгоритм работы Агента: выбрать действие в окружении; сделать действие; запомнить полученный опыт; обучиться на нём. Агенту для обучения даётся 100 эпизодов по 300 шагов в каждом. На каждом шаге Агент выполняет алгоритм из четырёх пунктов. После взаимодействия с окружением Агент получает текущее состояние (obs) и награду за действие (reward) — эти данные нужно сохранить в памяти. Для этого используется функция memorizeexp(), которая передаёт все входящие переменные объекту, хранящемуся в переменной memory. Автор создал класс AgentMemory, который хранит опыт взаимодействия Агента с Окружением и предоставляет методы для сохранения и извлечения опыта. Память Агента поделена на два раздела: новый опыт (memorynewexp) и позитивный опыт (memorypositive). Позитивный опыт выделяется по награде (reward): если награда текущего sample больше предыдущей, опыт считается позитивным.

Функция learn() вытаскивает из памяти позитивный опыт и передаёт его для обучения Политике Достижения наибольшей награды. Функция loadsamplerandom_positive() возвращает произвольно выбранный опыт, но не больше 200 sample за один раз. Если опыта меньше заданного порога, выбирается весь существующий опыт. Последний шаг — обучить Политику Достижения наибольшей выгоды на позитивном опыте взаимодействия Агента с Окружением.

import gymnasium as gym

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

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

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

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

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

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

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

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

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

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()

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

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

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

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

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)

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()