wassname/pyrobolearn
1
0
Fork 0
mirror of https://github.com/wassname/pyrobolearn.git synced 2026-09-11 12:31:07 +08:00
Code Issues Packages Projects Releases Wiki Activity

refactor estimators/returns/targets

Browse Source
This commit is contained in:
Brian Delhaisse
2019-04-15 14:28:40 +02:00
parent 4f8e791dc7
commit 4a391ed77a
16 changed files with 633 additions and 301 deletions
Download Patch File Download Diff File Expand all files Collapse all files
pyrobolearn/algos/ddpg.py
+2 -2
View File
@@ -14,7 +14,7 @@ from pyrobolearn.exploration import ActionExploration, GaussianActionExploration
from pyrobolearn.storages import ExperienceReplay
from pyrobolearn.samplers import BatchRandomSampler
from pyrobolearn.estimators import TDQValueReturn
from pyrobolearn.returns import TDQValueReturn
from pyrobolearn.losses import MSBELoss, QLoss
from pyrobolearn.optimizers import Adam
@@ -272,7 +272,7 @@ class DDPG(GradientRLAlgo):
exploration = ActionExploration(policy=policy, action=policy.actions)
# create experience replay
storage = ExperienceReplay(observation_shapes=policy.states, action_shapes=policy.actions, capacity=capacity)
storage = ExperienceReplay(state_shapes=policy.states, action_shapes=policy.actions, capacity=capacity)
sampler = BatchRandomSampler(storage)
# create target return estimator
pyrobolearn/algos/dqn.py
+1 -1
View File
@@ -16,7 +16,7 @@ from pyrobolearn.values import ParametrizedQValueOutput
from pyrobolearn.exploration import EpsilonGreedyActionExploration
from pyrobolearn.storages import ExperienceReplay
from pyrobolearn.estimators import TDQLearningReturn
from pyrobolearn.returns import TDQLearningReturn
from pyrobolearn.losses import MSBELoss, HuberLoss
from pyrobolearn.optimizers import Adam
pyrobolearn/algos/evaluator.py
+3 -3
View File
@@ -1,11 +1,11 @@
#!/usr/bin/env python
"""Provide the Evaluator class used in the second step of RL algorithms
The evaluator assesses the quality of the actions/trajectories performed by the policy using the given estimators.
The evaluator assesses the quality of the actions/trajectories performed by the policy using the given returns.
It is the step performed after the exploration phase, and before the update step.
"""
from pyrobolearn.estimators import Estimator
from pyrobolearn.returns import Estimator
__author__ = "Brian Delhaisse"
__copyright__ = "Copyright 2018, PyRoboLearn"
@@ -23,7 +23,7 @@ class Evaluator(object):
(Model-free) reinforcement learning algorithms requires 3 steps:
1. Explore: Explore and collect samples in the environment using the policy. The samples are stored in the
given memory/storage unit.
2. Evaluate: Assess the quality of the actions/trajectories using the estimators.
2. Evaluate: Assess the quality of the actions/trajectories using the returns.
3. Update: Update the policy (and/or value function) parameters based on the loss
This class focuses on the second step of RL algorithms.
pyrobolearn/algos/explorer.py
+1 -1
View File
@@ -31,7 +31,7 @@ class Explorer(object):
(Model-free) reinforcement learning algorithms requires 3 steps:
1. Explore: Explore and collect samples in the environment using the policy. The samples are stored in the
given memory/storage unit.
2. Evaluate: Assess the quality of the actions/trajectories using the estimators.
2. Evaluate: Assess the quality of the actions/trajectories using the returns.
3. Update: Update the policy (and/or value function) parameters based on the loss
This class focuses on the first step of RL algorithms. It accepts the environment, and the exploration strategy
pyrobolearn/algos/ppo.py
+3 -3
View File
@@ -16,7 +16,7 @@ from pyrobolearn.exploration import ActionExploration
from pyrobolearn.storages import RolloutStorage
from pyrobolearn.samplers import BatchRandomSampler
from pyrobolearn.estimators import GAE
from pyrobolearn.returns import GAE
from pyrobolearn.losses import CLIPLoss, ValueLoss, EntropyLoss
from pyrobolearn.optimizers import Adam
@@ -40,7 +40,7 @@ class PPO(GradientRLAlgo):
This class implements the PPO algorithm which was presented in [1], and is inspired on the implementation of [2, 3].
Compared to [2, 3], the algorithm is made such that it is more modular and flexible by decoupling and defining the
various concepts (storages, losses, estimators / returns, policies, value function approximators, and others)
various concepts (storages, losses, returns / returns, policies, value function approximators, and others)
outside the PPO class, and providing them as input to the constructor and thus privileging composition over
inheritance.
@@ -183,7 +183,7 @@ class PPO(GradientRLAlgo):
# create storage and estimator
states, actions = policy.states, policy.actions
logger.debug('create rollout storage')
storage = RolloutStorage(num_steps=1000, observation_shapes=states.merged_shape,
storage = RolloutStorage(num_steps=1000, state_shapes=states.merged_shape,
action_shapes=actions.merged_shape, num_processes=num_workers)
logger.debug('create return estimator (GAE)')
estimator = GAE(storage, gamma=gamma, tau=tau)
pyrobolearn/algos/reinforce.py
+2 -2
View File
@@ -16,7 +16,7 @@ from pyrobolearn.exploration import ActionExploration
from pyrobolearn.storages import RolloutStorage
from pyrobolearn.samplers import StorageSampler
from pyrobolearn.estimators import ActionRewardEstimator
from pyrobolearn.returns import ActionRewardEstimator
from pyrobolearn.losses import PGLoss, ValueLoss
from pyrobolearn.optimizers import Adam
@@ -188,7 +188,7 @@ class REINFORCE(GradientRLAlgo):
# create storage
states, actions = policy.states, policy.actions
storage = RolloutStorage(num_steps=1000, observation_shapes=states.shape, action_shapes=actions.shape,
storage = RolloutStorage(num_steps=1000, state_shapes=states.shape, action_shapes=actions.shape,
num_processes=num_workers)
sampler = StorageSampler(storage)
pyrobolearn/algos/sac.py
+3 -3
View File
@@ -16,7 +16,7 @@ from pyrobolearn.exploration import ActionExploration, GaussianActionExploration
from pyrobolearn.storages import ExperienceReplay
from pyrobolearn.samplers import BatchRandomSampler
from pyrobolearn.estimators import TDQValueReturn
from pyrobolearn.returns import TDQValueReturn
from pyrobolearn.losses import MSBELoss, QLoss
from pyrobolearn.optimizers import Adam
@@ -302,14 +302,14 @@ class SAC(GradientRLAlgo):
value_target = copy.deepcopy(value)
# create experience replay
storage = ExperienceReplay(observation_shapes=policy.states, action_shapes=policy.actions, capacity=capacity)
storage = ExperienceReplay(state_shapes=policy.states, action_shapes=policy.actions, capacity=capacity)
sampler = BatchRandomSampler(storage)
# create action exploration
exploration = ActionExploration(policy)
# create targets
# TODO
# q_target =
# create losses
pyrobolearn/algos/td3.py
+2 -2
View File
@@ -14,7 +14,7 @@ from pyrobolearn.exploration import ActionExploration, GaussianActionExploration
from pyrobolearn.storages import ExperienceReplay
from pyrobolearn.samplers import BatchRandomSampler
from pyrobolearn.estimators import TDQValueReturn
from pyrobolearn.returns import TDQValueReturn
from pyrobolearn.losses import MSBELoss, QLoss
from pyrobolearn.optimizers import Adam
@@ -212,7 +212,7 @@ class TD3(GradientRLAlgo):
exploration = ActionExploration(policy=policy, action=policy.actions)
# create experience replay
storage = ExperienceReplay(observation_shapes=policy.states, action_shapes=policy.actions, capacity=capacity)
storage = ExperienceReplay(state_shapes=policy.states, action_shapes=policy.actions, capacity=capacity)
sampler = BatchRandomSampler(storage)
# create target return estimator
pyrobolearn/algos/updater.py
+1 -1
View File
@@ -35,7 +35,7 @@ class Updater(object):
(Model-free) reinforcement learning algorithms requires 3 steps:
1. Explore: Explore and collect samples in the environment using the policy. The samples are stored in the
given memory/storage unit.
2. Evaluate: Assess the quality of the actions/trajectories using the estimators.
2. Evaluate: Assess the quality of the actions/trajectories using the returns.
3. Update: Update the policy (and/or value function) parameters based on the loss
This class focuses on the third step of RL algorithms.
pyrobolearn/estimators/__init__.py
-3
View File
@@ -1,3 +0,0 @@
# import the various estimators
from .estimator import *
pyrobolearn/losses/value_losses.py
+1 -1
View File
@@ -5,7 +5,7 @@
import torch
from pyrobolearn.losses.loss import Loss
from pyrobolearn.estimators.estimator import TDReturn
from pyrobolearn.returns import TDReturn
__author__ = "Brian Delhaisse"
__copyright__ = "Copyright 2018, PyRoboLearn"
pyrobolearn/estimators/README.md → pyrobolearn/returns/README.md
+3 -2
View File
@@ -1,6 +1,7 @@
## Estimators / Returns
## Returns / Estimators
This folder provides the various estimators / returns used in reinforcement learning (in the evaluation step of RL algorithms), such as:
This folder provides the various returns / estimators used in reinforcement learning (in the evaluation step of RL
algorithms), such as:
- Total reward estimator
- Action reward estimator
- Baseline reward estimator
pyrobolearn/returns/__init__.py
+9
View File
@@ -0,0 +1,9 @@
# import the various returns
from .estimator import *
# import the various targets
from .targets import *
# import the various returns
from .returns import *
pyrobolearn/estimators/estimator.py → pyrobolearn/returns/estimator.py
+7 -277
View File
@@ -1,6 +1,11 @@
#!/usr/bin/env python
"""Describes the various `Estimators` / `Returns` used in reinforcement learning.
- Estimators are evaluated on trajectories (i.e. RolloutStorage).
- Returns are estimated on batches of transition tuples (s, a, s', r, d).
In both case, they add
Dependencies:
- `pyrobolearn.storages`
"""
@@ -138,7 +143,7 @@ class Estimator(BaseReturn):
@property
def states(self):
"""Return the states / observations from the rollout storage."""
return self.storage.observations
return self.storage.states
###########
# Methods #
@@ -430,7 +435,7 @@ class GAE(Estimator):
super(GAE, self).__init__(storage=storage, gamma=gamma)
self.tau = tau
def _evaluate(self): # , next_value):
def _evaluate(self):
"""Evaluate the estimator / return."""
# self.values[-1] = next_value
gae = 0
@@ -438,278 +443,3 @@ class GAE(Estimator):
delta = self.rewards[t] + self.gamma * self.values[t + 1] * self.masks[t + 1] - self.values[t]
gae = delta + self.gamma * self.tau * self.masks[t + 1] * gae
self.returns[t] = gae + self.values[t]
class Return(BaseReturn):
r"""Return
Compared to the estimator that used the whole trajectory, it only uses transition tuples
:math:`(s_t, a_t, s_{t+1}, r_t, d)`, where is :math:`s_t` is the state at time :math:`t`, :math:`a_t` is the action
outputted by the policy in response to the state :math:`s_t`, :math:`s_{t+1}` is the next state returned by the
environment due to the policy's action :math:`a_t` and the current state :math:`s_t`, :math:`r_t` is the reward
signal returned by the environment, and :math:`d` is a boolean value that specifies if the task is over or not
(i.e. if it has failed or succeeded).
That is, estimators are estimated on the Monte-Carlo trajectories, while returns are estimated on temporal
difference errors [1].
References:
[1] "Reinforcement Learning: an Introduction" (chap 8.13), Sutton and Barto, 2018
"""
def __init__(self, gamma=1.):
"""
Initialize the return / estimator.
Args:
gamma (float): discount factor
"""
super(Return, self).__init__(gamma)
class TDReturn(Return):
r"""TD Return
Return based on the one-step temporal difference TD(0).
"""
pass
class Target(BaseReturn):
def compute(self, batch):
pass
def __call__(self, batch):
return self.compute(batch)
class ValueTarget(Target):
r"""Value target.
Compute the value target given by :math:`(r + \gamma (1-d) \min_i V_{\phi_i}(s'))`, where the index `i` is in
the case there are multiple value function approximators given to this class.
"""
def __init__(self, values, gamma=1.):
"""
Initialize the state value target.
Args:
values (Value, list of Value): state value function(s).
gamma (float): discount factor
"""
super(ValueTarget, self).__init__(gamma)
if not isinstance(values, collections.Iterable):
values = [values]
for i, value in enumerate(values):
if not isinstance(value, Value):
raise TypeError('The {}th value is not an instance of `Value`, instead got: {}'.format(i, type(value)))
self._values = values
def compute(self, batch):
r"""
Compute the value target :math:`(r + \gamma (1-d) \min_i V_{\phi_i}(s'))`
Args:
batch (Batch): batch containing the transitions.
"""
value = torch.min(torch.cat([value(batch['states']) for value in self._values], dim=1), dim=1)[0]
batch[self] = batch['rewards'] + self.gamma * (1 - batch['masks']) * value
return batch
class QValueTarget(Target):
r"""Q-Value target.
Compute the Q-value target given by :math:`(r + \gamma (1-d) \min_i Q_{\phi_i}(s',a'))`, where the index `i` is in
the case there are multiple Q-value function approximators given to this class.
"""
def __init__(self, q_values, gamma=1.):
"""
Initialize the state value target.
Args:
q_values (QValue, list of QValue): state-action value function(s).
gamma (float): discount factor
"""
super(QValueTarget, self).__init__(gamma)
if not isinstance(q_values, collections.Iterable):
q_values = [q_values]
for i, value in enumerate(q_values):
if not isinstance(value, QValue):
raise TypeError('The {}th value is not an instance of `QValue`, instead got: {}'.format(i, type(value)))
self._q_values = q_values
def compute(self, batch):
r"""
Compute the value target :math:`(r + \gamma (1-d) \min_i Q_{\phi_i}(s',a'))`
Args:
batch (Batch): batch containing the transitions.
"""
value = torch.min(torch.cat([value(batch['states']) for value in self._q_values], dim=1), dim=1)[0]
batch[self] = batch['rewards'] + self.gamma * (1 - batch['masks']) * value
return batch
class QLearningTarget(Target):
r"""Q-Learning target.
Compute the Q-value target given by :math:`(r + \gamma (1-d) \min_i \max_{a'} Q_{\phi_i}(s',a'))`, where the
index `i` is in the case there are multiple Q-value function approximators given to this class.
"""
def __init__(self, q_values, gamma=1.):
"""
Initialize the state value target.
Args:
q_values (QValue, list of QValue): state-action value function(s).
gamma (float): discount factor
"""
super(QLearningTarget, self).__init__(gamma)
if not isinstance(q_values, collections.Iterable):
q_values = [q_values]
for i, value in enumerate(q_values):
if not isinstance(value, QValue):
raise TypeError('The {}th value is not an instance of `QValue`, instead got: {}'.format(i, type(value)))
self._q_values = q_values
def compute(self, batch):
r"""
Compute the value target :math:`(r + \gamma (1-d) \min_i \max_{a'} Q_{\phi_i}(s',a'))`.
Args:
batch (Batch): batch containing the transitions.
"""
q_max = [torch.max(value(batch['states']), dim=1, keepdim=True)[0] for value in self._q_values]
value = torch.min(torch.cat(q_max, dim=1), dim=1)[0]
batch[self] = batch['rewards'] + self.gamma * (1 - batch['masks']) * value
return batch
class TDValueReturn(TDReturn):
r"""TD State Value Return
Compute the state value one-step temporal difference TD(0), given by:
.. math:: \delta_t^{V} = (r + \gamma (1-d) V_{\phi_{target}}(s')) - V_{\phi}(s)
where :math:`V_{\phi_{target}}` is the target value function.
"""
def __init__(self, value, target_value=None, gamma=1.):
"""
Initialize the TD state value return.
Args:
value (ValueApproximator): state value function.
target_value (ValueApproximator): target state value function.
gamma (float): discount factor
"""
super(TDValueReturn, self).__init__(gamma)
self.value = value
if target_value is None:
self.target_value = self.value
def evaluate(self, batch):
"""Evaluate the TD return on the given batch.
Args:
batch (Batch): batch containing transitions.
Returns:
Batch: batch
"""
target = batch['rewards'] + self.gamma * (1 - batch['masks']) * self.target_value(batch['states'])
batch[self] = target - self.value(batch['states'])
return batch
class TDQValueReturn(TDReturn):
r"""TD Action Value Return
Compute the action value one-step temporal difference TD(0), given by:
.. math:: (r + \gamma (1-d) Q_{\phi_{target}}(s',a')) - Q_{\phi}(s,a)
where :math:`a'` is the action performed by the policy given the state :math:`s'`.
This is also known as the Sarsa (an on-policy TD control) algorithm.
"""
def __init__(self, q_value, policy, target_qvalue=None, gamma=1.):
"""
Initialize the TD state-action value return.
Args:
q_value (QValueApproximator): Q-value function.
policy (Policy): policy to compute the action a'.
target_qvalue (QValueApproximator, None): target Q-value function. If None, it will use the given
:attr:`q_value`.
gamma (float): discount factor
"""
super(TDQValueReturn, self).__init__(gamma)
self.q_value = q_value
self.policy = policy
if target_qvalue is None:
self.target_qvalue = self.q_value
def evaluate(self, batch):
"""Evaluate the TD return on the given batch.
Args:
batch (Batch): batch containing transitions.
Returns:
Batch: batch
"""
action = self.policy.predict(batch['states'])
target = batch['rewards'] + self.gamma * (1 - batch['masks']) * self.target_qvalue(action)
batch[self] = target - self.q_value(batch['states'])
return batch
class TDQLearningReturn(TDReturn):
r"""TD Q-Learning Value Return
Compute the one-step Q-Learning, given by:
.. math:: (r + \gamma (1-d) \max_{a'} Q_{\phi_{target}}(s',a')) - Q_{\phi}(s,a)
where if the actions :math:`a` are discrete, then :math:`a'` is selected such that it maximizes the Q-value, while
if :math:`a` are continuous, :math:`a'`, with the assumption that the policy is fully differentiable, is selected
such that it maximizes locally the Q-value. That is, in the latter case, the action :math:`a'` is first computed
using the policy :math:`\pi(a'|s')`, then by taking the gradient of the Q-value with respect to the action, we
increment the initial action by :math:`a' = a' + \alpha \grad_{a} Q_{\phi_{target}}(s',a)`. This increment step
can be computed for few iterations such that it locally maximizes the :math:` Q_{\phi_{target}}`.
This is known as the Q-Learning (an off-policy TD control) algorithm.
"""
def __init__(self, q_value, target_qvalue=None, gamma=1.):
"""
Initialize the TD Q-Learning value return.
Args:
q_value (QValueApproximator): Q-value function.
target_qvalue (QValueApproximator): target Q-value function. If None, it will use the given
:attr:`q_value`.
gamma (float): discount factor
"""
super(TDQLearningReturn, self).__init__(gamma)
self.q_value = q_value
if target_qvalue is None:
self.target_qvalue = self.q_value
def evaluate(self, batch):
"""Evaluate the TD return on the given batch.
Args:
batch (Batch): batch containing transitions.
Returns:
Batch: batch
"""
q_max = torch.max(self.target_qvalue(batch['states']), dim=1, keepdim=True)[0]
target = batch['rewards'] + self.gamma * (1 - batch['masks']) * q_max
batch[self] = target - self.q_value(batch['states'])
return batch
pyrobolearn/returns/returns.py
+186
View File
@@ -0,0 +1,186 @@
#!/usr/bin/env python
"""Computes the various returns evaluated on batches of transitions (used in RL).
The targets that are evaluated are placed inside the given batch, which can then be accessed by other classes.
Dependencies:
- `pyrobolearn.storages`
- `pyrobolearn.values`
"""
import torch
from pyrobolearn.storages import Batch
from pyrobolearn.values import Value, QValue
from pyrobolearn.policies import Policy
from pyrobolearn.returns.estimator import BaseReturn
__author__ = "Brian Delhaisse"
__copyright__ = "Copyright 2018, PyRoboLearn"
__credits__ = ["Brian Delhaisse"]
__license__ = "MIT"
__version__ = "1.0.0"
__maintainer__ = "Brian Delhaisse"
__email__ = "briandelhaisse@gmail.com"
__status__ = "Development"
class Return(BaseReturn):
r"""Return
Compared to the estimator that used the whole trajectory, it only uses transition tuples
:math:`(s_t, a_t, s_{t+1}, r_t, d)`, where is :math:`s_t` is the state at time :math:`t`, :math:`a_t` is the action
outputted by the policy in response to the state :math:`s_t`, :math:`s_{t+1}` is the next state returned by the
environment due to the policy's action :math:`a_t` and the current state :math:`s_t`, :math:`r_t` is the reward
signal returned by the environment, and :math:`d` is a boolean value that specifies if the task is over or not
(i.e. if it has failed or succeeded).
That is, returns are estimated on the Monte-Carlo trajectories, while returns are estimated on temporal
difference errors [1].
References:
[1] "Reinforcement Learning: an Introduction" (chap 8.13), Sutton and Barto, 2018
"""
def __init__(self, gamma=1.):
"""
Initialize the return / estimator.
Args:
gamma (float): discount factor
"""
super(Return, self).__init__(gamma)
class TDReturn(Return):
r"""TD Return
Return based on the one-step temporal difference TD(0).
"""
pass
class TDValueReturn(TDReturn):
r"""TD State Value Return
Compute the state value one-step temporal difference TD(0), given by:
.. math:: \delta_t^{V} = (r + \gamma (1-d) V_{\phi_{target}}(s')) - V_{\phi}(s)
where :math:`V_{\phi_{target}}` is the target value function.
"""
def __init__(self, value, target_value=None, gamma=1.):
"""
Initialize the TD state value return.
Args:
value (ValueApproximator): state value function.
target_value (ValueApproximator): target state value function.
gamma (float): discount factor
"""
super(TDValueReturn, self).__init__(gamma)
self.value = value
if target_value is None:
self.target_value = self.value
def evaluate(self, batch):
"""Evaluate the TD return on the given batch.
Args:
batch (Batch): batch containing transitions.
Returns:
Batch: batch
"""
target = batch['rewards'] + self.gamma * (1 - batch['masks']) * self.target_value(batch['states'])
batch[self] = target - self.value(batch['states'])
return batch
class TDQValueReturn(TDReturn):
r"""TD Action Value Return
Compute the action value one-step temporal difference TD(0), given by:
.. math:: (r + \gamma (1-d) Q_{\phi_{target}}(s',a')) - Q_{\phi}(s,a)
where :math:`a'` is the action performed by the policy given the state :math:`s'`.
This is also known as the Sarsa (an on-policy TD control) algorithm.
"""
def __init__(self, q_value, policy, target_qvalue=None, gamma=1.):
"""
Initialize the TD state-action value return.
Args:
q_value (QValueApproximator): Q-value function.
policy (Policy): policy to compute the action a'.
target_qvalue (QValueApproximator, None): target Q-value function. If None, it will use the given
:attr:`q_value`.
gamma (float): discount factor
"""
super(TDQValueReturn, self).__init__(gamma)
self.q_value = q_value
self.policy = policy
if target_qvalue is None:
self.target_qvalue = self.q_value
def evaluate(self, batch):
"""Evaluate the TD return on the given batch.
Args:
batch (Batch): batch containing transitions.
Returns:
Batch: batch
"""
action = self.policy.predict(batch['next_states'])
target = batch['rewards'] + self.gamma * (1 - batch['masks']) * self.target_qvalue(action)
batch[self] = target - self.q_value(batch['states'])
return batch
class TDQLearningReturn(TDReturn):
r"""TD Q-Learning Value Return
Compute the one-step Q-Learning, given by:
.. math:: (r + \gamma (1-d) \max_{a'} Q_{\phi_{target}}(s',a')) - Q_{\phi}(s,a)
where if the actions :math:`a` are discrete, then :math:`a'` is selected such that it maximizes the Q-value, while
if :math:`a` are continuous, :math:`a'`, with the assumption that the policy is fully differentiable, is selected
such that it maximizes locally the Q-value. That is, in the latter case, the action :math:`a'` is first computed
using the policy :math:`\pi(a'|s')`, then by taking the gradient of the Q-value with respect to the action, we
increment the initial action by :math:`a' = a' + \alpha \grad_{a} Q_{\phi_{target}}(s',a)`. This increment step
can be computed for few iterations such that it locally maximizes the :math:` Q_{\phi_{target}}`.
This is known as the Q-Learning (an off-policy TD control) algorithm.
"""
def __init__(self, q_value, target_qvalue=None, gamma=1.):
"""
Initialize the TD Q-Learning value return.
Args:
q_value (QValueApproximator): Q-value function.
target_qvalue (QValueApproximator): target Q-value function. If None, it will use the given
:attr:`q_value`.
gamma (float): discount factor
"""
super(TDQLearningReturn, self).__init__(gamma)
self.q_value = q_value
if target_qvalue is None:
self.target_qvalue = self.q_value
def evaluate(self, batch):
"""Evaluate the TD return on the given batch.
Args:
batch (Batch): batch containing transitions.
Returns:
Batch: batch
"""
q_max = torch.max(self.target_qvalue(batch['states']), dim=1, keepdim=True)[0]
target = batch['rewards'] + self.gamma * (1 - batch['masks']) * q_max
batch[self] = target - self.q_value(batch['states'])
return batch
pyrobolearn/returns/targets.py
+409
View File
@@ -0,0 +1,409 @@
#!/usr/bin/env python
"""Computes the various targets based on value function on batches of trajectories/transitions (used in RL).
The targets that are evaluated are placed inside the given batch, which can then be accessed by other classes.
Dependencies:
- `pyrobolearn.storages`
- `pyrobolearn.values`
"""
from abc import ABCMeta
import collections
import torch
from pyrobolearn.storages import Batch
from pyrobolearn.values import Value, QValue
from pyrobolearn.policies import Policy
from pyrobolearn.exploration import Exploration
from pyrobolearn.returns.estimator import BaseReturn
__author__ = "Brian Delhaisse"
__copyright__ = "Copyright 2018, PyRoboLearn"
__credits__ = ["Brian Delhaisse"]
__license__ = "MIT"
__version__ = "1.0.0"
__maintainer__ = "Brian Delhaisse"
__email__ = "briandelhaisse@gmail.com"
__status__ = "Development"
class Target(object):
r"""Target
Targets are evaluated on a batch of transitions or trajectories, and inserted back in the given batch.
"""
def __init__(self, targets=None):
"""
Initialize the target.
Args:
targets (None, list of Target): inner targets.
"""
# check targets
if targets is not None:
if not isinstance(targets, collections.Iterable):
targets = [targets]
for i, target in enumerate(targets):
if not isinstance(target, Target):
raise TypeError("The {}th given target is not an instance of `Target`, instead got: "
"{}".format(i, type(target)))
else:
targets = []
self._targets = targets
def _compute(self, batch):
"""Compute the target on the given batch, and return the result.
Args:
batch (Batch): batch containing the transitions / trajectories.
Returns:
torch.Tensor: result of evaluating the batch.
"""
raise NotImplementedError
def compute(self, batch):
"""Compute/evaluate the target on the given batch and insert the result in the given batch.
Args:
batch (Batch): batch containing the transitions / trajectories.
Returns:
Batch: updated batch.
"""
batch[self] = self._compute(batch)
return batch
def __call__(self, batch):
return self.compute(batch)
class GammaTarget(Target):
__metaclass__ = ABCMeta
def __init__(self, gamma=1.):
"""
Initialize the gamma target.
Args:
gamma (float): discount factor
"""
super(GammaTarget, self).__init__()
self.gamma = gamma
@property
def gamma(self):
"""Return the discount factor"""
return self._gamma
@gamma.setter
def gamma(self, gamma):
"""Set the discount factor"""
if gamma > 1.:
gamma = 1.
elif gamma < 0.:
gamma = 0.
self._gamma = gamma
class VTarget(Target):
r"""Value Target.
Evaluate the value function given by :math:`V_{\phi}(s)` or :math:`V_{\phi}(s')`, depending on the flag.
"""
def __init__(self, value, flag=0):
r"""
Initialize the V-target.
Args:
value (Value): state value function
flag (int): If flag=0, the value function is evaluated on the 'states' :math:`s`, while if flag=1, it is
evaluated on the 'next_states' :math:`s'`.
"""
super(VTarget, self).__init__()
if not isinstance(value, Value):
raise TypeError("Expecting the given value to be an instance of 'Value', instead got: "
"{}".format(type(value)))
self._value = value
self._flag = flag % 2
def _compute(self, batch):
r"""
Compute :math:`V(s)`.
Args:
batch (Batch): batch containing the transitions.
Returns:
torch.Tensor: result
"""
if self._flag == 0:
return self._value(batch['states'])
return self._value(batch['next_states'])
class QTarget(Target):
r"""Value Target.
Evaluate the value function given by :math:`Q_{\phi}(s, a)` or :math:`Q_{\phi}(s', \pi(s'))`, depending if the
policy is given or not.
"""
def __init__(self, q_value, policy=None):
r"""
Initialize the Q-target.
Args:
q_value (Value): state value function
policy (Policy): policy.
"""
super(QTarget, self).__init__()
# check given value function
if not isinstance(q_value, Value):
raise TypeError("Expecting the given value to be an instance of 'Value', instead got: "
"{}".format(type(q_value)))
self._qvalue = q_value
# check given policy
if policy is not None and not isinstance(policy, Policy):
raise TypeError("Expecting the given policy to be None, or an instance of 'Policy', instead got: "
"{}".format(type(policy)))
self._policy = policy
def _compute(self, batch):
r"""
Compute :math:`Q(s, a)` or :math:`Q(s', \pi(s'))`.
Args:
batch (Batch): batch containing the transitions.
Returns:
torch.Tensor: result
"""
if self._policy is None:
return self._qvalue(batch['states'], batch['actions'])
actions = self._policy.predict(batch['states'])
return self._qvalue(batch['next_states'], actions)
class PolicyTarget(Target): # TODO
r"""Policy target.
Compute the log-likelihood on the action :math:`a` returned by the policy :math:`\pi(s)` or :math:`\pi(s')`
depending on the flag. That is, it computes :math:`\log \pi(a|s)` or :math:`\log \pi(`
"""
def __init__(self, policy, flag=0):
"""
Initialize the policy target.
Args:
policy (Exploration): wrapped policy with an exploration strategy.
flag (int): If flag=0, the policy is evaluated on the 'states' :math:`s`, while if flag=1, it is
evaluated on the 'next_states' :math:`s'`.
"""
super(PolicyTarget, self).__init__()
# check given policy
if not isinstance(policy, Exploration):
raise TypeError("Expecting the given policy to be None, or an instance of 'Exploration', instead got: "
"{}".format(type(policy)))
self._policy = policy
self._flag = flag % 2
def _compute(self, batch):
"""
Compute the log-likelihood on the action :math:`a` returned by the policy.
Args:
batch (Batch): batch containing the transitions.
Returns:
torch.Tensor: result
"""
if self._flag == 0:
return self._policy.predict(batch['states'])
return self._policy.predict(batch['next_states'])
class ValueTarget(GammaTarget):
r"""Value target.
Compute the value target given by :math:`(r + \gamma (1-d) \min_i V_{\phi_i}(s'))`, where the index `i` is in
the case there are multiple value function approximators given to this class.
"""
def __init__(self, values, gamma=1.):
"""
Initialize the state value target.
Args:
values (Value, list of Value): state value function(s).
gamma (float): discount factor
"""
super(ValueTarget, self).__init__(gamma)
if not isinstance(values, collections.Iterable):
values = [values]
for i, value in enumerate(values):
if not isinstance(value, Value):
raise TypeError('The {}th value is not an instance of `Value`, instead got: {}'.format(i, type(value)))
self._values = values
def _compute(self, batch):
r"""
Compute the value target :math:`(r + \gamma (1-d) \min_i V_{\phi_i}(s'))`
Args:
batch (Batch): batch containing the transitions.
Returns:
torch.Tensor: result
"""
value = torch.min(torch.cat([value(batch['next_states']) for value in self._values], dim=1), dim=1)[0]
return batch['rewards'] + self.gamma * (1 - batch['masks']) * value
class QValueTarget(GammaTarget):
r"""Q-Value target.
Compute the Q-value target given by :math:`(r + \gamma (1-d) \min_i Q_{\phi_i}(s',a'))`, where the index `i` is in
the case there are multiple Q-value function approximators given to this class.
"""
def __init__(self, q_values, policy, gamma=1.):
"""
Initialize the state value target.
Args:
q_values (QValue, list of QValue): state-action value function(s).
policy (Policy): policy.
gamma (float): discount factor
"""
super(QValueTarget, self).__init__(gamma)
# check Q-values
if not isinstance(q_values, collections.Iterable):
q_values = [q_values]
for i, value in enumerate(q_values):
if not isinstance(value, QValue):
raise TypeError('The {}th value is not an instance of `QValue`, instead got: {}'.format(i, type(value)))
self._q_values = q_values
# check policy
if not isinstance(policy, Policy):
raise TypeError("Expecting the policy to be an instance of `Policy`, instead got: {}".format(type(policy)))
self._policy = policy
def _compute(self, batch):
r"""
Compute the value target :math:`(r + \gamma (1-d) \min_i Q_{\phi_i}(s',a'))`.
Args:
batch (Batch): batch containing the transitions.
Returns:
torch.Tensor: result
"""
actions = self._policy.predict(batch['next_states'])
value = torch.min(torch.cat([value(batch['next_states'], actions) for value in self._q_values], dim=1), dim=1)[
0]
return batch['rewards'] + self.gamma * (1 - batch['masks']) * value
class QLearningTarget(GammaTarget):
r"""Q-Learning target.
Compute the Q-value target given by :math:`(r + \gamma (1-d) \min_i \max_{a'} Q_{\phi_i}(s',a'))`, where the
index `i` is in the case there are multiple Q-value function approximators given to this class.
"""
def __init__(self, q_values, gamma=1.):
"""
Initialize the state value target.
Args:
q_values (QValue, list of QValue): state-action value function(s).
gamma (float): discount factor
"""
super(QLearningTarget, self).__init__(gamma)
if not isinstance(q_values, collections.Iterable):
q_values = [q_values]
for i, value in enumerate(q_values):
if not isinstance(value, QValue):
raise TypeError('The {}th value is not an instance of `QValue`, instead got: {}'.format(i, type(value)))
self._q_values = q_values
def _compute(self, batch):
r"""
Compute the value target :math:`(r + \gamma (1-d) \min_i \max_{a'} Q_{\phi_i}(s',a'))`.
Args:
batch (Batch): batch containing the transitions.
Returns:
torch.Tensor: result
"""
q_max = [torch.max(value(batch['next_states']), dim=1, keepdim=True)[0] for value in self._q_values]
value = torch.min(torch.cat(q_max, dim=1), dim=1)[0]
return batch['rewards'] + self.gamma * (1 - batch['masks']) * value
class EntropyValueTarget(Target):
r"""Entropy regularized value target.
Compute the entropy regularized Q-value target given by
:math:`min_i Q_{\phi_i}(s, \tilde{a}) - \alpha \log \pi_\theta(\tilde{a}|s)`, where
:math:`\tilde{a} \sim \pi_\theta(.|s)`.
"""
def __init__(self, q_values, policy, alpha=0.2):
"""
Initialize the entropy value target.
Args:
q_values (QValue, list of QValue): state-action value funtion(s).
policy (Exploration): wrapped policy with an exploration strategy.
alpha (float): entropy regularization coefficient which controls the tradeoff between exploration and
exploitation. Higher :attr:`alpha` means more exploration, and lower :attr:`alpha` corresponds to more
exploitation.
"""
super(EntropyValueTarget, self).__init__()
# check Q-values
if not isinstance(q_values, collections.Iterable):
q_values = [q_values]
for i, value in enumerate(q_values):
if not isinstance(value, QValue):
raise TypeError('The {}th value is not an instance of `QValue`, instead got: {}'.format(i, type(value)))
# check policy
if not isinstance(policy, Exploration):
raise TypeError(
"Expecting the policy to be an instance of `Exploration`, instead got: {}".format(type(policy)))
self._policy = policy
self._alpha = float(alpha)
def _compute(self, batch):
r"""
Compute the entropy regularized Q-value target
:math:`min_i Q_{\phi_i}(s, \tilde{a}) - \alpha \log \pi_\theta(\tilde{a}|s)`, where
:math:`\tilde{a} \sim \pi_\theta(.|s)`.
Args:
batch (Batch): batch containing the transitions.
Returns:
torch.Tensor: result
"""
actions, distribution = self._policy.predict(batch['states'])
value = torch.min(torch.cat([value(batch['states'], actions) for value in self._q_values], dim=1), dim=1)[0]
return value - self._alpha * distribution.log_prob(actions)