Otimização de Política Próxima com PyTorch e Gym

1. Configurando o Ambiente 

Antes de implementar o PPO, precisamos instalar as bibliotecas de software pré-requisitas e escolher um ambiente adequado para aplicar a política. 

Instalando o PyTorch e as bibliotecas necessárias

Precisamos instalar o seguinte software:

• O PyTorch e outras bibliotecas de software, como numpy (para funções matemáticas e estatísticas) e matplotlib (para plotar gráficos).
• O pacote de software Gym de código aberto da OpenAI, uma biblioteca Python que simula diferentes ambientes e jogos, que podem ser resolvidos usando Aprendizado por Reforço. Você pode usar a API do Gym para fazer seu algoritmo interagir com o ambiente. Como a funcionalidade do gym às vezes muda durante o processo de atualização, neste exemplo, congelamos sua versão para 0.25.2.

Para instalar em um servidor ou máquina local, execute:

$ pip install torch numpy matplotlib gym==0.25.2

Para instalar usando um Notebook como Google Colab ou DataLab, use: 

!pip install torch numpy matplotlib gym==0.25.2

Criar o(s) ambiente(s) CartPole

Use OpenAI Gym para criar duas instâncias (uma para treinamento e outra para teste) do ambiente CartPole:

env_train = gym.make('CartPole-v1')
env_test = gym.make('CartPole-v1')

2. Implementando PPO em PyTorch

Agora, vamos implementar o PPO usando PyTorch.

Definindo a rede de política

Como explicado anteriormente, o PPO é implementado como um modelo ator-crítico. O ator implementa a política, e o crítico prevê seu valor estimado. Tanto as redes neurais do ator quanto do crítico recebem a mesma entrada – o estado em cada intervalo de tempo. Assim, os modelos do ator e do crítico podem compartilhar uma rede neural comum, que é chamada de arquitetura de backbone. O ator e o crítico podem estender a arquitetura de backbone com camadas adicionais. 

Defina a rede de backbone

Os seguintes passos descrevem a rede de backbone:

• Implemente uma rede com 3 camadas – uma camada de entrada, uma oculta e uma de saída. 
• Após as camadas de entrada e oculta, usamos uma função de ativação. Neste tutorial, escolhemos ReLU porque é computacionalmente eficiente.
• Também impomos uma função de dropout após as camadas de entrada e ocultas para obter uma rede robusta. A função de dropout zera aleatoriamente alguns neurônios. Isso reduz a dependência de neurônios específicos e previne o overfitting, tornando a rede mais robusta.

O código abaixo implementa a espinha dorsal:

class BackboneNetwork(nn.Module):
    def __init__(self, in_features, hidden_dimensions, out_features, dropout):
        super().__init__()
        self.layer1 = nn.Linear(in_features, hidden_dimensions)
        self.layer2 = nn.Linear(hidden_dimensions, hidden_dimensions)
        self.layer3 = nn.Linear(hidden_dimensions, out_features)
        self.dropout = nn.Dropout(dropout)
    def forward(self, x):
        x = self.layer1(x)
        x = f.relu(x)
        x = self.dropout(x)
        x = self.layer2(x)
        x = f.relu(x)
        x = self.dropout(x)
        x = self.layer3(x)
        return x

Defina a rede ator-crítico

Agora, podemos usar essa rede para definir a classe ator-crítico, ActorCritic. O ator modela a política e prevê a ação. O crítico modela a função de valor e prevê o valor. Ambos recebem o estado como entrada.

class ActorCritic(nn.Module):
    def __init__(self, actor, critic):
        super().__init__()
        self.actor = actor
        self.critic = critic
    def forward(self, state):
        action_pred = self.actor(state)
        value_pred = self.critic(state)
        return action_pred, value_pred

Instancie as redes ator e crítico

Vamos usar as redes definidas acima para criar um ator e um crítico. Em seguida, criaremos um agente, incluindo o ator e o crítico.

Antes de criar o agente, inicialize os parâmetros da rede:

• As dimensões da camada oculta, H, que é um parâmetro configurável. O tamanho e o número de camadas ocultas dependem da complexidade do problema. Usaremos uma camada oculta com dimensões 64 x 64.
• Recursos de entrada, N, onde N é o tamanho do array de estados. A camada de entrada tem dimensões N x H. No ambiente CartPole, o estado é um array de 4 elementos. Portanto, N é 4.
• Recursos de saída da rede ator, O, onde O é o número de ações no ambiente. A camada de saída do ator tem dimensões H x O. O ambiente CartPole tem 2 ações.
• Recursos de saída da rede crítica. Como a rede crítica prevê apenas o valor esperado (dado um estado de entrada), o número de recursos de saída é 1.
• Dropout como uma fração.

O código a seguir mostra como declarar as redes ator e crítica com base na rede principal:

def create_agent(hidden_dimensions, dropout):
    INPUT_FEATURES = env_train.observation_space.shape[0]
    HIDDEN_DIMENSIONS = hidden_dimensions
    ACTOR_OUTPUT_FEATURES = env_train.action_space.n
    CRITIC_OUTPUT_FEATURES = 1
    DROPOUT = dropout
    actor = BackboneNetwork(
            INPUT_FEATURES, HIDDEN_DIMENSIONS, ACTOR_OUTPUT_FEATURES, DROPOUT)
    critic = BackboneNetwork(
            INPUT_FEATURES, HIDDEN_DIMENSIONS, CRITIC_OUTPUT_FEATURES, DROPOUT)
    agent = ActorCritic(actor, critic)
    return agent

Calculando os retornos

O ambiente fornece uma recompensa indo de cada etapa para a próxima, dependendo da ação do agente. A recompensa, R, é expressa como:

O retorno é definido como o valor acumulado de recompensas futuras esperadas. As recompensas de etapas que estão mais distantes no futuro são menos valiosas do que recompensas imediatas. Assim, o retorno é comumente calculado como o retorno descontado, G, definido como:

Neste tutorial (e muitas outras referências), retorno refere-se ao retorno descontado.

Para calcular o retorno:

• Comece com as recompensas esperadas de todos os estados futuros.
• Multiplicar cada recompensa futura por um expoente do fator de desconto, . Por exemplo, a recompensa esperada após 2 etapas (do presente) é multiplicada por 2.
• Somar todas as recompensas futuras descontadas para calcular o retorno.
• Normalizar o valor do retorno.

A função calculate_returns() realiza esses cálculos, conforme mostrado abaixo:

def calculate_returns(rewards, discount_factor):
    returns = []
    cumulative_reward = 0
    for r in reversed(rewards):
        cumulative_reward = r + cumulative_reward * discount_factor
        returns.insert(0, cumulative_reward)
    returns = torch.tensor(returns)
    # normalizar o retorno
    returns = (returns - returns.mean()) / returns.std()
    return returns

Implementando a função de vantagem

A vantagem é calculada como a diferença entre o valor previsto pelo crítico e o retorno esperado das ações escolhidas pelo ator de acordo com a política. Para uma ação específica, a vantagem expressa o benefício de tomar essa ação específica em relação a uma ação arbitrária (média).

No artigo PPO original (equação 10), a vantagem, olhando para frente até o passo de tempo T é expressa como:

Ao codificar o algoritmo, a restrição de olhar para frente até um número definido de passos de tempo é imposta através do tamanho do lote. Portanto, a equação acima pode ser simplificada como a diferença entre o valor e os retornos esperados. Os retornos esperados são quantificados na função de valor estado-ação, Q. 

Assim, a fórmula simplificada abaixo expressa a vantagem de escolher: 

• uma ação específica 
• em um estado dado 
• sob uma política específica 
• em um passo de tempo específico 

Isto é expresso como: 

A OpenAI também utiliza essa fórmula para implementar RL. A calculate_advantages() função mostrada abaixo calcula a vantagem:

def calculate_advantages(returns, values):
    advantages = returns - values
    # Normaliza a vantagem
    advantages = (advantages - advantages.mean()) / advantages.std()
    return advantages

Perda de substituição e mecanismo de recorte

A perda de política seria a perda de gradiente de política padrão sem técnicas especiais como PPO. A perda de gradiente de política padrão é calculada como o produto de:

• As probabilidades de ação da política
• A função de vantagem, que é calculada como a diferença entre:
• O retorno da política
• O valor esperado

A perda de gradiente de política padrão não pode fazer correções para mudanças abruptas na política. A perda de substituição modifica a perda padrão para restringir a quantidade que a política pode mudar em cada iteração. É o mínimo de duas quantidades:

• O produto de:
• A razão da política. Essa razão expressa a diferença entre as probabilidades de ação antigas e novas.
• A função de vantagem
• O produto de:
• O valor limitado da razão da política. Essa razão é cortada de forma que a política atualizada esteja dentro de uma certa porcentagem da política antiga.
• A função de vantagem

Para o processo de otimização, a perda substituta é usada como um proxy para a perda real.

O mecanismo de recorte

A taxa de política, R, é a diferença entre as novas e antigas políticas e é dada como a razão das probabilidades logarítmicas da política sob os novos e antigos parâmetros:

A taxa de política recortada, R’, é restrita de forma que:

Dada a vantagem, At, conforme mostrado na seção anterior, e a taxa de política, como mostrado acima, a perda de substituição é calculada da seguinte forma:

O código abaixo mostra como implementar o mecanismo de recorte e a perda de substituição.

def calculate_surrogate_loss(
        actions_log_probability_old,
        actions_log_probability_new,
        epsilon,
        advantages):
    advantages = advantages.detach()
    policy_ratio = (
            actions_log_probability_new - actions_log_probability_old
            ).exp()
    surrogate_loss_1 = policy_ratio * advantages
    surrogate_loss_2 = torch.clamp(
            policy_ratio, min=1.0-epsilon, max=1.0+epsilon
            ) * advantages
    surrogate_loss = torch.min(surrogate_loss_1, surrogate_loss_2)
    return surrogate_loss

3. Treinando o Agente

Agora, vamos treinar o agente.

Calculando a perda de política e valor

Agora estamos prontos para calcular as perdas de política e valor:

• A perda da política é a soma da perda substituta e do bônus de entropia.
• A perda de valor é baseada na diferença entre o valor previsto pelo crítico e os retornos (recompensa acumulada) gerados pela política. O cálculo da perda de valor utiliza a função de perda Smooth L1. Isso ajuda a suavizar a função de perda e a torná-la menos sensível a outliers.

Ambas as perdas, conforme calculadas acima, são tensores. O gradiente descendente é baseado em valores escalares. Para obter um único valor escalar representando a perda, use a .sum() função para somar os elementos do tensor. A função abaixo mostra como fazer isso:

def calculate_losses(
        surrogate_loss, entropy, entropy_coefficient, returns, value_pred):
    entropy_bonus = entropy_coefficient * entropy
    policy_loss = -(surrogate_loss + entropy_bonus).sum()
    value_loss = f.smooth_l1_loss(returns, value_pred).sum()
    return policy_loss, value_loss

Definindo o loop de treinamento

Antes de iniciar o processo de treinamento, crie um conjunto de buffers como arrays vazios. O algoritmo de treinamento usará esses buffers para armazenar informações sobre as ações do agente, os estados do ambiente e as recompensas em cada etapa de tempo. A função abaixo inicializa esses buffers:

def init_training():
    states = []
    actions = []
    actions_log_probability = []
    values = []
    rewards = []
    done = False
    episode_reward = 0
    return states, actions, actions_log_probability, values, rewards, done, episode_reward

Cada iteração de treinamento executa o agente com os parâmetros de política para aquela iteração. O agente interage com o ambiente em passos de tempo em um loop até que ele atinja uma condição terminal.

Após cada passo de tempo, a ação, recompensa e valor do agente são adicionados aos buffers respectivos. Quando o episódio termina, a função retorna o conjunto atualizado de buffers, que resumem os resultados do episódio.

Antes de executar o loop de treinamento:

• Defina o modelo para o modo de treinamento usando agent.train().
• Redefina o ambiente para um estado aleatório usando env.reset(). Este é o estado inicial para esta iteração de treinamento.

Os seguintes passos explicam o que acontece em cada passo de tempo no loop de treinamento:

• Passe o estado para o agente.
• O agente retorna:
• A ação prevista dada o estado, com base na política (ator). Passe este tensor de ação previsto pela função softmax para obter o conjunto de probabilidades de ação.
• O valor previsto do estado, com base no crítico.
• O agente seleciona a ação a ser tomada:
• Use as probabilidades de ação para estimar a distribuição de probabilidade.
• Selecione uma ação aleatoriamente escolhendo uma amostra desta distribuição. A função dist.sample() faz isso.
• Use a função env.step() para passar esta ação para o ambiente e simular a resposta do ambiente para este passo de tempo. Com base na ação do agente, o ambiente gera:
• O novo estado
• A recompensa
• O valor booleano de retorno done (isso indica se o ambiente atingiu um estado terminal)
• Anexar aos buffers respectivos os valores da ação do agente, as recompensas, os valores previstos e o novo estado.

O episódio de treinamento termina quando a função env.step() retorna true para o valor booleano de retorno de done.

Depois que o episódio terminar, use os valores acumulados de cada passo de tempo para calcular os retornos cumulativos deste episódio adicionando as recompensas de cada passo de tempo. Usamos a função calculate_returns() descrita anteriormente para fazer isso. Os inputs desta função são o fator de desconto e o buffer contendo as recompensas de cada passo de tempo. Usamos esses retornos e os valores acumulados de cada passo de tempo para calcular as vantagens usando a função calculate_advantages().

A seguinte função em Python mostra como implementar essas etapas:

def forward_pass(env, agent, optimizer, discount_factor):
    states, actions, actions_log_probability, values, rewards, done, episode_reward = init_training()
    state = env.reset()
    agent.train()
    while not done:
        state = torch.FloatTensor(state).unsqueeze(0)
        states.append(state)
        action_pred, value_pred = agent(state)
        action_prob = f.softmax(action_pred, dim=-1)
        dist = distributions.Categorical(action_prob)
        action = dist.sample()
        log_prob_action = dist.log_prob(action)
        state, reward, done, _ = env.step(action.item())
        actions.append(action)
        actions_log_probability.append(log_prob_action)
        values.append(value_pred)
        rewards.append(reward)
        episode_reward += reward
    states = torch.cat(states)
    actions = torch.cat(actions)
    actions_log_probability = torch.cat(actions_log_probability)
    values = torch.cat(values).squeeze(-1)
    returns = calculate_returns(rewards, discount_factor)
    advantages = calculate_advantages(returns, values)
    return episode_reward, states, actions, actions_log_probability, advantages, returns

Atualizando os parâmetros do modelo

Cada iteração de treinamento executa o modelo por um episódio completo consistindo em muitos passos de tempo (até atingir uma condição terminal). Em cada passo de tempo, armazenamos os parâmetros da política, a ação do agente, os retornos e as vantagens. Após cada iteração, atualizamos o modelo com base no desempenho da política por todos os passos de tempo nessa iteração.

O número máximo de etapas no ambiente CartPole é 500. Em ambientes mais complexos, existem mais etapas, até milhões. Nesses casos, o conjunto de resultados do treinamento deve ser dividido em lotes. O número de etapas em cada lote é chamado de tamanho do lote de otimização.

Assim, os passos para atualizar os parâmetros do modelo são:

• Dividir o conjunto de resultados do treinamento em lotes.
• Para cada lote:
• Obter a ação do agente e o valor previsto para cada estado.
• Usar essas ações previstas para estimar a nova distribuição de probabilidade de ação.
• Use esta distribuição para calcular a entropia.
• Use esta distribuição para obter a probabilidade logarítmica das ações no conjunto de dados de resultados de treinamento. Este é o novo conjunto de probabilidades logarítmicas das ações no conjunto de dados de resultados de treinamento. O antigo conjunto de probabilidades logarítmicas dessas mesmas ações foi calculado no loop de treinamento explicado na seção anterior.
• Calcule a perda substituta usando as distribuições de probabilidade antigas e novas das ações.
• Calcule a perda de política e a perda de valor usando a perda substituta, a entropia e as vantagens.
• Execute .backward() separadamente nas perdas de política e valor. Isso atualiza os gradientes nas funções de perda.
• Execute .step() no otimizador para atualizar os parâmetros da política. Neste caso, utilizamos o otimizador Adam para equilibrar velocidade e robustez.
• Acumule as perdas da política e do valor.
• Repita o passo para trás (as operações acima) em cada lote algumas vezes, dependendo do valor do parâmetro PPO_STEPS. Repetir o passo para trás em cada lote é computacionalmente eficiente, pois aumenta efetivamente o tamanho do conjunto de dados de treinamento sem a necessidade de executar passagens para frente adicionais. O número de passos no ambiente em cada alternância entre amostragem e otimização é chamado de tamanho do lote de iteração.
• Retorne a perda média da política e a perda do valor.

O código abaixo implementa esses passos:

def update_policy(
        agent,
        states,
        actions,
        actions_log_probability_old,
        advantages,
        returns,
        optimizer,
        ppo_steps,
        epsilon,
        entropy_coefficient):
    BATCH_SIZE = 128
    total_policy_loss = 0
    total_value_loss = 0
    actions_log_probability_old = actions_log_probability_old.detach()
    actions = actions.detach()
    training_results_dataset = TensorDataset(
            states,
            actions,
            actions_log_probability_old,
            advantages,
            returns)
    batch_dataset = DataLoader(
            training_results_dataset,
            batch_size=BATCH_SIZE,
            shuffle=False)
    for _ in range(ppo_steps):
        for batch_idx, (states, actions, actions_log_probability_old, advantages, returns) in enumerate(batch_dataset):
            # obter novo log de probabilidade das ações para todos os estados de entrada
            action_pred, value_pred = agent(states)
            value_pred = value_pred.squeeze(-1)
            action_prob = f.softmax(action_pred, dim=-1)
            probability_distribution_new = distributions.Categorical(
                    action_prob)
            entropy = probability_distribution_new.entropy()
            # estimar novas probabilidades logarítmicas usando ações antigas
            actions_log_probability_new = probability_distribution_new.log_prob(actions)
            surrogate_loss = calculate_surrogate_loss(
                    actions_log_probability_old,
                    actions_log_probability_new,
                    epsilon,
                    advantages)
            policy_loss, value_loss = calculate_losses(
                    surrogate_loss,
                    entropy,
                    entropy_coefficient,
                    returns,
                    value_pred)
            optimizer.zero_grad()
            policy_loss.backward()
            value_loss.backward()
            optimizer.step()
            total_policy_loss += policy_loss.item()
            total_value_loss += value_loss.item()
    return total_policy_loss / ppo_steps, total_value_loss / ppo_steps

4. Executando o Agente PPO

Vamos finalmente executar o agente PPO.

Avaliando o desempenho

Para avaliar o desempenho do agente, crie um ambiente novo e calcule as recompensas cumulativas ao executar o agente nesse novo ambiente. Você precisa definir o agente para o modo de avaliação usando a função .eval(). Os passos são os mesmos do laço de treinamento. O trecho de código abaixo implementa a função de avaliação:

def evaluate(env, agent):
    agent.eval()
    rewards = []
    done = False
    episode_reward = 0
    state = env.reset()
    while not done:
        state = torch.FloatTensor(state).unsqueeze(0)
        with torch.no_grad():
            action_pred, _ = agent(state)
            action_prob = f.softmax(action_pred, dim=-1)
        action = torch.argmax(action_prob, dim=-1)
        state, reward, done, _ = env.step(action.item())
        episode_reward += reward
    return episode_reward

Visualizando os resultados do treinamento

Vamos usar a biblioteca Matplotlib para visualizar o progresso do processo de treinamento. A função abaixo mostra como plotar as recompensas tanto dos laços de treinamento quanto de teste:

def plot_train_rewards(train_rewards, reward_threshold):
    plt.figure(figsize=(12, 8))
    plt.plot(train_rewards, label='Training Reward')
    plt.xlabel('Episode', fontsize=20)
    plt.ylabel('Training Reward', fontsize=20)
    plt.hlines(reward_threshold, 0, len(train_rewards), color='y')
    plt.legend(loc='lower right')
    plt.grid()
    plt.show()

def plot_test_rewards(test_rewards, reward_threshold):
    plt.figure(figsize=(12, 8))
    plt.plot(test_rewards, label='Testing Reward')
    plt.xlabel('Episode', fontsize=20)
    plt.ylabel('Testing Reward', fontsize=20)
    plt.hlines(reward_threshold, 0, len(test_rewards), color='y')
    plt.legend(loc='lower right')
    plt.grid()
    plt.show()

Nos gráficos de exemplo abaixo, mostramos as recompensas de treinamento e teste, obtidas aplicando a política nos ambientes de treinamento e teste, respectivamente. Observe que a forma desses gráficos parecerá diferente toda vez que você executar o código. Isso ocorre devido à aleatoriedade inerente ao processo de treinamento.

Recompensas de treinamento (obtidas aplicando a política no ambiente de treinamento). Imagem por Autor.

Recompensas de teste (obtidas aplicando a política no ambiente de teste). Imagem por Autor.

Nos gráficos de saída mostrados acima, observe o progresso do processo de treinamento:

• A recompensa começa com valores baixos. Conforme o treinamento avança, as recompensas aumentam.
• As recompensas flutuam aleatoriamente enquanto aumentam. Isso ocorre devido ao agente explorar o espaço de políticas.
• O treinamento termina, e as recompensas de teste se estabilizaram em torno do limite (475) por muitas iterações.
• As recompensas são limitadas a 500. Essas são restrições impostas pelo ambiente (Gym CartPole v1).

Da mesma forma, você pode plotar os valores e as perdas de política ao longo das iterações:

def plot_losses(policy_losses, value_losses):
    plt.figure(figsize=(12, 8))
    plt.plot(value_losses, label='Value Losses')
    plt.plot(policy_losses, label='Policy Losses')
    plt.xlabel('Episode', fontsize=20)
    plt.ylabel('Loss', fontsize=20)
    plt.legend(loc='lower right')
    plt.grid()
    plt.show()

O gráfico de exemplo abaixo mostra as perdas rastreadas através dos episódios de treinamento:

Valor e perdas de política durante o processo de treinamento. Imagem do Autor

Observe o gráfico e note:

• As perdas parecem estar distribuídas aleatoriamente e não seguem nenhum padrão.
• Isso é típico do treinamento de RL, onde o objetivo não é minimizar a perda, mas maximizar as recompensas.

Execute o algoritmo PPO

Agora você tem todos os componentes para treinar o agente usando PPO. Para colocar tudo junto, você precisa:

• Declarar hiperparâmetros como fator de desconto, tamanho do lote, taxa de aprendizado, etc.
• Instancie buffers como arrays nulos para armazenar as recompensas e perdas de cada iteração.
• Crie uma instância de agente usando a função create_agent().
• Rode iterativamente os passes para frente e para trás usando as funções forward_pass() e update_policy().
• Teste o desempenho da política usando a função evaluate().
• Adicione a política, perdas de valor e recompensas das funções de treinamento e avaliação aos buffers respectivos.
• Calcule a média das recompensas e perdas nos últimos passos. O exemplo abaixo calcula a média das recompensas e perdas nos últimos 40 passos de tempo.
• Imprima os resultados da avaliação a cada poucos passos. O exemplo abaixo imprime a cada 10 passos.
• Termine o processo quando a média da recompensa ultrapassar um certo limiar.

O código abaixo mostra como declarar uma função que faz isso em Python:

def run_ppo():
    MAX_EPISODES = 500
    DISCOUNT_FACTOR = 0.99
    REWARD_THRESHOLD = 475
    PRINT_INTERVAL = 10
    PPO_STEPS = 8
    N_TRIALS = 100
    EPSILON = 0.2
    ENTROPY_COEFFICIENT = 0.01
    HIDDEN_DIMENSIONS = 64
    DROPOUT = 0.2
    LEARNING_RATE = 0.001
    train_rewards = []
    test_rewards = []
    policy_losses = []
    value_losses = []
    agent = create_agent(HIDDEN_DIMENSIONS, DROPOUT)
    optimizer = optim.Adam(agent.parameters(), lr=LEARNING_RATE)
    for episode in range(1, MAX_EPISODES+1):
        train_reward, states, actions, actions_log_probability, advantages, returns = forward_pass(
                env_train,
                agent,
                optimizer,
                DISCOUNT_FACTOR)
        policy_loss, value_loss = update_policy(
                agent,
                states,
                actions,
                actions_log_probability,
                advantages,
                returns,
                optimizer,
                PPO_STEPS,
                EPSILON,
                ENTROPY_COEFFICIENT)
        test_reward = evaluate(env_test, agent)
        policy_losses.append(policy_loss)
        value_losses.append(value_loss)
        train_rewards.append(train_reward)
        test_rewards.append(test_reward)
        mean_train_rewards = np.mean(train_rewards[-N_TRIALS:])
        mean_test_rewards = np.mean(test_rewards[-N_TRIALS:])
        mean_abs_policy_loss = np.mean(np.abs(policy_losses[-N_TRIALS:]))
        mean_abs_value_loss = np.mean(np.abs(value_losses[-N_TRIALS:]))
        if episode % PRINT_INTERVAL == 0:
            print(f'Episode: {episode:3} | \
                  Mean Train Rewards: {mean_train_rewards:3.1f} \
                  | Mean Test Rewards: {mean_test_rewards:3.1f} \
                  | Mean Abs Policy Loss: {mean_abs_policy_loss:2.2f} \
                  | Mean Abs Value Loss: {mean_abs_value_loss:2.2f}')
        if mean_test_rewards >= REWARD_THRESHOLD:
            print(f'Reached reward threshold in {episode} episodes')
            break
    plot_train_rewards(train_rewards, REWARD_THRESHOLD)
    plot_test_rewards(test_rewards, REWARD_THRESHOLD)
    plot_losses(policy_losses, value_losses)

Execute o programa:

run_ppo()

A saída deve se assemelhar ao exemplo abaixo:

Episode:  10 | Mean Train Rewards: 22.3 | Mean Test Rewards: 30.4 | Mean Abs Policy Loss: 0.37 | Mean Abs Value Loss: 0.39 
Episode:  20 | Mean Train Rewards: 38.6 | Mean Test Rewards: 69.8 | Mean Abs Policy Loss: 0.46 | Mean Abs Value Loss: 0.37 
.
.
.
Episode: 100 | Mean Train Rewards: 289.5 | Mean Test Rewards: 427.3 | Mean Abs Policy Loss: 1.73 | Mean Abs Value Loss: 0.21 
Episode: 110 | Mean Train Rewards: 357.7 | Mean Test Rewards: 461.4 | Mean Abs Policy Loss: 1.86 | Mean Abs Value Loss: 0.22 
Reached reward threshold in 116 episodes

Você pode visualizar e executar o programa em funcionamento neste notebook DataLab!

5. Sintonia e Otimização de Hiperparâmetros

No aprendizado de máquina, os hiperparâmetros controlam o processo de treinamento. Abaixo, explico alguns dos hiperparâmetros importantes usados no PPO: 

• Taxa de aprendizado: A taxa de aprendizado decide o quanto os parâmetros de política podem variar em cada iteração. No descida de gradiente estocástica, a quantidade pela qual os parâmetros de política são atualizados em cada iteração é decidida pelo produto da taxa de aprendizado e do gradiente. 
• Parâmetro de recorte: Isso também é chamado de epsilon, ε. Ele decide a extensão em que a taxa de política é recortada. A razão das novas e antigas políticas é permitida variar na faixa [1-ε, 1+ε]. Quando está além dessa faixa, é artificialmente recortado para estar dentro da faixa. 
• Tamanho do lote: Isso se refere ao número de etapas a serem consideradas para cada atualização de gradiente. No PPO, o tamanho do lote é o número de passos de tempo necessários para aplicar a política e calcular a perda substituta para atualizar os parâmetros da política. Neste artigo, usamos um tamanho de lote de 64. 
• Passos de iteração: Este é o número de vezes que cada lote é reutilizado para executar a passagem de retrocesso. O código neste artigo se refere a isso como PPO_STEPS. Em ambientes complexos, executar a passagem para frente muitas vezes é computacionalmente caro. Uma alternativa mais eficiente é reexecutar cada lote algumas vezes. Geralmente é recomendado usar um valor entre 5 e 10. 
• Fator de desconto: Isso também é referido como gamma, γ. Ele expressa o quanto as recompensas imediatas são mais valiosas do que as recompensas futuras. Isso é semelhante ao conceito de taxas de juros ao calcular o valor do tempo do dinheiro. Quando está mais próximo de 0, significa que as recompensas futuras são menos valiosas e o agente deve priorizar as recompensas imediatas. Quando está mais próximo de 1, significa que as recompensas futuras são importantes.
• Coeficiente de entropia: O coeficiente de entropia decide o bônus de entropia, que é calculado como o produto do coeficiente de entropia e a entropia da distribuição. O papel do bônus de entropia é introduzir mais aleatoriedade na política. Isso encoraja o agente a explorar o espaço da política. No entanto, o treinamento falha em convergir para uma política ideal quando essa aleatoriedade é muito alta.
• Critérios de sucesso para o treinamento: Você precisa definir os critérios para decidir quando o treinamento é bem-sucedido. Uma forma comum de fazer isso é estabelecer a condição de que as recompensas médias ao longo dos últimos N testes (episódios) sejam acima de um determinado limite. No código de exemplo acima, isso é expresso com a variável N_TRIALS. Quando isso é definido com um valor mais alto, o treinamento leva mais tempo porque a política precisa atingir a recompensa de limite ao longo de mais episódios. Isso também resulta em uma política mais robusta, embora seja mais caro computacionalmente. Note que PPO é uma política estocástica, e haverá episódios em que o agente não ultrapassa o limite. Portanto, se o valor de N_TRIALS for muito alto, seu treinamento pode não terminar.

Estratégias para otimizar o desempenho do PPO

Otimizar o desempenho dos algoritmos de treinamento PPO envolve tentativa e erro e experimentação com diferentes valores de hiperparâmetros. No entanto, existem algumas diretrizes amplas:

• Fator de desconto: Quando as recompensas de longo prazo são importantes, como no ambiente CartPole, onde o poste precisa permanecer estável ao longo do tempo, comece com um valor de gama moderado, como 0,99.
• Bônus de entropia: Em ambientes complexos, o agente deve explorar o espaço de ação para encontrar a política ótima. O bônus de entropia promove a exploração. O bônus de entropia é adicionado à perda substituta. Verifique a magnitude da perda substituta e a entropia da distribuição antes de decidir o coeficiente de entropia. Neste artigo, usamos um coeficiente de entropia de 0,01.
• Parâmetro de recorte: O parâmetro de recorte decide o quão diferente a política atualizada pode ser da política atual. Um valor grande do parâmetro de recorte incentiva uma melhor exploração do ambiente, mas também corre o risco de desestabilizar o treinamento. Você deseja um parâmetro de recorte que permita uma exploração gradual enquanto previne atualizações desestabilizadoras. Neste artigo, utilizamos um parâmetro de recorte de 0.2.
• Taxa de aprendizado: Quando a taxa de aprendizado é muito alta, a política é atualizada em passos grandes a cada iteração, e o processo de treinamento pode se tornar instável. Quando é muito baixa, o treinamento leva muito tempo. Este tutorial utilizou uma taxa de aprendizado de 0.001, que funciona bem para o ambiente. Em muitos casos, é recomendado usar uma taxa de aprendizado de 1e-5.

Desafios e Melhores Práticas em PPO

Depois de explicar os conceitos e detalhes de implementação do PPO, vamos discutir os desafios e as melhores práticas.

Desafios comuns no treinamento de PPO

Mesmo que o PPO seja amplamente utilizado, é necessário estar ciente dos desafios potenciais para resolver problemas do mundo real com sucesso usando essa técnica. Alguns desses desafios são:

• Convergência lenta:Em ambientes complexos, o PPO pode ser ineficiente em amostragem e precisa de muitas interações com o ambiente para convergir na política ótima. Isso torna o treinamento lento e caro.
• Sensibilidade aos hiperparâmetros: O PPO depende de explorar eficientemente o espaço de políticas. A estabilidade do processo de treinamento e a velocidade de convergência são sensíveis aos valores dos hiperparâmetros. Os valores ótimos desses hiperparâmetros frequentemente só podem ser determinados por tentativa e erro. 
• Sobreajuste:  Os ambientes de RL são tipicamente inicializados com parâmetros aleatórios. O treinamento do PPO é baseado em encontrar a política ótima com base no ambiente do agente. Às vezes, o processo de treinamento converge para um conjunto de parâmetros ótimos para um ambiente específico, mas não para qualquer ambiente randomizado. Isso é tipicamente abordado tendo muitas iterações, cada uma com um ambiente de treinamento randomizado de forma diferente. 
• Ambientes dinâmicos: Ambientes de RL simples, como o ambiente CartPole, são estáticos – as regras são as mesmas ao longo do tempo. Muitos outros ambientes, como um robô aprendendo a andar em uma superfície instável e em movimento, são dinâmicos – as regras do ambiente mudam com o tempo. Para se sair bem em tais ambientes, o PPO frequentemente precisa de ajustes adicionais.
• Exploração vs exploração: O mecanismo de recorte do PPO garante que as atualizações de política estejam dentro de uma região confiável. No entanto, isso também impede que o agente explore o espaço de ação. Isso pode levar à convergência para ótimos locais, especialmente em ambientes complexos. Por outro lado, permitir que o agente explore demais pode impedi-lo de convergir para qualquer política ótima.

Práticas recomendadas para treinar modelos PPO

Para obter bons resultados usando PPO, recomendo algumas melhores práticas, como:

• Normalizar as características de entrada: Normalizar os valores de retornos e vantagens reduz a variabilidade nos dados e leva a atualizações de gradientes estáveis. Normalizar os dados traz todos os valores para uma faixa numérica consistente. Isso ajuda a reduzir o efeito de valores atípicos e extremos, que poderiam distorcer as atualizações de gradientes e desacelerar a convergência.
• Use tamanhos de lote adequados: Lotes pequenos permitem atualizações e treinamentos mais rápidos, mas podem levar à convergência para ótimos locais e instabilidade no processo de treinamento. Tamanhos de lote maiores permitem que o agente aprenda políticas robustas, levando a um processo de treinamento estável. No entanto, tamanhos de lote muito grandes também são subótimos. Além de aumentar os custos computacionais, eles tornam as atualizações de políticas menos responsivas à função de valor porque as atualizações de gradiente são baseadas em médias estimadas ao longo de lotes grandes. Além disso, pode levar ao ajuste excessivo das atualizações para aquele lote específico. 
• Passos de iteração: Geralmente é aconselhável reutilizar cada lote por 5-10 iterações. Isso torna o processo de treinamento mais eficiente. Reutilizar o mesmo lote muitas vezes leva ao ajuste excessivo. O código se refere a este hiperparâmetro como PPO_STEPS. 
• Realize avaliações regulares: Para detectar o overfitting, é essencial monitorar periodicamente a eficácia da política. Se a política se mostrar ineficaz em determinados cenários, pode ser necessário realizar mais treinamento ou ajustes finos.
• Ajuste os hiperparâmetros: Como explicado anteriormente, o treinamento PPO é sensível aos valores dos hiperparâmetros. Experimente com vários valores de hiperparâmetros para determinar o conjunto correto de valores para o seu problema específico.
• Rede compartilhada de backbone: Conforme ilustrado neste artigo, o uso de um backbone compartilhado evita desequilíbrios entre as redes de atores e críticos. Compartilhar uma rede de backbone entre o ator e o crítico ajuda na extração de características compartilhadas e em uma compreensão comum do ambiente. Isso torna o processo de aprendizado mais eficiente e estável. Também ajuda a reduzir o espaço computacional e a complexidade temporal do algoritmo. 
• Número e tamanho das camadas ocultas: Aumente o número de camadas ocultas e dimensões para ambientes mais complexos. Problemas mais simples como o CartPole podem ser resolvidos com uma única camada oculta. A camada oculta usada neste artigo tem 64 dimensões. Tornar a rede muito maior do que o necessário é computacionalmente desperdiçoso e pode torná-la instável. 
• Parada antecipada: Parar o treinamento quando as métricas de avaliação são atendidas ajuda a evitar o excesso de treinamento e evita o desperdício de recursos. Uma métrica de avaliação comum é quando o agente excede as recompensas de limite nas últimas N ocorrências.

Conclusão

Neste artigo, discutimos o PPO como uma maneira de resolver problemas de RL. Em seguida, detalhamos os passos para implementar o PPO usando o PyTorch. Por fim, apresentamos algumas dicas de desempenho e melhores práticas para o PPO.

A melhor maneira de aprender é implementar o código você mesmo. Você também pode modificar o código para funcionar com outros ambientes de controle clássico no Gym. Para aprender a implementar agentes de RL usando Python e o Gymnasium da OpenAI, siga o curso Aprendizado por Reforço com Gymnasium em Python!