Noções básicas sobre derivação de política de gradiente

Estou tentando recriar um exemplo muito simples de Gradiente de política, a partir do blog de Andrej Karpathy, recurso de origem . Nesse artigo, você encontrará exemplos com CartPole e Policy Gradient com lista de peso e ativação do Softmax. Aqui está o meu exemplo recriado e muito simples de gradiente de política CartPole, que funciona perfeitamente .

import gym
import numpy as np
import matplotlib.pyplot as plt
from sklearn.preprocessing import PolynomialFeatures
import copy

NUM_EPISODES = 4000
LEARNING_RATE = 0.000025
GAMMA = 0.99


# noinspection PyMethodMayBeStatic
class Agent:
    def __init__(self):
        self.poly = PolynomialFeatures(1)
        self.w = np.random.rand(5, 2)

    def policy(self, state):
        z = state.dot(self.w)
        exp = np.exp(z)
        return exp/np.sum(exp)

    def __softmax_grad(self, softmax):
        s = softmax.reshape(-1,1)
        return np.diagflat(s) - np.dot(s, s.T)

    def grad(self, probs, action, state):
        dsoftmax = self.__softmax_grad(probs)[action,:]
        dlog = dsoftmax / probs[0,action]
        grad = state.T.dot(dlog[None,:])
        return grad

    def update_with(self, grads, rewards):

        for i in range(len(grads)):
            # Loop through everything that happend in the episode
            # and update towards the log policy gradient times **FUTURE** reward

            total_grad_effect = 0
            for t, r in enumerate(rewards[i:]):
                total_grad_effect += r * (GAMMA ** r)
            self.w += LEARNING_RATE * grads[i] * total_grad_effect
            print("Grads update: " + str(np.sum(grads[i])))



def main(argv):
    env = gym.make('CartPole-v0')
    np.random.seed(1)

    agent = Agent()
    complete_scores = []

    for e in range(NUM_EPISODES):
        state = env.reset()[None, :]
        state = agent.poly.fit_transform(state)

        rewards = []
        grads = []
        score = 0

        while True:

            probs = agent.policy(state)
            action_space = env.action_space.n
            action = np.random.choice(action_space, p=probs[0])

            next_state, reward, done,_ = env.step(action)
            next_state = next_state[None,:]
            next_state = agent.poly.fit_transform(next_state.reshape(1, 4))
            grad = agent.grad(probs, action, state)

            grads.append(grad)
            rewards.append(reward)

            score += reward
            state = next_state

            if done:
                break

        agent.update_with(grads, rewards)
        complete_scores.append(score)

    env.close()
    plt.plot(np.arange(NUM_EPISODES),
             complete_scores)
    plt.savefig('image1.png')


if __name__ == '__main__':
    main(None)

.

.

Questão

Estou tentando fazer, quase o mesmo exemplo, mas com a ativação do Sigmoid (apenas por simplicidade). É tudo o que preciso fazer. Alterne a ativação no modelo de softmaxpara sigmoid. O que deve funcionar com certeza (com base na explicação abaixo). Mas meu modelo de gradiente de política não aprende nada e é aleatório. Alguma sugestão?

import gym
import numpy as np
import matplotlib.pyplot as plt
from sklearn.preprocessing import PolynomialFeatures

NUM_EPISODES = 4000
LEARNING_RATE = 0.000025
GAMMA = 0.99


# noinspection PyMethodMayBeStatic
class Agent:
    def __init__(self):
        self.poly = PolynomialFeatures(1)
        self.w = np.random.rand(5, 1) - 0.5

    # Our policy that maps state to action parameterized by w
    # noinspection PyShadowingNames
    def policy(self, state):
        z = np.sum(state.dot(self.w))
        return self.sigmoid(z)

    def sigmoid(self, x):
        s = 1 / (1 + np.exp(-x))
        return s

    def sigmoid_grad(self, sig_x):
        return sig_x * (1 - sig_x)

    def grad(self, probs, action, state):
        dsoftmax = self.sigmoid_grad(probs)
        dlog = dsoftmax / probs
        grad = state.T.dot(dlog)
        grad = grad.reshape(5, 1)
        return grad

    def update_with(self, grads, rewards):
        if len(grads) < 50:
            return
        for i in range(len(grads)):
            # Loop through everything that happened in the episode
            # and update towards the log policy gradient times **FUTURE** reward

            total_grad_effect = 0
            for t, r in enumerate(rewards[i:]):
                total_grad_effect += r * (GAMMA ** r)
            self.w += LEARNING_RATE * grads[i] * total_grad_effect


def main(argv):
    env = gym.make('CartPole-v0')
    np.random.seed(1)

    agent = Agent()
    complete_scores = []

    for e in range(NUM_EPISODES):
        state = env.reset()[None, :]
        state = agent.poly.fit_transform(state)

        rewards = []
        grads = []
        score = 0

        while True:

            probs = agent.policy(state)
            action_space = env.action_space.n
            action = np.random.choice(action_space, p=[1 - probs, probs])

            next_state, reward, done, _ = env.step(action)
            next_state = next_state[None, :]
            next_state = agent.poly.fit_transform(next_state.reshape(1, 4))

            grad = agent.grad(probs, action, state)
            grads.append(grad)
            rewards.append(reward)

            score += reward
            state = next_state

            if done:
                break

        agent.update_with(grads, rewards)
        complete_scores.append(score)

    env.close()
    plt.plot(np.arange(NUM_EPISODES),
             complete_scores)
    plt.savefig('image1.png')


if __name__ == '__main__':
    main(None)

A plotagem de todo o aprendizado é aleatória. Nada ajuda no ajuste de hiper parâmetros. Abaixo da imagem de amostra.

Referências :

1) Aprendizado de reforço profundo: Pong from Pixels

2) Uma introdução aos Gradientes de Política com Cartpole e Doom

3) Obtenção de gradientes de política e implementação do REFORÇO

4) Truque do dia para aprendizado de máquina (5): Log Derivative Trick 12

ATUALIZAR

Parece que a resposta abaixo pode fazer algum trabalho com o gráfico. Mas não é a probabilidade de log e nem o gradiente da política. E altera todo o objetivo da Política de Gradiente da RL. Por favor, verifique as referências acima. Seguindo a imagem, próxima declaração.

Preciso fazer um Gradiente da função Log da minha Política (que é simplesmente pesos e sigmoidativação).

O problema está no gradmétodo.

def grad(self, probs, action, state):
    dsoftmax = self.sigmoid_grad(probs)
    dlog = dsoftmax / probs
    grad = state.T.dot(dlog)
    grad = grad.reshape(5, 1)
    return grad

No código original, Softmax foi usado junto com a função de perda CrossEntropy. Quando você alterna a ativação para Sigmoid, a função de perda adequada se torna Binário CrossEntropy. Agora, o objetivo do gradmétodo é calcular o gradiente da função de perda wrt. pesos. Poupando os detalhes, o gradiente adequado é dado (probs - action) * statena terminologia do seu programa. A última coisa é adicionar sinal de menos - queremos maximizar o negativo da função de perda.

gradMétodo adequado assim:

def grad(self, probs, action, state):
    grad = state.T.dot(probs - action)
    return -grad

Outra mudança que você pode querer adicionar é aumentar a taxa de aprendizado. LEARNING_RATE = 0.0001e NUM_EPISODES = 5000produzirá o seguinte gráfico:

A convergência será muito mais rápida se os pesos forem inicializados usando a distribuição Gaussiana com média zero e pequena variação:

def __init__(self):
    self.poly = PolynomialFeatures(1)
    self.w = np.random.randn(5, 1) * 0.01

ATUALIZAR

Adicionado código completo para reproduzir os resultados:

import gym
import numpy as np
import matplotlib.pyplot as plt
from sklearn.preprocessing import PolynomialFeatures

NUM_EPISODES = 5000
LEARNING_RATE = 0.0001
GAMMA = 0.99


# noinspection PyMethodMayBeStatic
class Agent:
    def __init__(self):
        self.poly = PolynomialFeatures(1)
        self.w = np.random.randn(5, 1) * 0.01

    # Our policy that maps state to action parameterized by w
    # noinspection PyShadowingNames
    def policy(self, state):
        z = np.sum(state.dot(self.w))
        return self.sigmoid(z)

    def sigmoid(self, x):
        s = 1 / (1 + np.exp(-x))
        return s

    def sigmoid_grad(self, sig_x):
        return sig_x * (1 - sig_x)

    def grad(self, probs, action, state):
        grad = state.T.dot(probs - action)
        return -grad

    def update_with(self, grads, rewards):
        if len(grads) < 50:
            return
        for i in range(len(grads)):
            # Loop through everything that happened in the episode
            # and update towards the log policy gradient times **FUTURE** reward

            total_grad_effect = 0
            for t, r in enumerate(rewards[i:]):
                total_grad_effect += r * (GAMMA ** r)
            self.w += LEARNING_RATE * grads[i] * total_grad_effect


def main(argv):
    env = gym.make('CartPole-v0')
    np.random.seed(1)

    agent = Agent()
    complete_scores = []

    for e in range(NUM_EPISODES):
        state = env.reset()[None, :]
        state = agent.poly.fit_transform(state)

        rewards = []
        grads = []
        score = 0

        while True:

            probs = agent.policy(state)
            action_space = env.action_space.n
            action = np.random.choice(action_space, p=[1 - probs, probs])

            next_state, reward, done, _ = env.step(action)
            next_state = next_state[None, :]
            next_state = agent.poly.fit_transform(next_state.reshape(1, 4))

            grad = agent.grad(probs, action, state)
            grads.append(grad)
            rewards.append(reward)

            score += reward
            state = next_state

            if done:
                break

        agent.update_with(grads, rewards)
        complete_scores.append(score)

    env.close()
    plt.plot(np.arange(NUM_EPISODES),
             complete_scores)
    plt.savefig('image1.png')


if __name__ == '__main__':
    main(None)