Обучение с подкреплением является (Reinforcement learning) одним из направлений ML. Суть этого метода заключается в том, что обучаемая система или агент учится принимать оптимальные решения через взаимодействие со средой. В отличие от других подходов, Reinforcement learning (RL) не требует заранее подготовленных данных с правильными ответами или явной структуры в них. 

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

В этой статье мы попробуем разобраться с тем, как работает Q‑обучение, а также посмотрим небольшой пример на Python.

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

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

Но давайте сначала рассмотрим очень простую реализацию Q‑обучения на Python — задача не из лёгких, поскольку большинство примеров в интернете слишком сложны для новичков.

import numpy as np
import pylab as plt

# map cell to cell, add circular cell to goal point
points_list = [(0,1), (1,5), (5,6), (5,4), (1,2), (2,3), (2,7)]

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

Итак, наша начальная точка это 0, наша конечная точка — 7.

Построим карту, которую необходимо будет пройти нашей модели.

goal = 7

import networkx as nx
G=nx.Graph()
G.add_edges_from(points_list)
pos = nx.spring_layout(G)
nx.draw_networkx_nodes(G,pos)
nx.draw_networkx_edges(G,pos)
nx.draw_networkx_labels(G,pos)
plt.show()

Карта показывает, что 0 — это точка, откуда наш бот начнёт свой путь, а точка 7 — его конечная цель. Дополнительные точки и ложные пути — это препятствия, с которыми боту придётся столкнуться.

Затем мы создаём график наград — это матричная версия нашей карты списка точек. Мы инициализируем матрицу высотой и шириной нашего списка точек (в данном примере 8) и инициализируем все значения значением -1:

# сколько точек в графике? x точек
MATRIX_SIZE = 8

# создать матрицу x*y
R = np.matrix(np.ones(shape=(MATRIX_SIZE, MATRIX_SIZE)))
R *= -1

Затем мы меняем значения на 0, если путь не подходит, и 100, если это путь к цели.

# присваиваем нули путям и 100 точке, достигающей цели.

for point in points_list:
    print(point)
    if point[1] == goal:
        R[point] = 100
    else:
        R[point] = 0

    if point[0] == goal:
        R[point[::-1]] = 100
    else:
        # reverse of point
        R[point[::-1]]= 0

# add goal point round trip
R[goal,goal]= 100

R
matrix([[  -1.,    0.,   -1.,   -1.,   -1.,   -1.,   -1.,   -1.],
        [   0.,   -1.,    0.,   -1.,   -1.,    0.,   -1.,   -1.],
        [  -1.,    0.,   -1.,    0.,   -1.,   -1.,   -1.,  100.],
        [  -1.,   -1.,    0.,   -1.,   -1.,   -1.,   -1.,   -1.],
        [  -1.,   -1.,   -1.,   -1.,   -1.,    0.,   -1.,   -1.],
        [  -1.,    0.,   -1.,   -1.,    0.,   -1.,    0.,   -1.],
        [  -1.,   -1.,   -1.,   -1.,   -1.,    0.,   -1.,   -1.],
        [  -1.,   -1.,    0.,   -1.,   -1.,   -1.,   -1.,  100.]])
 

В приведенной выше матрице ось Y отображает текущее состояние вашего бота, а ось X — ваши возможные дальнейшие действия. Затем мы строим матрицу Q‑обучения, которая будет содержать все уроки, полученные нашим ботом. Модель Q‑обучения использует формулу переходного правила, а гамма — параметр обучения.

Q = np.matrix(np.zeros([MATRIX_SIZE,MATRIX_SIZE]))

# learning parameter
gamma = 0.8

initial_state = 1

def available_actions(state):
    current_state_row = R[state,]
    av_act = np.where(current_state_row >= 0)[1]
    return av_act

available_act = available_actions(initial_state) 

def sample_next_action(available_actions_range):
    next_action = int(np.random.choice(available_act,1))
    return next_action

action = sample_next_action(available_act)

def update(current_state, action, gamma):
    
  max_index = np.where(Q[action,] == np.max(Q[action,]))[1]
  
  if max_index.shape[0] > 1:
      max_index = int(np.random.choice(max_index, size = 1))
  else:
      max_index = int(max_index)
  max_value = Q[action, max_index]
  
  Q[current_state, action] = R[current_state, action] + gamma * max_value
  print('max_value', R[current_state, action] + gamma * max_value)
  
  if (np.max(Q) > 0):
    return(np.sum(Q/np.max(Q)*100))
  else:
    return (0)
    
update(initial_state, action, gamma)

Далее запускаем функции обучения и тестирования, которые выполнят функцию обновления 700 раз, что позволит модели Q‑обучения определить наиболее эффективный путь:

# Training
scores = []
for i in range(700):
    current_state = np.random.randint(0, int(Q.shape[0]))
    available_act = available_actions(current_state)
    action = sample_next_action(available_act)
    score = update(current_state,action,gamma)
    scores.append(score)
    print ('Score:', str(score))
    
print("Trained Q matrix:")
print(Q/np.max(Q)*100)

# Testing
current_state = 0
steps = [current_state]

while current_state != 7:

    next_step_index = np.where(Q[current_state,] == np.max(Q[current_state,]))[1]
    
    if next_step_index.shape[0] > 1:
        next_step_index = int(np.random.choice(next_step_index, size = 1))
    else:
        next_step_index = int(next_step_index)
    
    steps.append(next_step_index)
    current_state = next_step_index

print("Most efficient path:")
print(steps)

plt.plot(scores)
plt.show()

Most efficient path:
[0, 1, 2, 7]

Мы видим, что модель правильно нашла наиболее эффективный путь от начального узла 0 до конечного узла 7 и потребовала около 400 итераций для сходимости к решению.

Пчелы и дым

Давайте сделаем шаг вперёд и усложним нашу задачу: снова взглянем на верхнее изображение и заметим, что фабрика окружена дымом, а улей — пчёлами. Предположим, пчёлы не любят дым и фабрики, поэтому рядом с дымом никогда не будет ни улья, ни пчёл. Что, если бы наш бот мог регистрировать эти факторы окружающей среды и превращать их в практические выводы? Обнаружив дым, бот может сразу же вернуться обратно, а не продолжать путь к фабрике. Обнаружив пчёл, он может остаться и предположить, что улей закрыт.

bees = [2]
smoke = [4,5,6]

G=nx.Graph()
G.add_edges_from(points_list)
mapping={0:'Start', 1:'1', 2:'2 - Bees', 3:'3', 4:'4 - Smoke', 5:'5 - Smoke', 6:'6 - Smoke', 7:'7 - Beehive'} 
H=nx.relabel_nodes(G,mapping) 
pos = nx.spring_layout(H)
nx.draw_networkx_nodes(H,pos, node_size=[200,200,200,200,200,200,200,200])
nx.draw_networkx_edges(H,pos)
nx.draw_networkx_labels(H,pos)
plt.show()

Мы назначаем узлу 2 наличие пчёл, а узлам 4, 5 и 6 — наличие дыма. Наш бот Q‑обучения пока не знает, есть ли там пчёлы или дым, а также не знает, что пчёлы полезны для поиска ульев, а дым — нет. Боту нужно выполнить ещё один запуск, как мы только что сделали, но на этот раз ему нужно собрать данные об окружающей среде.

Вот новая функция обновления с возможностью обновления результатов Q‑обучения при обнаружении пчёл или дыма.

# re-initialize the matrices for new run
Q = np.matrix(np.zeros([MATRIX_SIZE,MATRIX_SIZE]))

enviro_bees = np.matrix(np.zeros([MATRIX_SIZE,MATRIX_SIZE]))
enviro_smoke = np.matrix(np.zeros([MATRIX_SIZE,MATRIX_SIZE]))
 
initial_state = 1

def available_actions(state):
    current_state_row = R[state,]
    av_act = np.where(current_state_row >= 0)[1]
    return av_act
 
def sample_next_action(available_actions_range):
    next_action = int(np.random.choice(available_act,1))
    return next_action

def collect_environmental_data(action):
    found = []
    if action in bees:
        found.append('b')

    if action in smoke:
        found.append('s')
    return (found)
 
available_act = available_actions(initial_state) 
 
action = sample_next_action(available_act)

def update(current_state, action, gamma):
  max_index = np.where(Q[action,] == np.max(Q[action,]))[1]
  
  if max_index.shape[0] > 1:
      max_index = int(np.random.choice(max_index, size = 1))
  else:
      max_index = int(max_index)
  max_value = Q[action, max_index]
  
  Q[current_state, action] = R[current_state, action] + gamma * max_value
  print('max_value', R[current_state, action] + gamma * max_value)
  
  environment = collect_environmental_data(action)
  if 'b' in environment: 
    enviro_bees[current_state, action] += 1
  
  if 's' in environment: 
    enviro_smoke[current_state, action] += 1
  
  if (np.max(Q) > 0):
    return(np.sum(Q/np.max(Q)*100))
  else:
    return (0)

update(initial_state,action,gamma)

scores = []
for i in range(700):
    current_state = np.random.randint(0, int(Q.shape[0]))
    available_act = available_actions(current_state)
    action = sample_next_action(available_act)
    score = update(current_state,action,gamma)

# print environmental matrices
print('Bees Found')
print(enviro_bees)
print('Smoke Found')
print(enviro_smoke)
Bees Found
[[  0.   0.   0.   0.   0.   0.   0.   0.]
 [  0.   0.  21.   0.   0.   0.   0.   0.]
 [  0.   0.   0.   0.   0.   0.   0.   0.]
 [  0.   0.  93.   0.   0.   0.   0.   0.]
 [  0.   0.   0.   0.   0.   0.   0.   0.]
 [  0.   0.   0.   0.   0.   0.   0.   0.]
 [  0.   0.   0.   0.   0.   0.   0.   0.]
 [  0.   0.  41.   0.   0.   0.   0.   0.]]

Smoke Found
[[  0.   0.   0.   0.   0.   0.   0.   0.]
 [  0.   0.   0.   0.   0.  30.   0.   0.]
 [  0.   0.   0.   0.   0.   0.   0.   0.]
 [  0.   0.   0.   0.   0.   0.   0.   0.]
 [  0.   0.   0.   0.   0.  90.   0.   0.]
 [  0.   0.   0.   0.  23.   0.  29.   0.]
 [  0.   0.   0.   0.   0.  92.   0.   0.]
 [  0.   0.   0.   0.   0.   0.   0.   0.]]

Матрицы среды показывают, сколько пчёл и дыма бот обнаружил во время своего путешествия в поисках наиболее эффективного пути к улью. Чтобы упростить этот пример, я предполагаю два факта: мы смоделировали данные среды и обнаружили, что пчёлы имеют положительный коэффициент обнаружения ульев, а дым — отрицательный. Мы собираемся повторно использовать матрицу среды, уже составленную для нашего ландшафта. Более реалистичный подход заключался бы в динамическом анализе новой среды и назначении экологических отклонений по мере их обнаружения.

Q = np.matrix(np.zeros([MATRIX_SIZE,MATRIX_SIZE]))

# subtract bees with smoke, this gives smoke a negative effect
enviro_matrix = enviro_bees - enviro_smoke

# Get available actions in the current state
available_act = available_actions(initial_state) 

# Sample next action to be performed
action = sample_next_action(available_act)

# This function updates the Q matrix according to the path selected and the Q 
# learning algorithm
def update(current_state, action, gamma):
    
    max_index = np.where(Q[action,] == np.max(Q[action,]))[1]

    if max_index.shape[0] > 1:
        max_index = int(np.random.choice(max_index, size = 1))
    else:
        max_index = int(max_index)
    max_value = Q[action, max_index]

    Q[current_state, action] = R[current_state, action] + gamma * max_value
    print('max_value', R[current_state, action] + gamma * max_value)

    environment = collect_environmental_data(action)
    if 'b' in environment: 
        enviro_matrix[current_state, action] += 1
    if 's' in environment: 
        enviro_matrix[current_state, action] -= 1

    return(np.sum(Q/np.max(Q)*100))

update(initial_state,action,gamma)

enviro_matrix_snap = enviro_matrix.copy()

def available_actions_with_enviro_help(state):
    current_state_row = R[state,]
    av_act = np.where(current_state_row >= 0)[1]
    # if there are multiple routes, dis-favor anything negative
    env_pos_row = enviro_matrix_snap[state,av_act]
    if (np.sum(env_pos_row < 0)):
        # can we remove the negative directions from av_act?
        temp_av_act = av_act[np.array(env_pos_row)[0]>=0]
        if len(temp_av_act) > 0:
            print('going from:',av_act)
            print('to:',temp_av_act)
            av_act = temp_av_act
    return av_act

# Training
scores = []
for i in range(700):
    current_state = np.random.randint(0, int(Q.shape[0]))
    available_act = available_actions_with_enviro_help(current_state)
    action = sample_next_action(available_act)
    score = update(current_state,action,gamma)
    scores.append(score)
    print ('Score:', str(score))
 

plt.plot(scores)
plt.show()

Заключение

Мы видим, что бот сходится за меньшее количество попыток, скажем, примерно на 100, чем наша исходная модель. Это не контролируемая среда для сравнения двух подходов, а скорее стимулирование размышлений о различных способах применения обучения с подкреплением для исследования…

Если вам интересны методы обучения с подкреплением и вы хотели бы разбираться в них глубже, чем на уровне простых примеров, обратите внимание на курс Otus Reinforcement Learning. На странице курса можно записаться на бесплатные уроки.

А тем, кто настроен на серьезное системное обучение, рекомендуем рассмотреть Подписку — выбираете курсы под свои задачи, экономите на обучении, получаете профессиональный рост. Узнать подробнее