Привет, Хабр!

Actor-Critic — это класс алгоритмов в RL, суть которого довольно проста на словах, он сочетает в себе такие политики как policy-based и оценки value-based. У нас есть два главных действующих лица: Actor и Critic. Actor отвечает за выбор действий, т.е формирование политики поведения, он принимает решения исходя из текущего состояния окружающей среды. Critic оценивает, насколько хорошо или плохо Actor справляется со своей задачей, предоставляя обратную связь через оценку действий Actor'a.

Начнем с Actor

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

Реализуем просто Actor, который использует нейронную сеть для генерации действий, сделаем это с помощью TensorFlow:

import tensorflow as tf
from tensorflow.keras import layers

class SimpleActorModel(tf.keras.Model):
    def __init__(self, action_size):
        super(SimpleActorModel, self).__init__()
        self.dense1 = layers.Dense(128, activation="relu")
        self.dense2 = layers.Dense(128, activation="relu")
        self.output_action = layers.Dense(action_size, activation="softmax")

    def call(self, state):
        x = self.dense1(state)
        x = self.dense2(x)
        return self.output_action(x)

# используем
action_size = 2 # предположим, есть 2 возможных действия
model = SimpleActorModel(action_size)

state = tf.constant([[0.1, 0.2, 0.3]]) # пример состояния окружающей среды
action_probabilities = model(state)
print("Распределение вероятностей действий:", action_probabilities.numpy())

SimpleActorModel принимает на вход состояние окружающей среды, в нашем случае это вектор и выводит распределение вероятностей по возможным действиям. Здесь используется два скрытых слоя с активацией ReLU для обработки состояния, а на выходе — softmax, чтобы получить вероятности для каждого действия.

Critic

Critic оценивает, насколько хорошо или плохо выбранное действие Actor'а с точки зрения достижения конечной цели. Это делается путем расчета value function, которая оценивает будущие награды, получаемые за определенное действие или находясь в определенном состоянии. Critic пытается предсказать, какое вознаграждение агент может ожидать в долгосрочной перспективе, принимая во внимание текущее состояние и предпринимаемое действие.

Для реализации Critic можно использовать можно использовать TensorFlow:

import tensorflow as tf
from tensorflow.keras import layers

class SimpleCriticModel(tf.keras.Model):
    def __init__(self):
        super(SimpleCriticModel, self).__init__()
        self.dense1 = layers.Dense(128, activation="relu")
        self.dense2 = layers.Dense(128, activation="relu")
        self.value = layers.Dense(1)

    def call(self, state):
        x = self.dense1(state)
        x = self.dense2(x)
        return self.value(x)

# пример использования
model = SimpleCriticModel()

state = tf.constant([[0.1, 0.2, 0.3]]) # пример состояния окружающей среды
value = model(state)
print("Предсказанная ценность состояния:", value.numpy())

SimpleCriticModel анализирует текущее состояние среды и возвращает оценку ценности этого состояния, Critic выдает числовое значение, представляющее ожидаемую награду.

Actor-Critic алгоритм

И вот Actor-Critic объединяет два компонента.

Actor выбирает действие на основе текущего состояния среды.

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

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

Critic также обновляет свои параметры на основе полученного вознаграждения и разницы между ожидаемым и полученным вознаграждениями, чтобы в будущем делать более точные предсказания.

Примеры реализации

Базовый пример с PyTorch:

import torch
import torch.nn as nn
import torch.optim as optim

class Actor(nn.Module):
    def __init__(self, input_dim, output_dim):
        super(Actor, self).__init__()
        self.linear = nn.Linear(input_dim, output_dim)
    
    def forward(self, state):
        return torch.softmax(self.linear(state), dim=-1)

class Critic(nn.Module):
    def __init__(self, input_dim):
        super(Critic, self).__init__()
        self.linear = nn.Linear(input_dim, 1)
    
    def forward(self, state):
        return self.linear(state)

def train(actor, critic, state, action, reward, next_state, done):
    optimizer_actor = optim.Adam(actor.parameters(), lr=1e-3)
    optimizer_critic = optim.Adam(critic.parameters(), lr=1e-3)
    
    # Critic update
    value = critic(state)
    next_value = critic(next_state)
    td_error = reward + (1 - done) * 0.99 * next_value - value
    critic_loss = td_error.pow(2)

    optimizer_critic.zero_grad()
    critic_loss.backward()
    optimizer_critic.step()

    # Actor update
    log_prob = torch.log(actor(state)[action])
    actor_loss = -log_prob * td_error.detach()

    optimizer_actor.zero_grad()
    actor_loss.backward()
    optimizer_actor.step()

Реализация на TensorFlow:

import tensorflow as tf
from tensorflow.keras import layers

class ActorCritic(tf.keras.Model):
    def __init__(self, num_actions):
        super(ActorCritic, self).__init__()
        self.common = layers.Dense(128, activation='relu')
        self.actor = layers.Dense(num_actions, activation='softmax')
        self.critic = layers.Dense(1)

    def call(self, inputs):
        x = self.common(inputs)
        return self.actor(x), self.critic(x)

model = ActorCritic(num_actions=4)
optimizer = tf.keras.optimizers.Adam(lr=0.01)

def train_step(state, action, reward, next_state, done):
    with tf.GradientTape() as tape:
        action_probs, critic_value = model(state)
        _, critic_value_next = model(next_state)
        action_log_probs = tf.math.log(action_probs[0, action])
        td_error = reward + 0.99 * critic_value_next * (1 - done) - critic_value
        actor_loss = -action_log_probs * tf.stop_gradient(td_error)
        critic_loss = td_error**2

    grads = tape.gradient([actor_loss, critic_loss], model.trainable_variables)
    optimizer.apply_gradients(zip(grads, model.trainable_variables))

TensorFlow и Keras с Functional API:

import tensorflow as tf
from tensorflow.keras.layers import Input, Dense
from tensorflow.keras.models import Model

def create_actor(input_shape, output_shape):
    inputs = Input(shape=input_shape)
    x = Dense(64, activation='relu')(inputs)
    outputs = Dense(output_shape, activation='softmax')(x)
    model = Model(inputs, outputs)
    return model


def create_critic(input_shape):
    inputs = Input(shape=input_shape)
    x = Dense(64, activation='relu')(inputs)
    outputs = Dense(1)(x)
    model = Model(inputs, outputs)
    return model

actor = create_actor(input_shape=(4,), output_shape=2)
critic = create_critic(input_shape=(4,))

PyTorch с параллельным обновленеим:

import torch
import torch.nn as nn
import torch.optim as optim

class ActorCritic(nn.Module):
    def __init__(self, input_dim, action_dim):
        super(ActorCritic, self).__init__()
        self.common = nn.Linear(input_dim, 64)
        self.actor = nn.Linear(64, action_dim)
        self.critic = nn.Linear(64, 1)
    
    def forward(self, state):
        x = torch.relu(self.common(state))
        return torch.softmax(self.actor(x), dim=-1), self.critic(x)

def train(model, state, action, reward, next_state, done):
    optimizer = optim.Adam(model.parameters(), lr=5e-4)
    
    # Compute loss
    action_probs, value = model(state)
    _, next_value = model(next_state)
    td_error = reward + (1 - done) * 0.99 * next_value - value
    action_log_probs = torch.log(action_probs[action])
    actor_loss = -action_log_probs * td_error.detach()
    critic_loss = td_error.pow(2)

    # Optimize the model
    optimizer.zero_grad()
    (actor_loss + critic_loss).backward()
    optimizer.step()

Advantage Actor-Critic

В Advantage Actor-Critic акцент делается на балансе между оценкой текущих действий actor и критической оценкой этих действий critic, используя функцию преимущества.

(Advantage Function, A(s,a)) вычисляет разницу между ожидаемой наградой за выбранное действие и средней ожидаемой наградой в данном состоянии. Формально, A(s,a) = Q(s,a) - V(s), где Q(s,a) — это ожидаемая награда за действие a в состоянии s, а V(s) — это ожидаемая награда в состоянии s, не зависящая от действия. Это позволяет actor обновлять политику в направлении увеличения преимущества, то есть выбирать действия, которые работают лучше, чем среднее, для данного состояния.

Asynchronous Advantage Actor-Critic (A3C): один из наиболее известных вариантов A2C, где несколько акторов параллельно исследуют пространство и обновляют глобальную модель асинхронно, пример:

import torch
import torch.nn as nn
import torch.optim as optim
import numpy as np
import threading
import gym

class ActorCritic(nn.Module):
    def __init__(self, num_inputs, num_actions):
        super(ActorCritic, self).__init__()
        self.critic = nn.Linear(num_inputs, 1)
        self.actor = nn.Linear(num_inputs, num_actions)
    
    def forward(self, x):
        value = self.critic(x)
        probs = torch.softmax(self.actor(x), dim=-1)
        return probs, value

def worker(global_model, optimizer, env_name, global_results, lock, worker_id):
    local_model = ActorCritic(num_inputs, num_actions)
    local_model.load_state_dict(global_model.state_dict())
    
    env = gym.make(env_name)
    state = env.reset()
    
    while True:
        # код для выполнения действий агентом и обновления модели
        
        with lock:
            global_model.load_state_dict(local_model.state_dict())

# глобал инициализация и запуск воркеров
global_model = ActorCritic(num_inputs, num_actions)
optimizer = optim.Adam(global_model.parameters())
workers = [threading.Thread(target=worker, args=(global_model, optimizer, 'CartPole-v1', global_results, lock, i)) for i in range(num_workers)]
for w in workers:
    w.start()
for w in workers:
    w.join()

Synchronous Advantage Actor-Critic (Synchronous A2C или A2C): в отличие от A3C, A2C обновляет глобальную модель синхронно, используя средние значения градиентов от всех акторов, это в какой-то мере уменьшает дисперсию обновлений, хотя потенциально может быть медленнее по сравнению с асинхронным подходом. Пример:

import torch
import torch.nn as nn
import torch.optim as optim
import numpy as np
import gym

class ActorCritic(nn.Module):
    # конструктор и forward метод аналогичны A3C

def update_global(optimizer, global_model, loss):
    optimizer.zero_grad()
    loss.backward()
    optimizer.step()

env = gym.make('CartPole-v1')
global_model = ActorCritic(num_inputs, num_actions)
optimizer = optim.Adam(global_model.parameters())

# синхронное обновление
for episode in range(total_episodes):
    state = env.reset()
    done = False
    while not done:
        # выполнение действий и сбор градиентов для обновления
        loss = compute_loss(...)  # функция вычисления потерь на основе действий и оценок
        update_global(optimizer, global_model, loss)

Proximal Policy Optimization (PPO): хотя не является прямым вариантом A2C, PPO развивает идеи Actor-Critic методов, вводя функцию потерь, которая ограничивает изменения в политике:

import torch
import torch.nn as nn
import torch.optim as optim
import numpy as np
import gym

class ActorCritic(nn.Module):
    def __init__(self, num_inputs, num_actions):
        super(ActorCritic, self).__init__()
        self.critic = nn.Linear(num_inputs, 1)
        self.actor = nn.Linear(num_inputs, num_actions)
    
    def forward(self, x):
        value = self.critic(x)
        probs = torch.softmax(self.actor(x), dim=-1)
        return probs, value

def ppo_loss(old_log_probs, new_log_probs, advantages, clip_param=0.2):
    ratios = torch.exp(new_log_probs - old_log_probs)
    surr1 = ratios * advantages
    surr2 = torch.clamp(ratios, 1.0 - clip_param, 1.0 + clip_param) * advantages
    return -torch.min(surr1, surr2).mean()

def compute_advantages(rewards, values, gamma=0.99, tau=0.95):
    advantages = torch.zeros_like(rewards)
    gae = 0
    for t in reversed(range(len(rewards))):
        delta = rewards[t] + gamma * values[t + 1] - values[t]
        gae = delta + gamma * tau * gae
        advantages[t] = gae
    return advantages

# инициализация среды, модели и оптимизатора
env = gym.make('CartPole-v1')
num_inputs = env.observation_space.shape[0]
num_actions = env.action_space.n
model = ActorCritic(num_inputs, num_actions)
optimizer = optim.Adam(model.parameters(), lr=1e-3)

for epoch in range(10):  # Пример количества эпох обучения
    # реализация сбора данных среды, выполнения действий и т.д.
    
    # предполагается, что мы уже собрали данные о состояниях, действиях, наградах
    # states, actions, rewards, next_states, dones
    
    # вычисление логарифмов вероятностей, преимуществ, функции потерь и выполнение шага оптимизации
    old_log_probs = ...  # логарифмы старых вероятностей действий
    new_log_probs = ...  # ллогарифмы новых вероятностей действий, полученных из модели
    advantages = ...  # преимущества, вычисленные с использованием функции compute_advantages
    
    loss = ppo_loss(old_log_probs, new_log_probs, advantages)
    optimizer.zero_grad()
    loss.backward()
    optimizer.step()

В завершение хочу пригласить вас на бесплатный вебинар про применение фреймворка FinRL для моделирования торгового агента. Регистрация доступна по ссылке.