This is a Python implementation of the method presented in the paper "Neuronlike Adaptive Elements That Can Solve Difficult Learning Control Problems", published in 1983.
The original C implementation written by the authors can be found here.
The "ASE" and "ACE" are neurons which:
- have a set of weights which determins the output of the element
- ASE's weight determine the outputed action
- ACE's weights determine the corrected reward
- update weights according to a
- reward input
- exponentially-decaying eligibility trace
In the paper the raw observation vectors are first passed to a decoder which creates the $x$ seen below. $x$ is one-hot encoded.
So even though the below formulas are written in terms of vectors and dot products, anything involving $x$ is really just an index.
e.g. $\mathbf{w}^\intercal \mathbf{x}$
is equivalent to
"Let i be the index where there is a 1 in w. Return x[i]"
Decision function of the ASE:¶
$ y(t) = \text{sign}(\mathbf{w}^\intercal \mathbf{x} + \text{noise}(t))$
Update rules:¶
$\mathbf{w}(t+1) = \mathbf{w} + \alpha r(t) \mathbf{e}(t) $
$\mathbf{e}(t+1) = \delta \mathbf{e}(t) + (1-\delta) y(t)\mathbf{x}(t) $
ACE Output (an improved prediction):¶
$\hat{r}(t) = r(t) + \gamma p(t) - p(t-1)$
where $p$ is the prediction of the reward $p(t) = \mathbf{v}^\intercal \mathbf{x}$
Update rules:¶
$\mathbf{v}(t+1) = \mathbf{v}(t) + \beta{\hat{r}}\mathbf{\bar{x}}(t)$
$\mathbf{\bar{x}}(t+1) = \lambda\mathbf{\bar{x}}(t) + (1-\lambda) \mathbf{x}$
import numpy as np
import numpy as np
from sklearn.utils import check_random_state
class ASE(object):
def __init__(self, n_input_dim, learning_rate, decay_rate, sigma=0.01, random_state=None):
self.n_input_dim = n_input_dim
self.learning_rate = learning_rate
self.decay_rate = decay_rate
self.sigma = sigma
self.random_state = random_state
self.reset_values()
def reset_values(self):
# Weights
self.w = np.zeros(self.n_input_dim)
# Eligibility
self.e = np.zeros(self.n_input_dim)
self.rs = check_random_state(self.random_state)
def step(self, x, reward=None):
'''
Updates the weights and eligibility trace.
Parameters
----------
'x' : the state vector
'reward' : the reward value
Returns
-------
Action : 0 or 1
'''
# Probabilistic action
action = self.w.dot(x) + self.rs.randn()*self.sigma
action = np.clip(action, -50, 50)
action = int((1 + np.exp(-action)) ** (-1) > 0.5)
if reward is not None:
self.w += self.learning_rate * reward * self.e
self.e *= self.decay_rate
self.e += (1. - self.decay_rate) * (action*2 - 1) * x
return action
class ACE(object):
def __init__(self, n_input_dim, learning_rate, decay_rate, discount_factor, random_state=None):
self.n_input_dim = n_input_dim
self.learning_rate = learning_rate
self.decay_rate = decay_rate
self.discount_factor = discount_factor
self.random_state = random_state
self.reset_values()
def reset_values(self):
self.rs = check_random_state(self.random_state)
self.w = np.zeros(self.n_input_dim)
self.trace = np.zeros(self.n_input_dim)
self.prev_p = 0.
def step(self, x, reward=None, done=False):
if done:
p = 0.
else:
p = self.w.dot(x)
if reward is None:
self.prev_p = p
return None
else:
revised_reward = reward + self.discount_factor * p - self.prev_p
self.w += self.learning_rate * revised_reward * self.trace
self.trace *= self.decay_rate
self.trace += (1. - self.decay_rate) * x
self.prev_p = p
return revised_reward
ONE_DEGREE = 1. * np.pi / 180
SIX_DEGREES = 6. * np.pi / 180
FIFTY_DEGREES = 50. * np.pi / 180
def get_box(observation):
x, x_dot, theta, theta_dot = observation
box=0
bin_edges = [[-0.8, 0.8],
[-0.5, 0.5],
[-SIX_DEGREES, -ONE_DEGREE, 0, ONE_DEGREE, SIX_DEGREES],
[-FIFTY_DEGREES, FIFTY_DEGREES]]
box = 0
for s, edges in zip(observation, bin_edges):
i = np.digitize([s], edges)[0]
box = box * (len(edges)+1) + i
vec = np.zeros(162)
vec[box] = 1.
return vec
import gym
N_TRIALS = 150
MAX_STEPS = 100000
TERMINATE_ON_MAX_STEPS = True
SEED = 12345
env = gym.make('CartPole-v0')
ase = ASE(n_input_dim=162, learning_rate=1000, decay_rate=0.9, random_state=SEED)
ace = ACE(n_input_dim=162, learning_rate=0.5, decay_rate=0.8, discount_factor=0.95, random_state=SEED)
for trial in range(1, N_TRIALS+1):
obs = env.reset()
reward = None
done = False
for t in range(1, MAX_STEPS+1):
x = get_box(obs)
revised_reward = ace.step(x, reward, done)
#print reward, revised_reward
action = ase.step(x, revised_reward)
if done:
break
obs, _, done, _ = env.step(action)
if done:
reward = -1
else:
reward = 0
if trial % 10 == 0 :
print "Trial {}: survived {} steps".format(trial, t)
if TERMINATE_ON_MAX_STEPS and t == MAX_STEPS:
print "Terminated after {} trials. Successfully balanced pole for MAX_STEPS={}".format(trial, MAX_STEPS)
break
This method succesfully solves Cart-Pole in under 100 iterations!
What if I didn't use any boxes? What if I just used the raw 4-element observation vector??¶
The formulas are all writen in terms of vectors and dot products so maybe I can skip the state-space, one-hot encoding and just pass the raw observation vector?
import gym
N_TRIALS = 1500
MAX_STEPS = 100000
TERMINATE_ON_MAX_STEPS = True
SEED = 12345
env = gym.make('CartPole-v0')
ase = ASE(n_input_dim=4, learning_rate=10, decay_rate=0.5, random_state=SEED)
ace = ACE(n_input_dim=4, learning_rate=0.01, decay_rate=0.5, discount_factor=0.05, random_state=SEED)
for trial in range(1,N_TRIALS+1):
obs = env.reset()
reward = None
done = False
for t in range(1, MAX_STEPS+1):
x = obs
revised_reward = ace.step(x, reward, done)
#print reward, revised_reward
action = ase.step(x, revised_reward)
if done:
break
obs, _, done, _ = env.step(action)
if done:
reward = -1
else:
reward = 0
if trial % 10 == 0 :
print "Trial {}: survived {} steps".format(trial, t)
if TERMINATE_ON_MAX_STEPS and t == MAX_STEPS:
print "Terminated after {} trials. Successfully balanced pole for MAX_STEPS={}".format(trial, MAX_STEPS)
break
Nope. It doesn't work. It seems this method is too strongly reliant on the input vector being a one-hot encoding of discrete states. This is not surprising since the paper was written as a successor to another method which had the name "Boxes".
from itertools import product
import numpy as np
import gym
GRANULARITY = 8
env = gym.make('CartPole-v0')
# Split each dimension up into `GRANULARITY` number of buckets
# store these buckets in `linspaces`
bucket_edges = []
ranges = zip(env.observation_space.low, env.observation_space.high)
for low, high in ranges:
if np.isneginf(low):
low = -5
if np.isinf(high):
high = 5
bucket_edges.append(np.linspace(low, high, num=GRANULARITY+1, endpoint=True)[1:-1])
bucket_edges
def get_box(observation):
x, x_dot, theta, theta_dot = observation
box=0
bin_edges = bucket_edges
box = 0
for s, edges in zip(observation, bin_edges):
i = np.digitize([s], edges)[0]
box = box * (len(edges)+1) + i
vec = np.zeros(GRANULARITY**observation.shape[0])
vec[box] = 1.
return vec
import gym
N_TRIALS = 1000
MAX_STEPS = 100000
TERMINATE_ON_MAX_STEPS = True
SEED = 12345
env = gym.make('CartPole-v0')
ase = ASE(n_input_dim=GRANULARITY**4, learning_rate=1000, decay_rate=0.9, random_state=SEED)
ace = ACE(n_input_dim=GRANULARITY**4, learning_rate=0.5, decay_rate=0.8, discount_factor=0.95, random_state=SEED)
for trial in range(1, N_TRIALS):
obs = env.reset()
reward = None
done = False
for t in range(1, MAX_STEPS+1):
x = get_box(obs)
revised_reward = ace.step(x, reward, done)
#print reward, revised_reward
action = ase.step(x, revised_reward)
if done:
break
obs, _, done, _ = env.step(action)
if done:
reward = -1
else:
reward = 0
if trial % 10 == 0 :
print "Trial {}: survived {} steps".format(trial, t)
if TERMINATE_ON_MAX_STEPS and t == MAX_STEPS:
print "Terminated after {} trials. Successfully balanced pole for MAX_STEPS={}".format(trial, MAX_STEPS)
break
Their method works with this very rough discretization but (unsurprisingly) it takes much longer to learn.
No comments :
Post a Comment