Введение
Стремительное развитие глубокого обучения с подкреплением (deep reinforcement learning – DRL), представляющего собой комбинацию глубокого обучения (DL) и обучения с подкреплением (RL), привлекает все больше исследователей из самых разных сфер науки к применению DRL для решения задач в своих областях исследований. Благодаря способности глубокого обучения работать с непрерывным или сложным пространством состояний и способности обучения с подкреплением учиться методом проб и ошибок в сложной среде, DRL особенно хорошо подходит для решения задач, для которых не хватает хороших точных или эвристических методов в сложных средах. Поскольку решение большинства задач обучения с подкреплением требует чрезвычайно большого количества данных, большинство DRL(или RL)-агентов обучаются в симулированной среде. Основным выбором для обучения DRL-моделей, благодаря своей внушительной библиотеке инструментов машинного обучения, стал Python. Однако использовать Python как язык программирования для создания крупномасштабных симуляций, имитирующих сложные среды, довольно сложно.
AnyLogic — идеальная платформа для создания симуляционных моделей для обучения DRL-агентов в сложных средах. Недавно разработанная библиотека Alpyne — это библиотека Python, которая позволяет пользователям обучать DRL-агентов на Python, взаимодействуя с моделью AnyLogic. К сожалению, она все еще недостаточно стабильна для работы со сложными симуляционными моделями.
В этой статье мы представляем новый способ взаимодействия DRL с симуляционными моделям в AnyLogic с помощью библиотеки Pypeline. Этот метод также может быть использован для (не глубокого) обучения с подкреплением, но благодаря своей простоте большинство сред, для которых хватает простого RL, могут быть смоделированы непосредственно в самих языках программирования, таких как Python.
Стандартным способом обучения DRL-агента является взаимодействие с симуляционными моделями из Python. В рамках этого метода DRL-агент вызывается из симуляционной модели, чтобы наблюдать и взаимодействовать с моделью через определенные временные интервалы, и сохраняет все свои критические компоненты, например, буфер воспроизведения и нейронные сети, в локальной памяти в конце каждого эпизода. Такой подход обеспечивает стабильный способ реализации DRL в моделях AnyLogic.
Начнем же мы с краткого обзора основных компонентов, задействованных в этом методе. В частности, для демонстрации мы здесь используем реализацию Deep Q-Learning, но этот метод может быть применен к различным RL-алгоритмам. Затем мы рассмотрим простой пример (упрощенный OpenAI Gym Taxi-v3), чтобы продемонстрировать реализацию этого метода.
Краткий обзор основных компонентов
Компоненты на стороне AnyLogic (среда)
Чтобы взаимодействовать с Python, сначала нужно установить в модель AnyLogic библиотеку Pypeline. Поскольку в данной статье речь идет не о библиотеке Pypeline, детальные инструкции по установке и использованию библиотеки вы можете найти на сайте.
После установки библиотеки Pypeline нам нужно импортировать модуль Python для обучения DRL и создать инстанс класса обучения DRL в секции On Startup главного агента. В дальнейшем этот инстанс класса обучения DRL будет вызываться во время работы симуляции на каждом шаге действия для получения информации о состоянии, совершения какого-либо действия на ее основе и получения вознаграждения от симуляционной среды.
Для обучения RL-агента симуляционная среда должна обладать четырьмя важными возможностями:
выводить информацию о состоянии из среды (1),
выдавать вознаграждение от среды (2),
получать и выполнять действия от RL-агента (3), и
сообщать RL-агенту о завершении эпизода (4).
Таким образом, в симуляции должны быть предусмотрены функции для всех этих четырех возможностей. Специально для нашей реализации были созданы функция под пункты (1) и (2), и еще одна функция для пунктов (3) и (4). Функция для (1) просто возвращает информацию о текущем состоянии в виде вещественного или целочисленного списка. Функция для (2) просто возвращает текущее вознаграждение в вещественном или целочисленном виде. Функция для (3) и (4) принимает от RL-агента входные данные о действиях в среде и возвращает, будет ли симуляция закончена после выполнения действия.
Наконец, функция, коммуницирующая с действием RL, должна использовать все четыре вышеупомянутые возможности для взаимодейстия с RL-агентом на каждом шаге действия.
Компоненты на стороне Python (RL-агент)
Как говорилось выше, в начале каждого эпизода инициализируется новый инстанс RL-агента. Поскольку в каждом эпизоде инициализируется новый RL-агент, очень важно реализовать какой-нибудь способ локальной записи важной информации об обучении, чтобы эта информация не была потеряна в конце каждого эпизода обучения. Здесь мы используем JSON и функции сохранения из библиотек наподобие PyTorch для сохранения информации в конце каждого эпизода обучения и загрузки ее при инициализации. На примере Deep Q-Learning, важная информация включает, но не ограничивается буфером воспроизведения, сетью политик, целевой сетью, количеством пройденных шагов, буфером вознаграждения, историей потерь и оптимизатором (если используется оптимизатор импульса, например ADAM). Более подробно об алгоритме Deep Q-Learning можно узнать из [1].
Логирование важной информации позволяет нам непрерывно обучать RL-агента между эпизодами. Однако необходимо решить еще одну задачу: симуляционная модель выводит только текущее состояние, вознаграждение и информацию о том, закончен ли эпизод (в дальнейшем мы будем говорить об этом, как о флаге DONE), но RL-агенту необходимо предыдущее состояние, чтобы сформировать переход для отправки в буфер воспроизведения. Эта проблема решается путем инициализации предыдущего состояния и значений действий в null. После получения информации о состоянии, вознаграждении и DONE от симуляции состояние станет новым предыдущим состоянием, а действие на основе состояния — новым предыдущим действием. Если значения предыдущего состояния и предыдущего действия не равны null, то в буфер воспроизведения будет добавлен новый переход, состоящий из предыдущего состояния, предыдущего действия, текущего состояния, вознаграждения и DONE.
Простая демонстрация – Упрощенное Taxi-v3
Давайте перейдем непосредственно к реализации. Модель AnyLogic с файлами Python, созданными для этой демонстрации, доступны в этом репозитории.
Среда
В демонстрационных целях мы показываем наш метод на примере упрощенной среды OpenAI Gym Taxi-v3, реплицированной в AnyLogic. Тем не менее, этот метод достаточно стабилен, чтобы его можно было применять к крупномасштабным и гораздо более сложным средам. Возможно, он даже лучше подходит для более сложных сред, потому что в более сложных средах накладными расходами на коммуникацию между AnyLogic и Python по сути можно пренебречь.
Эта среда состоит из сетки 4*4, такси, управляемого RL, и пассажира. Визуализация сетки показана на рисунке 1. Зеленые линии обозначают стены, через которые такси не может проехать. Начальное местоположение пассажира — G, а пункт назначения — Y. Такси будет инициализировано в произвольном месте, отличном от местоположения пассажира. Задача такси — сначала забрать пассажира, а затем высадить его в пункте назначения. Как только пассажир будет высажен или будет сделано более 200 шагов, эпизод завершится. Пространство действий в этой среде: 0 – движение вверх, 1 – движение вниз, 2 – движение влево, 3 – движение вправо, 4 – подобрать пассажира и 5 – высадить. Пространство состояний — это положение такси по оси X, положение такси по оси Y, а также был ли подобран пассажир (0 или 1). Если такси неудачно подобрало или высадило пассажира, оно получает вознаграждение -10. Когда такси успешно высаживает пассажира, оно получает награду +20. Такси получает награду -1, если не срабатывает одно из вышеупомянутых вознаграждений.
Реализация в AnyLogic
В этой модели есть несколько важных функций, которые позволяют обучать RL-агента. Функция f_State возвращает целочисленный список, представляющий текущее состояние. Функция f_Reward возвращает вознаграждение, полученное в результате выполнения действия. Функция f_TaxiAction реализует действие от RL-агента и возвращает, завершился ли эпизод после выполнения этого действия. Если параметр модели deploy установлен в true, функция f_TaxiAction изменит визуализацию в соответствии с действием. Функция f_RLAction вызывает RL-агента для выбора действия в соответствии с текущим состоянием и предоставляет RL-агенту необходимую для обучения информацию с помощью вышеупомянутых трех функций. Во время выполнения симуляции функция вызывается в цикле f_RLAction с интервалом в 0.1 секунду.
Реализация на Python
Для реализации Deep Q-Learning на Python мы используем библиотеку глубокого обучения PyTorch. За исключением нескольких дополнительных строк кода для сохранения и загрузки важной информации для обучения, эта реализация ничем не отличается от других стандартных реализаций Deep Q-Learning. Поскольку основной темой этой статьи не являются алгоритмы обучения с подкреплением, и чтобы не утомлять вас техническими подробностями, в этом разделе будут рассмотрены только те части кода, которые имеют отношение к применению RL в AnyLogic. В данной реализации для обучения с подкреплением выделено два файла Python: Train.py и DQNModel.py. Поскольку задача DQNModel.py заключается только в построении нейронной сети, он не рассматривается в этом блоге.
Здесь следует отметить, что, учитывая, что мы создаем связь между AnyLogic и Python, код Python лучше модулировать, чтобы сделать эту связь проще и чище. Здесь мы создали класс для обучающего агента Deep Q-Learning под названием DQN_Main.
Чтобы инициализировать инстанс класса DQN_Main (это происходит в начале каждого эпизода), нам нужно сначала загрузить необходимую информацию с локального диска с помощью JSON и функции load из PyTorch, затем установить значение предыдущего состояния и предыдущего действия в null, а вознаграждение эпизода — в 0. Информация, необходимая для этого инстанса, выделена красным на рисунке 2.
Затем на каждом шаге действий определяется функция act, которая вызывается из AnyLogic, чтобы поместить опыт в буфер воспроизведения, вызвать функцию train для обучения нейронной сети и, если эпизод завершен, сохранить важную информацию на локальном диске.
После вызова функции act функция train обучает нейронную сеть в течение одной эпохи и сохраняет на локальном диске важную информацию, которая изменилась в процессе обучения, если эпизод завершен.
При желании вы можете добавить функции для генерации графиков вознаграждения и потерь на локальный диск, чтобы вы могли наблюдать за прогрессом вашего RL-агента. Сгенерированные графики для данного инстанса отмечены желтым цветом на рисунке 2.
Ниже прилагается полный код Train.py:
import os.path
import os
import torch
import json
from DQNModel import DQN
import random
import numpy as np
import torch.nn.functional as F
import matplotlib.pyplot as plt
class DQN_Main:
def __init__(self):
self.BUFFER_SIZE = 200000
self.MIN_REPLAY_SIZE = 50000
self.device = torch.device("cuda" if torch.cuda.is_available() else "cpu")
self.GAMMA = 0.99
self.BATCH_SIZE = 128
self.EPSILON_START = 0.99
self.EPSILON_END = 0.1
self.EPSILON_DECAY = 0.000025
self.TARGET_UPDATE_FREQ = 10000
self.LR = 0.00025
os.environ["KMP_DUPLICATE_LIB_OK"] = "TRUE"
if os.path.exists('replay_buffer.json'):
with open("replay_buffer.json", "r") as read_content:
self.replay_buffer = json.load(read_content)
else:
self.replay_buffer = []
if os.path.exists('reward_buffer.json'):
with open("reward_buffer.json", "r") as read_content:
self.reward_buffer = json.load(read_content)
else:
self.reward_buffer = []
if os.path.exists('step.json'):
with open("step.json", "r") as read_content:
self.step = json.load(read_content)
else:
self.step = 0
self.policy_net = DQN(device=self.device).to(self.device)
self.target_net = DQN(device=self.device).to(self.device)
if os.path.exists('policy_net.pth'):
self.policy_net.load_state_dict(torch.load('policy_net.pth'))
self.target_net.load_state_dict(torch.load('target_net.pth'))
else:
self.target_net.load_state_dict(self.policy_net.state_dict())
self.optimizer = torch.optim.Adam(self.policy_net.parameters(), lr=self.LR)
if os.path.exists('policy_net_adam.pth'):
self.optimizer.load_state_dict(torch.load('policy_net_adam.pth'))
if os.path.exists('loss_hist.json'):
with open("loss_hist.json", "r") as read_content:
self.loss_hist = json.load(read_content)
else:
self.loss_hist = []
if os.path.exists('loss_hist_show.json'):
with open("loss_hist_show.json", "r") as read_content:
self.loss_hist_show = json.load(read_content)
else:
self.loss_hist_show = []
self.episode_reward = 0
self.prev_state = None
self.prev_action = None
def save_hyperparams(self):
hyperparams_dict = {
'BUFFER SIZE': self.BUFFER_SIZE,
'MIN REPLAY SIZE': self.MIN_REPLAY_SIZE,
'GAMMA': self.GAMMA,
'BATCH SIZE': self.BATCH_SIZE,
'EPSILON START': self.EPSILON_START,
'EPSILON END': self.EPSILON_END,
'EPSILON DECAY': self.EPSILON_DECAY,
'TARGET UPDATE FREQ': self.TARGET_UPDATE_FREQ,
'LR': self.LR,
}
with open("hyperparameters.json", "w") as write:
json.dump(hyperparams_dict, write)
def train(self, done):
# добавьте сюда шаг обучения
transitions = random.sample(self.replay_buffer, self.BATCH_SIZE)
states = np.asarray([t[0] for t in transitions])
actions = np.asarray([t[1] for t in transitions])
rewards = np.asarray([t[2] for t in transitions])
dones = np.asarray([t[3] for t in transitions])
next_states = np.asarray([t[4] for t in transitions])
states_t = torch.as_tensor(states, dtype=torch.float32).to(self.device)
actions_t = torch.as_tensor(actions, dtype=torch.int64).unsqueeze(-1).to(self.device)
rewards_t = torch.as_tensor(rewards, dtype=torch.float32).unsqueeze(-1).to(self.device)
dones_t = torch.as_tensor(dones, dtype=torch.float32).unsqueeze(-1).to(self.device)
next_states_t = torch.as_tensor(next_states, dtype=torch.float32).to(self.device)
# вычисление цели
_, actions_target = self.policy_net(next_states_t).max(dim=1, keepdim=True)
target_q_values_1 = self.target_net(next_states_t).gather(dim=1, index=actions_target)
targets_1 = rewards_t + self.GAMMA * (1 - dones_t) * target_q_values_1
# вычисление потери
q_values = self.policy_net(states_t)
action_q_values = torch.gather(input=q_values, dim=1, index=actions_t)
# градиентный спуск
loss = F.mse_loss(action_q_values, targets_1)
self.loss_hist.append(loss.item())
self.optimizer.zero_grad()
loss.backward()
self.optimizer.step()
if self.step % 200 == 0:
self.loss_hist_show.append(sum(self.loss_hist[-300:])/300)
self.plot_loss_hist()
# обновление целевой сети
if self.step % self.TARGET_UPDATE_FREQ == 0:
self.target_net.load_state_dict(self.policy_net.state_dict())
# нам нужен параметр done, так как нам нужно сохранить нейросети, если эпизод завершен
if done:
torch.save(self.policy_net.state_dict(), 'policy_net.pth')
torch.save(self.target_net.state_dict(), 'target_net.pth')
torch.save(self.optimizer.state_dict(), 'policy_net_adam.pth')
with open("loss_hist.json", "w") as write:
json.dump(self.loss_hist, write)
with open("loss_hist_show.json", "w") as write:
json.dump(self.loss_hist_show, write)
def random_action(self):
return random.choice(self.policy_net.action_space)
def act(self, state, reward, done, deploy):
if deploy:
with torch.no_grad():
state_t = torch.tensor(state)
action = self.policy_net.act(state_t)
return action
if len(self.replay_buffer) >= self.MIN_REPLAY_SIZE:
rnd = random.random()
epsilon = self.EPSILON_START - self.EPSILON_DECAY * self.step
self.step += 1
if epsilon < self.EPSILON_END:
epsilon = self.EPSILON_END
if rnd <= epsilon:
action = self.random_action()
else:
with torch.no_grad():
state_t = torch.tensor(state)
action = self.policy_net.act(state_t)
else:
# заполнение буфера воспроизведения
action = self.random_action()
if self.prev_state is None:
# начало эпизода, мы просто берем действие, ничего не добавляем в буфер воспроизведения
self.prev_state = state.copy()
self.prev_action = action
# здесь нам все еще нужно обучить нашу нейронную сеть
if len(self.replay_buffer) >= self.MIN_REPLAY_SIZE:
# if done neural nets will be saved in the train function
self.train(done)
return action
else:
# здесь мы добавляем переходы в буфер воспроизведения
self.episode_reward += reward
transition = (self.prev_state, self.prev_action, reward, done, state)
self.replay_buffer.append(transition)
if len(self.replay_buffer) > self.BUFFER_SIZE:
self.replay_buffer.pop(0)
# корректировка предыдущего состояния и действия
self.prev_state = state.copy()
self.prev_action = action
if done:
self.reward_buffer.append(self.episode_reward)
# поскольку мы подключаемся к AnyLogic, нам приходится сохранять все в каждом эпизоде
with open("reward_buffer.json", "w") as write:
json.dump(self.reward_buffer, write)
with open("replay_buffer.json", "w") as write:
json.dump(self.replay_buffer, write)
with open("step.json", "w") as write:
json.dump(self.step, write)
if len(self.reward_buffer)%100 == 0:
self.plot_reward_buffer()
if len(self.replay_buffer) >= self.MIN_REPLAY_SIZE:
# если done, нейронные сети будут сохранены в функции train
self.train(done)
return action
def plot_reward_buffer(self):
plt.plot(self.reward_buffer)
plt.xlabel('Episodes')
plt.ylabel('Rewards')
plt.savefig('reward buffer.jpg')
plt.close()
def plot_loss_hist(self):
plt.plot(self.loss_hist_show[10:])
plt.xlabel('100 Epoch')
plt.ylabel('Loss')
plt.savefig('Loss History.jpg')
plt.close()Обучение с помощью эксперимента AnyLogic
Обучение RL-агента происходит с помощью эксперимента Монте-Карло от AnyLogic. В эксперименте Монте-Карло мы задаем количество эпизодов, на которых мы хотим обучить RL-агента, в разделе Replications, после чего нам остается только сидеть и наблюдать, как RL-агент развивается, получая опыт от симуляции!
Если вы хотите прервать текущее обучение и заново обучить нового RL-агента, вы можете сделать это, удалив все файлы, отмеченные желтым или красным цветом на рисунке 2.
Результат обучения
Результат обучения подтверждает, что наш метод работает не хуже любого другого метода обучения RL-моделей. На рисунке 3 показано, что вознаграждение постоянно улучшается, и модель успешно сходится.
Успех обучения подтверждается визуализацией эксперимента AnyLogic (желтый: такси, красный: пассажир, зеленый: пункт назначения):
Спасибо за внимание! Если у вас возникнут дополнительные вопросы, не стесняйтесь посетить мою страницу на GitHub и оставить свои вопросы в разделе обсуждений! :)
Ссылки:
Mnih, V., Kavukcuoglu, K., Silver, D. et al. Human-level control through deep reinforcement learning. Nature 518, 529–533 (2015).
Научиться создавать имитационные модели в ПО AnyLogic и применять их для анализа проектов можно под руководством экспертов области на онлайн-курсе «Имитационное моделирование на базе AnyLogic».