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

1. Configurando o Ambiente

Antes de implementar o PPO, precisamos instalar as bibliotecas de software necessárias e escolher um ambiente adequado para aplicar a política.

Instalando o PyTorch e as bibliotecas necessárias

Precisamos instalar o seguinte software: 

• 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 de código aberto Gym da OpenAI, uma biblioteca Python que simula diferentes ambientes e jogos, os quais podem ser resolvidos usando Aprendizado por Reforço. Você pode utilizar a API do Gym para fazer com que seu algoritmo interaja 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 PPO usando PyTorch.

Definindo a rede de política

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

Defina a rede backbone

Os seguintes passos descrevem a rede backbone:

• Implemente uma rede com 3 camadas – uma 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 aplicamos 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 evita o overfitting, tornando assim 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 de 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 estado. 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 de 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.
• Desistência 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 que vai 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. Recompensas de etapas que estão mais distantes no futuro são menos valiosas do que as 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 (a partir do presente) é multiplicada por 2.
• Somar todas as recompensas futuras descontadas para calcular o retorno.
• Normalize o valor do retorno.

A função calcular_retornos() realiza esses cálculos, como 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 é aplicada por meio do tamanho do lote. Assim, 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 valor de estado-ação, Q.

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

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

Isso é expresso como: 

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

def calculate_advantages(returns, values):
    advantages = returns - values
    # Normalizar 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 padrão do gradiente de política sem técnicas especiais como PPO. A perda padrão do gradiente de política é 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 padrão do gradiente de política não pode fazer correções para mudanças abruptas na política. A perda substituta 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 proporção da política. Essa proporçã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 proporção da política. Essa proporçã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 corte

A taxa de política, R, é a diferença entre as políticas nova e antiga 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 cortada, R’, é limitada de tal forma que:

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

O código abaixo mostra como implementar o mecanismo de corte e a perda substituta.

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 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 cumulativa) gerados pela política. O cálculo da perda de valor usa a função de perda Smooth L1 Loss. Isso ajuda a suavizar a função de perda e torná-la menos sensível a valores discrepantes.

Ambas as perdas, conforme calculadas acima, são tensores. O descida do gradiente é baseada em valores escalares. Para obter um único valor escalar representando a perda, use a função .sum() 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 passo 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 essa 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 anexados 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 passos a seguir explicam o que acontece em cada passo 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 prevista 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 aleatoriamente uma ação escolhendo uma amostra desta distribuição. A função dist.sample() faz isso.
• Use a função env.step() para passar essa 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 de retorno booleano done (indica se o ambiente atingiu um estado terminal)
• Adicione aos buffers respectivos os valores da ação do agente, das recompensas, dos valores previstos e do novo estado.

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

Após o episódio ter terminado, 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 dessa 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 esses passos:

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 através de um episódio completo consistindo de muitos passos de tempo (até que atinja 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 através de todos os passos de tempo nessa iteração.

O número máximo de passos na envoltura CartPole é 500. Em ambientes mais complexos, existem mais passos, até milhões. Nesses casos, o conjunto de dados de resultados de treinamento deve ser dividido em lotes. O número de passos 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 dados de resultados de 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.
• Utilize esta distribuição para calcular a entropia.
• Utilize esta distribuição para obter a probabilidade do log das ações no conjunto de dados de resultados de treinamento. Este é o novo conjunto de probabilidades do log das ações no conjunto de dados de resultados de treinamento. O antigo conjunto de probabilidades do log dessas mesmas ações foi calculado no loop de treinamento explicado na seção anterior.
• Calcule a perda substituta utilizando as distribuições de probabilidade antigas e novas das ações.
• Calcule a perda de política e a perda de valor utilizando a perda substituta, a entropia e as vantagens.
• Execute .backward() separadamente nas perdas de política e de 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 a passagem para trás (as operações acima) em cada lote algumas vezes, dependendo do valor do parâmetro PPO_STEPS. Repetir a passagem para trás em cada lote é computacionalmente eficiente porque aumenta efetivamente o tamanho do conjunto de dados de treinamento sem a necessidade de executar passagens para frente adicionais. O número de passos do ambiente em cada alternância entre amostragem e otimização é chamado de tamanho do lote de iteração.
• Retorne a média da perda da política e da perda de valor.

O código abaixo implementa essas etapas:

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 nova probabilidade de log 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 de log utilizando 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 acumuladas ao executar o agente neste 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 loop 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 loops 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 a partir de valores baixos. À medida que 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. Estas são restrições impostas pelo ambiente (Gym CartPole v1).

Da mesma forma, você pode traçar as perdas de valor e de política através 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 ao longo dos episódios de treinamento:

Valores 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 de forma aleatória e não seguem nenhum padrão.
• Isto é típico do treinamento de RL, onde o objetivo não é minimizar a perda, mas sim 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().
• Execute 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 poucos passos de tempo. 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.
• Encerre o processo quando a média da recompensa ultrapassar um certo limite.

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 funcionando neste caderno 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 é referido como epsilon, ε. Decide em que medida a relação de políticas é recortada. A relação entre as novas e antigas políticas pode variar no intervalo [1-ε, 1+ε]. Quando está além desse intervalo, é artificialmente recortado para se situar dentro do intervalo.
• Tamanho do lote: Isso se refere ao número de passos a considerar 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.
• Etapas de iteração: Este é o número de vezes que cada lote é reutilizado para executar a passagem para trás. 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. Normalmente, é recomendado usar um valor entre 5 e 10.
• Fator de desconto: Este também é conhecido como gamma, γ. Expressa o quanto recompensas imediatas são mais valiosas do que recompensas futuras. Isso é semelhante ao conceito de taxas de juros no cálculo do valor do dinheiro no tempo. 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 ótima 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 maneira comum de fazer isso é estabelecer uma condição de que as recompensas médias ao longo dos últimos N testes (episódios) estejam 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 maior, o treinamento leva mais tempo porque a política precisa atingir a recompensa limite ao longo de mais episódios. Isso também resulta em uma política mais robusta, embora seja mais caro computacionalmente. Note que o 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 gerais:

• 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 quão diferente a política atual pode ser da política atualizada. Um valor alto do parâmetro de recorte encoraja uma melhor exploração do ambiente, mas há o risco de desestabilizar o treinamento. Você deseja um parâmetro de recorte que permita uma exploração gradual, ao mesmo tempo que previne atualizações desestabilizadoras. Neste artigo, usamos 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 em cada iteração e o processo de treinamento pode se tornar instável. Quando é muito baixa, o treinamento leva muito tempo. Este tutorial usou 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

Após 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, é importante 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 termos de 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 muitas vezes só podem ser determinados por tentativa e erro.
• Overfitting:  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 resolvido 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 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, também impede o agente de explorar 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 boas práticas, tais 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 gradiente estáveis. Normalizar os dados traz todos os valores para uma faixa numérica consistente. Isso ajuda a reduzir o efeito de outliers e valores extremos, que de outra forma poderiam distorcer as atualizações de gradiente e desacelerar a convergência.
• Use tamanhos de lote adequadamente grandes: 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ítica menos responsivas à função de valor porque as atualizações de gradiente são baseadas em médias estimadas em lotes grandes. Além disso, pode levar ao ajuste excessivo das atualizações a esse 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 overfitting, é essencial monitorar periodicamente a eficácia da política. Se a política se mostrar ineficaz em determinadas situações, pode ser necessário realizar mais treinamento ou ajustes.
• Ajuste os hiperparâmetros: Como explicado anteriormente, o treinamento PPO é sensível aos valores dos hiperparâmetros. Experimente com várias combinações de valores de hiperparâmetros para determinar o conjunto correto de valores para o seu problema específico.
• Rede central compartilhada: Conforme ilustrado neste artigo, o uso de uma rede central compartilhada evita desequilíbrios entre as redes de ator e crítico. Compartilhar uma rede central entre o ator e o crítico ajuda na extração compartilhada de características e em uma compreensão comum do ambiente. Isso torna o processo de aprendizagem 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 CartPole podem ser resolvidos com uma única camada oculta. A camada oculta usada neste artigo possui 64 dimensões. Tornar a rede muito maior do que o necessário é um desperdício computacional e pode torná-la instável. 
• Parada precoce: Parar o treinamento quando as métricas de avaliação são atendidas ajuda a prevenir o overtraining e evita o desperdício de recursos. Uma métrica de avaliação comum é quando o agente excede as recompensas de limite sobre os últimos N eventos.

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 como implementar agentes de RL usando Python e o Gymnasium da OpenAI, siga o curso Aprendizado por Reforço com Gymnasium em Python!