Выстраиваем стабильное соединение для обучения с подкреплением на Python на моделях AnyLogic

Введение

Стремительное развитие глубокого обучения с подкреплением (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 и оставить свои вопросы в разделе обсуждений! 🙂

Ссылки:

1. Mnih, V., Kavukcuoglu, K., Silver, D. et al. Human-level control through deep reinforcement learning. Nature 518, 529–533 (2015).

---

Научиться создавать имитационные модели в ПО AnyLogic и применять их для анализа проектов можно под руководством экспертов области на онлайн-курсе «Имитационное моделирование на базе AnyLogic».