Cão Robô que aprende sobre seu corpo e sobre o ambiente via ARS

Cão Robô que aprende sobre seu corpo e sobre o

ambiente via ARS

 

Para realizarmos a implementação de uma inteligência artificial baseada em Augumented Random Search, usaremos de um exemplo bastante simples porém capaz de nos elucidar os principais pontos em destaque deste tipo específico de rede neural artificial intuitiva que é o de um cão (ou qualquer outro animal semelhante) onde o mesmo aprenderá sobre o seu corpo (como controlar o mesmo) e sobre o ambiente a ser explorado, de forma que o aprendizado realizado para seu corpo não interfere no aprendizado realizado para sua mente.

Raciocine que, como mencionado no capítulo teórico, nesse contexto o

processo de aprendizado para o corpo é contínuo e ciclo após ciclo de processamento o mesmo é cada vez mais refinado, uma vez que não se “desaprende” a caminhar. Porém, nosso Agente terá toda uma estrutura lógica onde sim, será aplicado o conceito teórico de aprendizado por reforço para uma rede neural artificial intuitiva, onde o mesmo por tentativa e erro, sistema de recompensas e penalidades baseadas em estados e ações aprenderá a realizar uma determinada tarefa.

Nesse simples exemplo, nosso Agente é um cão robô que inicialmente não sabe de absolutamente nada sobre sua estrutura, ele identificará cada estrutura de sua anatomia simulada, ciclos após ciclos de processamento o mesmo será capaz de ficar em pé e caminhar, a partir disso ele será submetido a um percurso em um ambiente simulado inclusive cheio de obstáculos, para que aprenda a caminhar perfeitamente superando obstáculos.

 

De forma resumida, o que faremos é:

 

- Importação das bibliotecas e módulos utilizados ao longo do código.

- Carregamento do ambiente de exemplo e definição de seus hyperparâmetros.

- Criação das políticas de tomadas de decisão que separarão os processos para o seu corpo e sua mente.

- Construção da arquitetura da rede neural artificial intuitiva e seu sistema de dependências.

- Criação dos estados iniciais de nosso Agente.

- Treinamento da IA para o seu devido propósito.

- Execução de nossa rede neural artificial intuitiva.

- Interpretação dos dados obtidos em processamento e em pós processamento.

 

Neste exemplo estaremos implementando a inteligência artificial de um

cão robô. Muito provavelmente você já deve ter visto, ao menos em vídeo, algum destes tipos de robô em particular, onde o mesmo quando ativado pela primeira vez fica realizando movimentos totalmente aleatórios com seus membros, depois progressivamente vai realizando a interação dos membros com o corpo, ficando em pé, aprendendo a caminhar em alguma direção e por fim até mesmo superar obstáculos pelo caminho.

Estes tipos de robô, inicialmente idealizados para fins militares (para

realizar atividades como carregar grandes volumes de equipamentos ou explorar ambientes sem pôr em risco um humano), hoje pode ser facilmente adaptado para os mais diversos propósitos.

O que não temos o costume de raciocinar sobre esses tipos de robô é que

não há uma programação fixa scriptada para cada ação do mesmo, mas sim uma inteligência artificial intuitiva onde o mesmo aprende sobre seu corpo, seu ambiente e sobre a tarefa a ser realizada.

 

Estaremos na sequência implementando passo-a-passo a inteligência artificial deste tipo de robô em particular. Assim como ao final do processo teremos um modelo funcional pronto para ser colocado em execução.

 

Instalação das Dependências

Para nosso exemplo, única biblioteca adicional que deverá ser instalada é

a PyBullet, por meio dela teremos o ambiente virtual ao qual nosso Agente estará inserido. O processo de instalação é feito como de costume, via Pip, através do comando pip install pybullet.

Partindo para o Código:

 

 

Como sempre, todo processo se inicia com as devidas importações das

bibliotecas, módulos e ferramentas que estaremos utilizando ao longo de nosso código. Única diferença das demais importações realizadas nos outros exemplos é que agora estamos importando os ambientes de simulação da biblioteca PyBullet por meio de seu módulo pybullet_envs.

 

Em seguida damos início a estrutura de código em si, primeiramente,

criando o que para alguns autores é chamado de hyperparâmetros, o que em outras palavras simplesmente nos remete aos parâmetros que poderão ser livremente configurados para testes de eficiência de nosso modelo.

Sendo assim, inicialmente é criada a classe Hp( ), dentro de si há um

simples método construtor __init__ e alguns objetos que definem alguns parâmetros. Como a essa altura você já deve estar familiarizado com a nomenclatura usual, vamos apenas revisar alguns pontos. Primeiro deles nb_steps define quantos ciclos estarão sendo realizados, o mesmo que quantas vezes todo o código era executado com seus reajustes dos pesos em modelos de redes neurais artificiais convencionais. Em seguida episode_length define um intervalo de execuções dos ciclos (lembrando que aqui, apenas para exemplo, nosso Agente irá encerrar sua atividade ao alcançar o seu objetivo, em situações reais ele permanece constantemente ligado e funcional, apenas processando novas instruções de novas tarefas a serem realizadas, mas seu processamento é contínuo).

Da mesma forma como em outros exemplos, learning_rate especifica a

taxa de aprendizado, de quantas em quantas amostras serão realizados os reajustes dos pesos; nb_directions e nb_best_directions aqui definem manualmente o número de ações a serem consideradas por ciclo, tenha em mente que agora múltiplas ações são realizadas em conjunto apenas por parte do corpo de nosso Agente se movendo, fora as várias possíveis ações a serem tomadas perante o ambiente.

Na sequência criamos uma estrutura condicional especial via assert, que

por sua vez realiza uma ou mais verificações considerando o tempo de execução, para se certificar de o número atribuído para nb_best_directions deve ser igual ou menos que o valor de nb_directions. Isso é feito porque é possível que todas as ações que foram tomadas estejam corretas em seu contexto, mas uma exceção será criada caso esse valor exceda o número padrão máximo de possíveis ações a serem tomadas por nosso Agente.

Dando sequência é criado o objeto noise com valor atribuído de 0.03,

este objeto define literalmente um ruído, um número específico de amostras aleatórias que serão enxertadas em meio aos estados e ações para eliminar vícios de processamento de nossa rede neural. Se a mesma estiver funcionando com o esperado irá imediatamente identificar esses ruídos e os eliminar de seu processamento.

Em seguida seed com valor 1 simplesmente é o gatilho inicial para a

execução dos devidos processamentos.

Por fim, para env_name passamos como atributo

‘HalfCheetahBulletEnv-v0’, o que fará que seja carregado nosso modelo de Agente assim como seu ambiente de exemplo.

Na sequência criamos uma classe que ficará responsável por uma série

 

de normalizações para os estados de nosso Agente. Inicialmente criamos a classe Normalizer( ), dentro da mesma é criado o método construtor __init__ que receberá um parâmetro para nb_inputs. Seguindo, são criados os objetos n que por sua vez recebe como atributo uma matriz de zeros a ser atualizada com os valores de nb_inputs; Da mesma forma mean, mean_diff e var são criadas da mesma forma.

 

Ainda dentro da classe Normalizer( ) criamos a função observe( ) que

recebe um objeto x. Dentro de sua estrutura de código é inicialmente definido que o valor da soma de n com ele mesmo deve ser igual a 1. Em seguida é criado o objeto last_mean que receberá o último valor atribuído para mean por meio da função copy( ); mean por sua vez, agora justificando seu nome que significa média, recebe como atributo o valor da divisão da soma de seu valor por ele mesmo subtraído de x, dividido pelo valor de n. Note que aqui o que é feito é a simples média dos valores de mean para atualizar o valor de last_mean.

Na sequência é feito algo parecido para mean_diff, mas agora, para

obtermos o valor da diferença entre essas médias calculadas, mean_diff recebe como valor a soma de seu valor pela multiplicação de last_mean e mean subtraídos de x. Uma vez que temos os valores das médias e da diferença entre elas, nosso objeto var recebe como atributo o resultado da divisão entre mean_diff e n, dentro de um intervalo mínimo pré-estabelecido entre 1 e -2 via função clip( ).

 

Ainda indentado para Normalize( ) é criada a função normalize( ) que

recebe inputs como parâmetro. Dentro de si o objeto obs_mean recebe como atributo o valor de mean, da mesma forma obs_std recebe o valor da raiz quadrada de var, finalizando retornando o valor da divisão entre a subtração dos valores de inputs e obs_mean para obs_std.

 

Logo após é iniciada a criação da estrutura voltada para a inteligência

artificial em si, para isso inicialmente é criada a classe Policy( ), dentro de si é criado um método construtor __init__ que recebe input_size e output_size, finalizando, é criado o objeto theta que por sua vez recebe como atributo inicialmente uma matriz de zeros a ser preenchida com os valores de output_size e de input_size, respectivamente.

Seguindo com o código, indentado para Policy( ) criamos a função

evaluate( ) que recebe um dado/valor para input e já define manualmente que delta e direction são inicializados em None, ou seja, sem nenhum dado atribuído, apenas reservando essas variáveis para um propósito futuro.

Dentro de evaluate( ) é criada uma estrutura condicional onde se o

estado de direction for None, é retornado o valor do produto escalar de theta por meio da função dot( ) parametrizada com os valores de input. Caso essa primeira condição não for verdadeira, é verificado então se o dado atribuído para direction é “positive”, caso sim, é retornado o produto escalar da soma entre os valores de theta e noise multiplicado por delta, equiparado a input. Por fim, caso essa condição não seja verdadeira, é retornado a subtração entre theta e a multiplicação entre noise e delta, equiparado a input.

 

Ainda em Policy( ), é criada a função sample_deltas, que retorna uma

lista preenchida com os valores lidos para nb_directions convertidos para o formato de theta.

Encerrando esse bloco é criada a função update( ) que receberá rollouts e

sigma_r. Dentro de sua estrutura de código é criado o objeto step, inicialmente parametrizado com uma matriz de zeros no formato de theta. Em seguida é criado um laço de repetição onde para as variáveis temporárias r_pos, r_neg e d em rollouts, step tem seu valor atualizado com o valor da soma entre seu valor atual e a diferença entre r_pos e r_neg, multiplicado pelo valor de d. Finalizando, theta também tem seu valor atualizado, recebendo o valor da divisão entre leaning_rate e nb_best_directions multiplicado por sigma_r, multiplicado pelo valor de step.

Vamos entender a lógica do que está acontecendo aqui, lembre-se que

conceitualmente a ideia é termos o processo de aprendizado por reforço as ações para o corpo e separadamente para a mente de nosso Agente. A nível de código isso é feito reutilizando estados do corpo para novos processamentos da mente do mesmo.

Para alguns autores, o mecanismo de rollouts é caracterizado desta maneira, nosso agente em um determinado estado, quando penalizado, reaproveitará todo processamento realizado para o corpo, mudando somente as diretrizes lógicas em busca de recompensa positiva. Como dito anteriormente no embasamento teórico, nosso Agente está caminhando de um ponto A para um ponto B, quando o mesmo erra uma ação, ele não precisa desaprender a caminhar, mas apenas desaprender a ação lógica incorreta.

Seguindo com nosso código é criada a função explore( ) que receberá

env, normalizer, policy e define que direction e delta são instanciadas, porém inicialmente sem nenhum dado.

Dentro de sua estrutura de código é criado o objeto state que por sua vez,

devido a sequência da leitura léxica por parte do interpretador, reseta todos estados de nosso Agente por meio da função reset( ).

Na mesma lógica done é iniciado como False, já que seu estado será

alterado para True apenas quando a tarefa a ser realizada por nosso Agente for concluída; num_plays e sum_rewards são inicializados com valor 0, pois no estado inicial ainda não há uma memória criada.

Em seguida é criada uma estrutura condicional onde enquanto nem o

valor de done nem o de num_plays for menor que episode_length, os objetos normalizer.observe e state tem seus valores atualizados com os dados oriundos de state. Da mesma forma action realiza alguns testes por meio da função evaluate( ) por sua vez parametrizada com os dados de state, delta e direction.

Na sequência os objetos state, reward, done e a variável vazia _ chamam

a função step( ) parametrizada com os dados de action; reward por sua vez recebe o valor máximo atribuído para o mínimo de reward, isso faz com que se crie um intervalo de amostras a serem consideradas, tentando obter amostras dos melhores valores encontrados.

Finalizando, sum_rewards é atualizado com o seu próprio valor somado

com o de reward, o mesmo é feito para num_plays, que por sua vez tem seu valor atualizado pela soma de seu próprio valor acrescido de 1. Por fim é retornado o valor de sum_rewards.

Repare que o que é feito aqui nada mais é do que fazer o uso daquela

política de tomada de decisão baseada no último estado, porém como não havia um estado anterior, é necessário criar um estado, mesmo que com valores zerados, para que se possa ser atualizado.

Uma vez criada toda a estrutura base, podemos agora de fato treinar a AI

 

de nosso Agente por meio de algumas sequências de cruzamento de dados. Para isso inicialmente criamos a função train( ) que recebe env, policy, normalizer e hp.

Dentro do bloco de código associado a train( ) já de início temos um laço

de repetição onde para cada step em nb_steps uma série de atualizações será realizada. Primeira delas para o objeto deltas é atribuído os valores padrão de policy.sample_deltas( ), a serem importados dos arquivos de exemplo, normalmente uma matriz preenchida de zeros. Em seguida para positive_rewards e para negative_rewards é atribuído o valor de seu índice zero multiplicado pelo valor de nb_directions.

Na sequência, após alocar espaço para estes dados, hora de preenche-los,

e isso será feito por um segundo laço de repetição onde uma variável temporária k percorre toda a extensão de nb_directions, atualizando os valores de positive_rewards por meio da função explore( ) parametrizada com env, normalizer, policy e especificando manualmente que para este objeto direction será “positive” e que delta receberá seus valores na posição k. O mesmo é feito com um terceiro laço de repetição, porém processando dados para negative_rewards.

Finalizando esse bloco mais algumas atualizações devem ser feitas,

inicialmente o objeto all_rewards recebe atribuído para si uma matriz composta dos dados da soma entre positive_rewards e negative_rewards; sigma_r recebe para si os valores da média absoluta de all_rewards.

Em seguida scores recebe por sua vez um dicionário alimentado com os

valores máximos de cada posição k. Em função disso é aplicado um laço de repetição em que k percorrerá cada valor de positive_rewards e de negative_rewards.

Na sequência order recebe para si como atributos os valores encontrados

para scores e nb_best_directions, respectivamente ordenados dos valores maiores para os menores referentes as suas chaves. Da mesma forma rollouts remonta sua lista, agora com os valores ordenados de positive_rewards, negative_rewards e deltas. Em outras palavras aqui serão considerados os maiores valores encontrados pois estes, nos processos de reajustes dos pesos terão maior peso no processamento em contraponto a outros dados de menor relevância que poderão ser descartados conforme escolha do supervisor.

Encerrando esse bloco, é executada a função update( ) de policy,

repassando como parâmetros os dados de rollouts e de sigma_r. Método semelhante é feito para reward_evaluation que chama a função explore repassando para mesma como parâmetros os dados de env, normalizer e policy.

Por fim é exibido em console cada estado e sua respectiva recompensa

ou penalidade, instanciando tais dados de step e de reward_evaluation, respectivamente.

Em seguida é realizada uma pequena codificação para criação de um

diretório onde serão salvos algumas amostras de nosso modelo em operação.

 

Terminando esse processo, são feitas algumas últimas definições e instancias de alguns objetos com seus “hyperparâmetros”. Se até esse momento não houve nenhum erro de sintaxe, será possível acompanhar via console o andamento de nosso Agente, assim como ao final do processo será possível visualizar várias amostras referentes a várias etapas do processo de aprendizado por reforço de nosso Agente, em formato de vídeo, no diretório criado anteriormente.

Código Completo:

 

import os

import numpy as np

import gym

from gym import wrappers

import pybullet_envs

 

class Hp():

def __init__(self):

self.nb_steps = 1000

self.episode_length = 1000

self.learning_rate = 0.02

self.nb_directions = 16

self.nb_best_directions = 16

assert self.nb_best_directions <= self.nb_directions

self.noise = 0.03

self.seed = 1

self.env_name = 'HalfCheetahBulletEnv-v0'

 

class Normalizer():

def __init__(self, nb_inputs):

self.n = np.zeros(nb_inputs)

self.mean = np.zeros(nb_inputs)

self.mean_diff = np.zeros(nb_inputs)

self.var = np.zeros(nb_inputs)

 

def observe(self, x):

self.n += 1.

last_mean = self.mean.copy()

self.mean += (x - self.mean) / self.n

self.mean_diff += (x - last_mean) * (x - self.mean)

self.var = (self.mean_diff / self.n).clip(min = 1e-2)

 

def normalize(self, inputs):

obs_mean = self.mean

obs_std = np.sqrt(self.var)

return (inputs - obs_mean) / obs_std

 

class Policy():

def __init__(self, input_size, output_size):

self.theta = np.zeros((output_size, input_size))

 

def evaluate(self, input, delta = None, direction = None):

if direction is None:

return self.theta.dot(input)

elif direction == "positive":

return (self.theta + hp.noise*delta).dot(input)

else:

return (self.theta - hp.noise*delta).dot(input)

 

def sample_deltas(self):

return [np.random.randn(*self.theta.shape) for _ in range(hp.nb_directions)]

 

def update(self, rollouts, sigma_r):

step = np.zeros(self.theta.shape)

for r_pos, r_neg, d in rollouts:

step += (r_pos - r_neg) * d

self.theta += hp.learning_rate / (hp.nb_best_directions * sigma_r) * step

 

def explore(env, normalizer, policy, direction = None, delta = None):

state = env.reset()

done = False

num_plays = 0.

sum_rewards = 0

while not done and num_plays < hp.episode_length:

normalizer.observe(state)

state = normalizer.normalize(state)

action = policy.evaluate(state, delta, direction)

state, reward, done, _ = env.step(action)

reward = max(min(reward, 1), -1)

sum_rewards += reward

num_plays += 1

return sum_rewards

 

def train(env, policy, normalizer, hp):

for step in range(hp.nb_steps):

deltas = policy.sample_deltas()

positive_rewards = [0] * hp.nb_directions

negative_rewards = [0] * hp.nb_directions

for k in range(hp.nb_directions):

positive_rewards[k] = explore(env, normalizer, policy,

direction = "positive", delta = deltas[k])

for k in range(hp.nb_directions):

negative_rewards[k] = explore(env, normalizer, policy,

direction = "negative", delta = deltas[k])

all_rewards = np.array(positive_rewards + negative_rewards)

sigma_r = all_rewards.std()

scores = {k:max(r_pos, r_neg) for k, (r_pos,r_neg) in enumerate(zip(positive_rewards, negative_rewards))}

order = sorted(scores.keys(), key = lambda x:scores[x]) [:hp.nb_best_directions]

rollouts = [(positive_rewards[k], negative_rewards[k], deltas[k])

policy.update(rollouts, sigma_r)

reward_evaluation = explore(env, normalizer, policy)

print('Step:', step, 'Reward:', reward_evaluation)

 

def mkdir(base, name):

path = os.path.join(base, name)

if not os.path.exists(path):

os.makedirs(path)

return path

work_dir = mkdir('exp', 'brs')

monitor_dir = mkdir(work_dir, 'monitor')

 

hp = Hp()

np.random.seed(hp.seed)

env = gym.make(hp.env_name)

env = wrappers.Monitor(env, monitor_dir, force = True) nb_inputs = env.observation_space.shape[0] nb_outputs = env.action_space.shape[0] policy = Policy(nb_inputs, nb_outputs)

normalizer = Normalizer(nb_inputs)

train(env, policy, normalizer, hp)