Введение в обучение с подкреплением: от многорукого бандита до полноценного RL агента

Привет, Хабр! Обучение с подкреплением является одним из самых перспективных направлений машинного обучения. С его помощью искусственный интеллект сегодня способен решать широчайший спектр задач: от робототехники и видеоигр до моделирования поведения покупателей и здравоохранения. В этой вводной статье мы изучим главную идею reinforcement learning и с нуля построим собственного самообучающегося бота.

Введение

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

Если вы привыкли работать с задачами обучения с учителем, то в случае reinforcement learning действует немного иная логика. Вместо того, чтобы создавать алгоритм, который обучается на наборе пар «факторы — правильный ответ», в обучении с подкреплением необходимо научить агента взаимодействовать с окружающей средой, самостоятельно генерируя эти пары. Затем на них же он будет обучаться через систему наблюдений (observations), выигрышей (reward) и действий (actions).

Очевидно, что теперь в каждый момент времени у нас нет постоянного правильного ответа, поэтому задача становится немного хитрее. В этой серии статей мы будем создавать и обучать агентов обучения с подкреплением. Начнем с самого простого варианта агента, чтобы основная идея reinforcement learning была предельно понятна, а затем перейдем к более сложным задачам.

Многорукий бандит

Самый простой пример задачи обучения с подкреплением — задача о многоруком бандите (она достаточно широко освещена на Хабре, в частности, тут и тут). В нашей постановке задачи есть n игровых автоматов, в каждом из которых фиксирована вероятность выигрыша. Тогда цель агента — найти слот-машину с наибольшим ожидаемым выигрышем и всегда выбирать именно ее. Для простоты у нас будет всего четыре игровых автомата, из которых нужно будет выбирать.

По правде говоря, эту задачу можно с натяжкой отнести к reinforcement learning, поскольку задачам из этого класса характерны следующие свойства:

• Разные действия приводят к разным выигрышам. К примеру, при поиске сокровищ в лабиринте поворот налево может означать кучу бриллиантов, а поворот направо — яму ядовитых змей.
• Агент получает выигрыш с задержкой во времени. Это значит, что, повернув налево в лабиринте, мы не сразу поймем, что это правильный выбор.
• Выигрыш зависит от текущего состояния системы. Продолжая пример выше, поворот налево может быть правильным в текущей части лабиринта, но не обязательно в остальных.

В задаче о многоруком бандите нет ни второго, ни третьего условия, что существенно ее упрощает и позволяет нам сконцентрироваться лишь на выявлении оптимального действия из всех возможных вариантов. На языке reinforcement learning это означает найти «правило поведения» (policy). Мы будем использовать метод, называемый policy gradients, при котором нейросеть обновляет свое правило поведения следующим образом: агент совершает действие, получает обратную связь от среды и на ее основе корректирует веса модели через градиентный спуск.

В области обучения с подкреплением есть и другой подход, при котором агенты обучают value functions. Вместо того, чтобы находить оптимальное действие в текущем состоянии, агент учиться предсказывать, насколько выгодно находиться в данном состоянии и совершать данное действие. Оба подхода дают хорошие результаты, однако логика policy gradient более очевидна.

Policy Gradient

Как мы уже выяснили, в нашем случае ожидаемый выигрыш каждого из игровых автоматов не зависит от текущего состояния среды. Получается, что наша нейросеть будет состоять лишь из набора весов, каждый из которых соответствует одному игровому автомату. Эти веса и будут определять, за какую ручку нужно дернуть, чтобы получить максимальный выигрыш. К примеру, если все веса инициализировать равными 1, то агент будет одинаково оптимистичен по поводу выигрыша во всех игровых автоматах.

Для обновления весов модели мы будем использовать e-жадную линию поведения. Это значит, что в большинстве случаев агент будет выбирать действие, максимизирующее ожидаемый выигрыш, однако иногда (с вероятностью равной e) действие будет случайным. Так будет обеспечен выбор всех возможных вариантов, что позволит нейросети «узнать» больше о каждом из них.

Совершив одно из действий, агент получает обратную связь от системы: 1 или -1 в зависимости от того, выиграл ли он. Это значение затем используется для расчета функции потерь:

A (advantage) — важный элемент всех алгоритмов обучения с подкреплением. Он показывает, насколько совершенное действие лучше, чем некий baseline. В дальнейшем мы будем использовать более сложный baseline, а пока примем его равным 0, то есть A будет просто равен награде за каждое действие (1 или -1). п — это правило поведения, вес нейросети, соответствующий ручке слот-машины, которую мы выбрали на текущем шаге.

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

Реализация алгоритма

Бандиты. Сначала мы создадим наших бандитов (в быту игровой автомат называют бандитом). В нашем примере их будет 4. Функция pullBandit генерирует случайное число из стандартного нормального распределения, а затем сравнивает его со значением бандита и возвращает результат игры. Чем дальше по списку находится бандит, тем больше вероятность, что агент выиграет, выбрав именно его. Таким образом, мы хотим, чтобы наш агент научился всегда выбирать последнего бандита.

import tensorflow as tf
import numpy as np

#Список наших бандитов. Бандит №4 наиболее оптимален для выбора.
bandits = [0.2,0,-0.2,-5]
num_bandits = len(bandits)
def pullBandit(bandit):
    #Сгенерировать случайное число
    result = np.random.randn(1)
    if result > bandit:
        #Выигрыш
        return 1
    else:
        #Проигрыш
        return -1

Агент. Кусок кода ниже создает нашего простого агента, который состоит из набора значений для бандитов. Каждое значение соответствует выигрышу/проигрышу в зависимости от выбора того или иного бандита. Чтобы обновлять веса агента мы используем policy gradient, то есть выбираем действия, минимизирующие функцию потерь:

tf.reset_default_graph()

#Эти 2 строчки создают feed-forward часть нейросети. Здесь и происходит выбор действия.
weights = tf.Variable(tf.ones([num_bandits]))
chosen_action = tf.argmax(weights,0)

#Следующие 6 строчек устанавливают процедуру обучения. Нейросеть принимает на вход действие и его результат, чтобы оценить функцию потерь и обновить веса сети.
reward_holder = tf.placeholder(shape=[1],dtype=tf.float32)
action_holder = tf.placeholder(shape=[1],dtype=tf.int32)
responsible_weight = tf.slice(weights,action_holder,[1])
loss = -(tf.log(responsible_weight)*reward_holder)
optimizer = tf.train.GradientDescentOptimizer(learning_rate=0.001)
update = optimizer.minimize(loss)

Обучение агента. Мы будем обучать агента, путем выбора определенных действий и получения выигрышей/проигрышей. Используя полученные значения, мы будем знать, как именно обновить веса модели, чтобы чаще выбирать бандитов с большим ожидаемым выигрышем:

total_episodes = 1000 #Количество итераций обучения
total_reward = np.zeros(num_bandits) #Изначальный выигрыш всех бандитов равен 0
e = 0.1 #Вероятность случайного выбора
init = tf.global_variables_initializer()

#Запускаем граф tensorflow
with tf.Session() as sess:
    sess.run(init)
    i = 0
    while i < total_episodes:
        
        #Выбираем действие либо случайно либо на основе нейросети
        if np.random.rand(1) < e:
            action = np.random.randint(num_bandits)
        else:
            action = sess.run(chosen_action)
        #Получаем результат игры, выбрав одного из бандитов
        reward = pullBandit(bandits[action]) 
        
        #Обновляем веса
        _,resp,ww = sess.run([update,responsible_weight,weights], 
                      feed_dict={reward_holder:[reward],action_holder:[action]})
        
        #Обновляем общий выигрыш каждого бандита
        total_reward[action] += reward
        if i % 50 == 0:
            print("Общий выигрыш бандитов сейчас равен " + str(num_bandits) + 
            " bandits: " + str(total_reward))
        i+=1
print("Агент думает, что бандит №" + str(np.argmax(ww)+1) + " идеален...")
if np.argmax(ww) == np.argmax(-np.array(bandits)):
    print("...и он прав!")
else:
    print("...и он не прав!")

Результат:

Общий выигрыш для 4 бандитов сейчас равен: [-1.  0.  0.  0.]
Общий выигрыш для 4 бандитов сейчас равен: [ -1.  -1.   0.  45.]
Общий выигрыш для 4 бандитов сейчас равен: [ -2.  -1.   1.  91.]
Общий выигрыш для 4 бандитов сейчас равен: [  -3.   -1.    1.  138.]
Общий выигрыш для 4 бандитов сейчас равен: [  -2.    0.    2.  183.]
Общий выигрыш для 4 бандитов сейчас равен: [   0.    0.    2.  229.]
Общий выигрыш для 4 бандитов сейчас равен: [  -1.    0.    3.  277.]
Общий выигрыш для 4 бандитов сейчас равен: [  -1.   -1.    4.  323.]
Общий выигрыш для 4 бандитов сейчас равен: [  -2.   -1.    2.  370.]
Общий выигрыш для 4 бандитов сейчас равен: [  -2.   -2.    3.  416.]
Общий выигрыш для 4 бандитов сейчас равен: [  -2.   -2.    3.  464.]
Общий выигрыш для 4 бандитов сейчас равен: [  -3.   -1.    3.  508.]
Общий выигрыш для 4 бандитов сейчас равен: [  -7.    1.    4.  549.]
Общий выигрыш для 4 бандитов сейчас равен: [  -9.    0.    3.  593.]
Общий выигрыш для 4 бандитов сейчас равен: [  -9.    0.    4.  640.]
Общий выигрыш для 4 бандитов сейчас равен: [  -9.    0.    5.  687.]
Общий выигрыш для 4 бандитов сейчас равен: [  -9.   -2.    5.  735.]
Общий выигрыш для 4 бандитов сейчас равен: [  -9.   -4.    7.  781.]
Общий выигрыш для 4 бандитов сейчас равен: [ -10.   -4.    8.  829.]
Общий выигрыш для 4 бандитов сейчас равен: [ -13.   -4.    7.  875.]
Агент думает, что бандит №4 идеален...
...и он прав!

Полный Jupyter Notebook можно скачать тут.

Решение полноценной задачи обучения с подкреплением

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

Системы, в которых может быть решена обучения с подкреплением называются Марковскими процессами принятия решений (Markov Decision Processes, MDP). Для таких систем характерны выигрыши и действия, обеспечивающие переход из одного состояния в другое, причем эти выигрыши зависят от текущего состояния системы и решения, которое принимает агент в этом состоянии. Выигрыш может быть получен с задержкой во времени.

Формально Марковский процесс принятия решений может быть определен следующим образом. MDP состоит из набора всех возможных состояний S и действий А, причем в каждый момент времени он находится в состоянии s и совершает действие a из этих наборов. Таким образом, дан кортеж (s, a) и для него определены T(s,a) — вероятность перехода в новое состояние s' и R(s,a) — выигрыш. В итоге в любой момент времени в MDP агент находится в состоянии s, принимает решение a и в ответ получает новое состояние s' и выигрыш r.

Для примера, даже процесс открывания двери можно представить в виде Марковского процесса принятия решений. Состоянием будет наш взгляд на дверь, а также расположение нашего тела и двери в мире. Все возможные движения тела, что мы можем сделать, и являются набором A, а выигрыш — это успешное открытие двери. Определенные действия (например, шаг в сторону двери) приближают нас к достижению цели, однако сами по себе не приносят выигрыша, так как его обеспечивает только непосредственно открывание двери. В итоге, агент должен совершать такие действия, которые рано или поздно приведут к решению задачи.

Задача стабилизации перевернутого маятника

Воспользуемся OpenAI Gym — платформой для разработки и тренировки AI ботов с помощью игр и алгоритмических испытаний и возьмем классическую задачу оттуда: задача стабилизации перевернутого маятника или Cart-Pole. В нашем случае суть задачи заключается в том, чтобы как можно дольше удерживать стержень в вертикальном положении, двигая тележку по горизонтали:


В отличии от задачи о многоруком бандите, в данной системе есть:

• Наблюдения. Агент должен знать, где стержень находится сейчас и под каким углом. Это наблюдение нейросеть будет использовать для оценки вероятности того или иного действия.
• Отсроченный выигрыш. Необходимо двигать тележку таким образом, чтобы это было выгодно как на данный момент, так и в будущем. Для этого будем сопоставлять пару «наблюдение — действие» со скорректированным значением выигрыша. Корректировка осуществляется функцией, которая взвешивает действия по времени.

Чтобы учитывать задержку выигрыша во времени, нам нужно использовать policy gradient метод с некоторыми поправками. Во-первых, теперь необходимо обновлять агента, который имеет более одного наблюдения в единицу времени. Для этого все наблюдения мы будем собирать в буфер, а затем использовать их одновременно, чтобы обновить веса модели. Этот набор наблюдений за единицу времени затем сопоставляется с дисконтированным выигрышем.

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

Реализация алгоритма

Импортируем библиотеки и загрузим среду задачи Cart-Pole:

import tensorflow as tf
import tensorflow.contrib.slim as slim
import numpy as np
import gym
import matplotlib.pyplot as plt
%matplotlib inline

try:
    xrange = xrange
except:
    xrange = range

env = gym.make('CartPole-v0') #загружаем среду задачи

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

gamma = 0.99 # коэффициент дисконтирования
 
def discount_rewards(r):
    """ принимая на вход вектор выигришей, 
    вернуть вектор дисконтированных выигрышей"""
    discounted_r = np.zeros_like(r)
    running_add = 0
    for t in reversed(xrange(0, r.size)):
        running_add = running_add * gamma + r[t]
        discounted_r[t] = running_add
    return discounted_r

Теперь создадим нашего агента:

class agent():
    def __init__(self, lr, s_size,a_size,h_size):
        #Ниже инициализирована feed-forward часть нейросети. 
        #Агент оценивает состояние среды и совершает действие
        self.state_in= tf.placeholder(shape=[None,s_size],dtype=tf.float32)
        hidden = slim.fully_connected(self.state_in,h_size,
                      biases_initializer=None,activation_fn=tf.nn.relu)
        self.output = slim.fully_connected(hidden,a_size,
                      activation_fn=tf.nn.softmax,biases_initializer=None)
        self.chosen_action = tf.argmax(self.output,1) # выбор действия

        #Следующие 6 строк устанавливают процедуру обучения. 
        #Нейросеть принимает на вход выбранное действие
        # и соответствующий выигрыш,
        #чтобы оценить функцию потерь и обновить веса модели.
        self.reward_holder = tf.placeholder(shape=[None],dtype=tf.float32)
        self.action_holder = tf.placeholder(shape=[None],dtype=tf.int32)
        
        self.indexes = tf.range(0, 
        tf.shape(self.output)[0])*tf.shape(self.output)[1] + self.action_holder

        self.responsible_outputs = tf.gather(tf.reshape(self.output, [-1]), 
        self.indexes)
        #функция потерь
        self.loss = -tf.reduce_mean(tf.log(self.responsible_outputs)*
        self.reward_holder) 
        
        tvars = tf.trainable_variables()
        self.gradient_holders = []
        for idx,var in enumerate(tvars):
            placeholder = tf.placeholder(tf.float32,name=str(idx)+'_holder')
            self.gradient_holders.append(placeholder)
        
        self.gradients = tf.gradients(self.loss,tvars)
        
        optimizer = tf.train.AdamOptimizer(learning_rate=lr) 
        self.update_batch = optimizer.apply_gradients(zip(self.gradient_holders,
        tvars))

Обучение агента. Теперь, наконец, перейдем к обучению агента:

tf.reset_default_graph() #Очищаем граф tensorflow

myAgent = agent(lr=1e-2,s_size=4,a_size=2,h_size=8) #Инициализируем агента

total_episodes = 5000 #Количество итераций обучения
max_ep = 999
update_frequency = 5

init = tf.global_variables_initializer()

#Запуск графа tensorflow
with tf.Session() as sess:
    sess.run(init)
    i = 0
    total_reward = []
    total_lenght = []
        
    gradBuffer = sess.run(tf.trainable_variables())
    for ix,grad in enumerate(gradBuffer):
        gradBuffer[ix] = grad * 0
        
    while i < total_episodes:
        s = env.reset()
        running_reward = 0
        ep_history = []
        for j in range(max_ep):
            #Выбрать действие на основе вероятностей, оцененных нейросетью
            a_dist = sess.run(myAgent.output,feed_dict={myAgent.state_in:[s]})
            a = np.random.choice(a_dist[0],p=a_dist[0])
            a = np.argmax(a_dist == a)

            s1,r,d,_ = env.step(a) #Получить награду за совершенное действие
            ep_history.append([s,a,r,s1])
            s = s1
            running_reward += r
            if d == True:
                #Обновить нейросеть
                ep_history = np.array(ep_history)
                ep_history[:,2] = discount_rewards(ep_history[:,2])
                feed_dict = {myAgent.reward_holder:ep_history[:,2],
                             myAgent.action_holder:ep_history[:,1],
                             myAgent.state_in:np.vstack(ep_history[:,0])}
                grads = sess.run(myAgent.gradients, feed_dict=feed_dict)
                for idx,grad in enumerate(grads):
                    gradBuffer[idx] += grad

                if i % update_frequency == 0 and i != 0:
                    feed_dict = dictionary = dict(zip(myAgent.gradient_holders, 
                                     gradBuffer))
                    _ = sess.run(myAgent.update_batch, feed_dict=feed_dict)
                    for ix,grad in enumerate(gradBuffer):
                        gradBuffer[ix] = grad * 0
                
                total_reward.append(running_reward)
                total_lenght.append(j)
                break

        
            #Обновить общий выигрыш
        if i % 100 == 0:
            print(np.mean(total_reward[-100:]))
        i += 1

Результат:

16.0
21.47
25.57
38.03
43.59
53.05
67.38
90.44
120.19
131.75
162.65
156.48
168.18
181.43

Полный Jupyter Notebook вы можете посмотреть тут. Увидимся в следующих статьях, где мы продолжим изучать обучение с подкреплением!