Index
Utility classes and functions for neural networks
functional
¶
ModuleExpectingFlatParameters
¶
A wrapper which brings a functional interface around a torch module.
For obtaining the functional interface, this class internally uses
the functorch library.
Similar to functorch.FunctionalModule, ModuleExpectingFlatParameters
turns a torch.nn.Module instance to a function which expects a new
leftmost argument representing the parameters of the network.
Unlike functorch.FunctionalModule, a ModuleExpectingFlatParameters
instance, as its name suggests, expects the network parameters to be
given as a 1-dimensional (i.e. flattened) tensor.
Also, unlike functorch.FunctionalModule, an instance of
ModuleExpectingFlatParameters is NOT an instance of torch.nn.Module.
PyTorch modules with buffers can be wrapped by this class, but it is assumed that those buffers are constant. If the wrapped module changes the value(s) of its buffer(s) during its forward passes, most probably things will NOT work right.
As an example, let us consider the following linear layer.
The functional counterpart of net can be obtained via:
from evotorch.neuroevolution.net import ModuleExpectingFlatParameters
fnet = ModuleExpectingFlatParameters(net)
Now, fnet is a callable object which expects network parameters
and network inputs. Let us call fnet with randomly generated network
parameters and with a randomly generated input tensor.
param_length = fnet.parameter_length
random_parameters = torch.randn(param_length)
random_input = torch.randn(3)
result = fnet(random_parameters, random_input)
Source code in evotorch/neuroevolution/net/functional.py
class ModuleExpectingFlatParameters:
"""
A wrapper which brings a functional interface around a torch module.
For obtaining the functional interface, this class internally uses
the `functorch` library.
Similar to `functorch.FunctionalModule`, `ModuleExpectingFlatParameters`
turns a `torch.nn.Module` instance to a function which expects a new
leftmost argument representing the parameters of the network.
Unlike `functorch.FunctionalModule`, a `ModuleExpectingFlatParameters`
instance, as its name suggests, expects the network parameters to be
given as a 1-dimensional (i.e. flattened) tensor.
Also, unlike `functorch.FunctionalModule`, an instance of
`ModuleExpectingFlatParameters` is NOT an instance of `torch.nn.Module`.
PyTorch modules with buffers can be wrapped by this class, but it is
assumed that those buffers are constant. If the wrapped module changes
the value(s) of its buffer(s) during its forward passes, most probably
things will NOT work right.
As an example, let us consider the following linear layer.
```python
import torch
from torch import nn
net = nn.Linear(3, 8)
```
The functional counterpart of `net` can be obtained via:
```python
from evotorch.neuroevolution.net import ModuleExpectingFlatParameters
fnet = ModuleExpectingFlatParameters(net)
```
Now, `fnet` is a callable object which expects network parameters
and network inputs. Let us call `fnet` with randomly generated network
parameters and with a randomly generated input tensor.
```python
param_length = fnet.parameter_length
random_parameters = torch.randn(param_length)
random_input = torch.randn(3)
result = fnet(random_parameters, random_input)
```
"""
def __init__(self, module: nn.Module, disable_autograd_tracking: bool = False):
# Declare the variables which will store information regarding the parameters of the module.
self.__param_shapes = []
self.__param_length = 0
self.__param_slices = []
self.__num_params = 0
self.__buffers = []
# Iterate over the parameters of the module and fill the related information.
i = 0
j = 0
for p in module.parameters():
shape = p.shape
self.__param_shapes.append(shape)
length = _shape_length(shape)
self.__param_length += length
j = i + length
self.__param_slices.append(slice(i, j))
i = j
self.__num_params += 1
self.__fmodel, _, self.__buffers = make_functional_with_buffers(
module, disable_autograd_tracking=bool(disable_autograd_tracking)
)
self.__buffers = list(self.__buffers)
def __transfer_buffers(self, x: torch.Tensor):
"""
Transfer the buffer tensors to the device of the given tensor.
Args:
x: The tensor whose device will also store the buffer tensors.
"""
n = len(self.__buffers)
for i in range(n):
self.__buffers[i] = torch.as_tensor(self.__buffers[i], device=x.device)
@property
def buffers(self) -> tuple:
"""Get the stored buffers"""
return self.__buffers
@property
def parameter_length(self) -> int:
return self.__param_length
def __call__(self, parameter_vector: torch.Tensor, x: torch.Tensor, h: Any = None) -> Any:
"""
Call the wrapped module's forward pass procedure.
Args:
parameter_vector: A 1-dimensional tensor which represents the
parameters of the tensor.
x: The inputs.
h: Hidden state(s), in case this is a recurrent network.
Returns:
The result of the forward pass.
"""
if parameter_vector.ndim != 1:
raise ValueError(
f"Expected the parameters as 1 dimensional,"
f" but the received parameter vector has {parameter_vector.ndim} dimensions"
)
if len(parameter_vector) != self.__param_length:
raise ValueError(
f"Expected a parameter vector of length {self.__param_length},"
f" but the received parameter vector's length is {len(parameter_vector)}."
)
state_args = [] if h is None else [h]
params = []
for i in range(self.__num_params):
param_slice = self.__param_slices[i]
param_shape = self.__param_shapes[i]
param = parameter_vector[param_slice].reshape(param_shape)
params.append(param)
# Make sure that the tensors are in the same device with x
self.__transfer_buffers(x)
# Run the functional module and return the results
return self.__fmodel(params, self.__buffers, x, *state_args)
buffers: tuple
property
readonly
¶
Get the stored buffers
__call__(self, parameter_vector, x, h=None)
special
¶
Call the wrapped module's forward pass procedure.
Parameters:
| Name | Type | Description | Default |
|---|---|---|---|
parameter_vector |
Tensor |
A 1-dimensional tensor which represents the parameters of the tensor. |
required |
x |
Tensor |
The inputs. |
required |
h |
Any |
Hidden state(s), in case this is a recurrent network. |
None |
Returns:
| Type | Description |
|---|---|
Any |
The result of the forward pass. |
Source code in evotorch/neuroevolution/net/functional.py
def __call__(self, parameter_vector: torch.Tensor, x: torch.Tensor, h: Any = None) -> Any:
"""
Call the wrapped module's forward pass procedure.
Args:
parameter_vector: A 1-dimensional tensor which represents the
parameters of the tensor.
x: The inputs.
h: Hidden state(s), in case this is a recurrent network.
Returns:
The result of the forward pass.
"""
if parameter_vector.ndim != 1:
raise ValueError(
f"Expected the parameters as 1 dimensional,"
f" but the received parameter vector has {parameter_vector.ndim} dimensions"
)
if len(parameter_vector) != self.__param_length:
raise ValueError(
f"Expected a parameter vector of length {self.__param_length},"
f" but the received parameter vector's length is {len(parameter_vector)}."
)
state_args = [] if h is None else [h]
params = []
for i in range(self.__num_params):
param_slice = self.__param_slices[i]
param_shape = self.__param_shapes[i]
param = parameter_vector[param_slice].reshape(param_shape)
params.append(param)
# Make sure that the tensors are in the same device with x
self.__transfer_buffers(x)
# Run the functional module and return the results
return self.__fmodel(params, self.__buffers, x, *state_args)
make_functional_module(net)
¶
Wrap a torch module so that it has a functional interface.
For obtaining a functional interface, this function internally uses the
functorch library.
Similar to functorch.make_functional(...), this function turns a
torch.nn.Module instance to a function which expects a new leftmost
argument representing the parameters of the network.
Unlike with functorch.make_functional(...), the parameters of the
network are expected in a 1-dimensional (i.e. flattened) tensor.
PyTorch modules with buffers can be wrapped by this class, but it is assumed that those buffers are constant. If the wrapped module changes the value(s) of its buffer(s) during its forward passes, most probably things will NOT work right.
As an example, let us consider the following linear layer.
The functional counterpart of net can be obtained via:
Now, fnet is a callable object which expects network parameters
and network inputs. Let us call fnet with randomly generated network
parameters and with a randomly generated input tensor.
param_length = fnet.parameter_length
random_parameters = torch.randn(param_length)
random_input = torch.randn(3)
result = fnet(random_parameters, random_input)
Parameters:
| Name | Type | Description | Default |
|---|---|---|---|
net |
Module |
The |
required |
Returns:
| Type | Description |
|---|---|
ModuleExpectingFlatParameters |
The functional wrapper, as an instance of
|
Source code in evotorch/neuroevolution/net/functional.py
def make_functional_module(net: nn.Module) -> ModuleExpectingFlatParameters:
"""
Wrap a torch module so that it has a functional interface.
For obtaining a functional interface, this function internally uses the
`functorch` library.
Similar to `functorch.make_functional(...)`, this function turns a
`torch.nn.Module` instance to a function which expects a new leftmost
argument representing the parameters of the network.
Unlike with `functorch.make_functional(...)`, the parameters of the
network are expected in a 1-dimensional (i.e. flattened) tensor.
PyTorch modules with buffers can be wrapped by this class, but it is
assumed that those buffers are constant. If the wrapped module changes
the value(s) of its buffer(s) during its forward passes, most probably
things will NOT work right.
As an example, let us consider the following linear layer.
```python
import torch
from torch import nn
net = nn.Linear(3, 8)
```
The functional counterpart of `net` can be obtained via:
```python
from evotorch.neuroevolution.net import make_functional_module
fnet = make_functional_module(net)
```
Now, `fnet` is a callable object which expects network parameters
and network inputs. Let us call `fnet` with randomly generated network
parameters and with a randomly generated input tensor.
```python
param_length = fnet.parameter_length
random_parameters = torch.randn(param_length)
random_input = torch.randn(3)
result = fnet(random_parameters, random_input)
```
Args:
net: The `torch.nn.Module` instance to be wrapped by a functional
interface.
Returns:
The functional wrapper, as an instance of
`evotorch.neuroevolution.net.ModuleExpectingFlatParameters`.
"""
return ModuleExpectingFlatParameters(net)
layers
¶
Various neural network layer types
Apply (Module)
¶
A torch module for applying an arithmetic operator on an input tensor
Source code in evotorch/neuroevolution/net/layers.py
class Apply(nn.Module):
"""A torch module for applying an arithmetic operator on an input tensor"""
def __init__(self, operator: str, argument: float):
"""`__init__(...)`: Initialize the Apply module.
Args:
operator: Must be '+', '-', '*', '/', or '**'.
Indicates which operation will be done
on the input tensor.
argument: Expected as a float, represents
the right-argument of the operation
(the left-argument being the input
tensor).
"""
nn.Module.__init__(self)
self._operator = str(operator)
assert self._operator in ("+", "-", "*", "/", "**")
self._argument = float(argument)
def forward(self, x):
op = self._operator
arg = self._argument
if op == "+":
return x + arg
elif op == "-":
return x - arg
elif op == "*":
return x * arg
elif op == "/":
return x / arg
elif op == "**":
return x**arg
else:
raise ValueError("Unknown operator:" + repr(op))
def extra_repr(self):
return "operator={}, argument={}".format(repr(self._operator), self._argument)
__init__(self, operator, argument)
special
¶
__init__(...): Initialize the Apply module.
Parameters:
| Name | Type | Description | Default |
|---|---|---|---|
operator |
str |
Must be '+', '-', '', '/', or '*'. Indicates which operation will be done on the input tensor. |
required |
argument |
float |
Expected as a float, represents the right-argument of the operation (the left-argument being the input tensor). |
required |
Source code in evotorch/neuroevolution/net/layers.py
def __init__(self, operator: str, argument: float):
"""`__init__(...)`: Initialize the Apply module.
Args:
operator: Must be '+', '-', '*', '/', or '**'.
Indicates which operation will be done
on the input tensor.
argument: Expected as a float, represents
the right-argument of the operation
(the left-argument being the input
tensor).
"""
nn.Module.__init__(self)
self._operator = str(operator)
assert self._operator in ("+", "-", "*", "/", "**")
self._argument = float(argument)
extra_repr(self)
¶
Set the extra representation of the module
To print customized extra information, you should re-implement this method in your own modules. Both single-line and multi-line strings are acceptable.
forward(self, x)
¶
Defines the computation performed at every call.
Should be overridden by all subclasses.
.. note::
Although the recipe for forward pass needs to be defined within
this function, one should call the :class:Module instance afterwards
instead of this since the former takes care of running the
registered hooks while the latter silently ignores them.
Source code in evotorch/neuroevolution/net/layers.py
Bin (Module)
¶
A small torch module for binning the values of tensors.
In more details, considering a lower bound value lb, an upper bound value ub, and an input tensor x, each value within x closer to lb will be converted to lb and each value within x closer to ub will be converted to ub.
Source code in evotorch/neuroevolution/net/layers.py
class Bin(nn.Module):
"""A small torch module for binning the values of tensors.
In more details, considering a lower bound value lb,
an upper bound value ub, and an input tensor x,
each value within x closer to lb will be converted to lb
and each value within x closer to ub will be converted to ub.
"""
def __init__(self, lb: float, ub: float):
"""`__init__(...)`: Initialize the Clip operator.
Args:
lb: Lower bound
ub: Upper bound
"""
nn.Module.__init__(self)
self._lb = float(lb)
self._ub = float(ub)
self._interval_size = self._ub - self._lb
self._shrink_amount = self._interval_size / 2.0
self._shift_amount = (self._ub + self._lb) / 2.0
def forward(self, x: torch.Tensor):
x = x - self._shift_amount
x = x / self._shrink_amount
x = torch.sign(x)
x = x * self._shrink_amount
x = x + self._shift_amount
return x
def extra_repr(self):
return "lb={}, ub={}".format(self._lb, self._ub)
__init__(self, lb, ub)
special
¶
__init__(...): Initialize the Clip operator.
Parameters:
| Name | Type | Description | Default |
|---|---|---|---|
lb |
float |
Lower bound |
required |
ub |
float |
Upper bound |
required |
Source code in evotorch/neuroevolution/net/layers.py
def __init__(self, lb: float, ub: float):
"""`__init__(...)`: Initialize the Clip operator.
Args:
lb: Lower bound
ub: Upper bound
"""
nn.Module.__init__(self)
self._lb = float(lb)
self._ub = float(ub)
self._interval_size = self._ub - self._lb
self._shrink_amount = self._interval_size / 2.0
self._shift_amount = (self._ub + self._lb) / 2.0
extra_repr(self)
¶
Set the extra representation of the module
To print customized extra information, you should re-implement this method in your own modules. Both single-line and multi-line strings are acceptable.
forward(self, x)
¶
Defines the computation performed at every call.
Should be overridden by all subclasses.
.. note::
Although the recipe for forward pass needs to be defined within
this function, one should call the :class:Module instance afterwards
instead of this since the former takes care of running the
registered hooks while the latter silently ignores them.
Clip (Module)
¶
A small torch module for clipping the values of tensors
Source code in evotorch/neuroevolution/net/layers.py
class Clip(nn.Module):
"""A small torch module for clipping the values of tensors"""
def __init__(self, lb: float, ub: float):
"""`__init__(...)`: Initialize the Clip operator.
Args:
lb: Lower bound. Values less than this will be clipped.
ub: Upper bound. Values greater than this will be clipped.
"""
nn.Module.__init__(self)
self._lb = float(lb)
self._ub = float(ub)
def forward(self, x: torch.Tensor):
return x.clamp(self._lb, self._ub)
def extra_repr(self):
return "lb={}, ub={}".format(self._lb, self._ub)
__init__(self, lb, ub)
special
¶
__init__(...): Initialize the Clip operator.
Parameters:
| Name | Type | Description | Default |
|---|---|---|---|
lb |
float |
Lower bound. Values less than this will be clipped. |
required |
ub |
float |
Upper bound. Values greater than this will be clipped. |
required |
Source code in evotorch/neuroevolution/net/layers.py
extra_repr(self)
¶
Set the extra representation of the module
To print customized extra information, you should re-implement this method in your own modules. Both single-line and multi-line strings are acceptable.
forward(self, x)
¶
Defines the computation performed at every call.
Should be overridden by all subclasses.
.. note::
Although the recipe for forward pass needs to be defined within
this function, one should call the :class:Module instance afterwards
instead of this since the former takes care of running the
registered hooks while the latter silently ignores them.
FeedForwardNet (Module)
¶
Representation of a feed forward neural network as a torch Module.
An example initialization of a FeedForwardNet is as follows:
net = drt.FeedForwardNet(4, [(8, 'tanh'), (6, 'tanh')])
which means that we would like to have a network which expects an input vector of length 4 and passes its input through 2 tanh-activated hidden layers (with neurons count 8 and 6, respectively). The output of the last hidden layer (of length 6) is the final output vector.
The string representation of the module obtained via the example above is:
FeedForwardNet(
(layer_0): Linear(in_features=4, out_features=8, bias=True)
(actfunc_0): Tanh()
(layer_1): Linear(in_features=8, out_features=6, bias=True)
(actfunc_1): Tanh()
)
Source code in evotorch/neuroevolution/net/layers.py
class FeedForwardNet(nn.Module):
"""
Representation of a feed forward neural network as a torch Module.
An example initialization of a FeedForwardNet is as follows:
net = drt.FeedForwardNet(4, [(8, 'tanh'), (6, 'tanh')])
which means that we would like to have a network which expects an input
vector of length 4 and passes its input through 2 tanh-activated hidden
layers (with neurons count 8 and 6, respectively).
The output of the last hidden layer (of length 6) is the final
output vector.
The string representation of the module obtained via the example above
is:
FeedForwardNet(
(layer_0): Linear(in_features=4, out_features=8, bias=True)
(actfunc_0): Tanh()
(layer_1): Linear(in_features=8, out_features=6, bias=True)
(actfunc_1): Tanh()
)
"""
LengthActTuple = Tuple[int, Union[str, Callable]]
LengthActBiasTuple = Tuple[int, Union[str, Callable], Union[bool]]
def __init__(self, input_size: int, layers: List[Union[LengthActTuple, LengthActBiasTuple]]):
"""`__init__(...)`: Initialize the FeedForward network.
Args:
input_size: Input size of the network, expected as an int.
layers: Expected as a list of tuples,
where each tuple is either of the form
`(layer_size, activation_function)`
or of the form
`(layer_size, activation_function, bias)`
in which
(i) `layer_size` is an int, specifying the number of neurons;
(ii) `activation_function` is None, or a callable object,
or a string containing the name of the activation function
('relu', 'selu', 'elu', 'tanh', 'hardtanh', or 'sigmoid');
(iii) `bias` is a boolean, specifying whether the layer
is to have a bias or not.
When omitted, bias is set to True.
"""
nn.Module.__init__(self)
for i, layer in enumerate(layers):
if len(layer) == 2:
size, actfunc = layer
bias = True
elif len(layer) == 3:
size, actfunc, bias = layer
else:
assert False, "A layer tuple of invalid size is encountered"
setattr(self, "layer_" + str(i), nn.Linear(input_size, size, bias=bias))
if isinstance(actfunc, str):
if actfunc == "relu":
actfunc = nn.ReLU()
elif actfunc == "selu":
actfunc = nn.SELU()
elif actfunc == "elu":
actfunc = nn.ELU()
elif actfunc == "tanh":
actfunc = nn.Tanh()
elif actfunc == "hardtanh":
actfunc = nn.Hardtanh()
elif actfunc == "sigmoid":
actfunc = nn.Sigmoid()
elif actfunc == "round":
actfunc = Round()
else:
raise ValueError("Unknown activation function: " + repr(actfunc))
setattr(self, "actfunc_" + str(i), actfunc)
input_size = size
def forward(self, x):
i = 0
while hasattr(self, "layer_" + str(i)):
x = getattr(self, "layer_" + str(i))(x)
f = getattr(self, "actfunc_" + str(i))
if f is not None:
x = f(x)
i += 1
return x
__init__(self, input_size, layers)
special
¶
__init__(...): Initialize the FeedForward network.
Parameters:
| Name | Type | Description | Default |
|---|---|---|---|
input_size |
int |
Input size of the network, expected as an int. |
required |
layers |
List[Union[Tuple[int, Union[str, Callable]], Tuple[int, Union[str, Callable], bool]]] |
Expected as a list of tuples,
where each tuple is either of the form
|
required |
Source code in evotorch/neuroevolution/net/layers.py
def __init__(self, input_size: int, layers: List[Union[LengthActTuple, LengthActBiasTuple]]):
"""`__init__(...)`: Initialize the FeedForward network.
Args:
input_size: Input size of the network, expected as an int.
layers: Expected as a list of tuples,
where each tuple is either of the form
`(layer_size, activation_function)`
or of the form
`(layer_size, activation_function, bias)`
in which
(i) `layer_size` is an int, specifying the number of neurons;
(ii) `activation_function` is None, or a callable object,
or a string containing the name of the activation function
('relu', 'selu', 'elu', 'tanh', 'hardtanh', or 'sigmoid');
(iii) `bias` is a boolean, specifying whether the layer
is to have a bias or not.
When omitted, bias is set to True.
"""
nn.Module.__init__(self)
for i, layer in enumerate(layers):
if len(layer) == 2:
size, actfunc = layer
bias = True
elif len(layer) == 3:
size, actfunc, bias = layer
else:
assert False, "A layer tuple of invalid size is encountered"
setattr(self, "layer_" + str(i), nn.Linear(input_size, size, bias=bias))
if isinstance(actfunc, str):
if actfunc == "relu":
actfunc = nn.ReLU()
elif actfunc == "selu":
actfunc = nn.SELU()
elif actfunc == "elu":
actfunc = nn.ELU()
elif actfunc == "tanh":
actfunc = nn.Tanh()
elif actfunc == "hardtanh":
actfunc = nn.Hardtanh()
elif actfunc == "sigmoid":
actfunc = nn.Sigmoid()
elif actfunc == "round":
actfunc = Round()
else:
raise ValueError("Unknown activation function: " + repr(actfunc))
setattr(self, "actfunc_" + str(i), actfunc)
input_size = size
forward(self, x)
¶
Defines the computation performed at every call.
Should be overridden by all subclasses.
.. note::
Although the recipe for forward pass needs to be defined within
this function, one should call the :class:Module instance afterwards
instead of this since the former takes care of running the
registered hooks while the latter silently ignores them.
LSTM (Module)
¶
Source code in evotorch/neuroevolution/net/layers.py
class LSTM(nn.Module):
def __init__(
self,
input_size: int,
hidden_size: int,
*,
dtype: torch.dtype = torch.float32,
device: Union[str, torch.device] = "cpu",
):
super().__init__()
input_size = int(input_size)
hidden_size = int(hidden_size)
self.input_size = input_size
self.hidden_size = hidden_size
def input_weight():
return nn.Parameter(torch.randn(self.hidden_size, self.input_size, dtype=dtype, device=device))
def weight():
return nn.Parameter(torch.randn(self.hidden_size, self.hidden_size, dtype=dtype, device=device))
def bias():
return nn.Parameter(torch.zeros(self.hidden_size, dtype=dtype, device=device))
self.W_ii = input_weight()
self.W_if = input_weight()
self.W_ig = input_weight()
self.W_io = input_weight()
self.W_hi = weight()
self.W_hf = weight()
self.W_hg = weight()
self.W_ho = weight()
self.b_ii = bias()
self.b_if = bias()
self.b_ig = bias()
self.b_io = bias()
self.b_hi = bias()
self.b_hf = bias()
self.b_hg = bias()
self.b_ho = bias()
def forward(self, x: torch.Tensor, hidden=None) -> tuple:
sigm = torch.sigmoid
tanh = torch.tanh
if hidden is None:
h_prev = torch.zeros(self.hidden_size, dtype=x.dtype, device=x.device)
c_prev = torch.zeros(self.hidden_size, dtype=x.dtype, device=x.device)
else:
h_prev, c_prev = hidden
i_t = sigm(self.W_ii @ x + self.b_ii + self.W_hi @ h_prev + self.b_hi)
f_t = sigm(self.W_if @ x + self.b_if + self.W_hf @ h_prev + self.b_hf)
g_t = tanh(self.W_ig @ x + self.b_ig + self.W_hg @ h_prev + self.b_hg)
o_t = sigm(self.W_io @ x + self.b_io + self.W_ho @ h_prev + self.b_ho)
c_t = f_t * c_prev + i_t * g_t
h_t = o_t * tanh(c_t)
return h_t, (h_t, c_t)
def __repr__(self) -> str:
clsname = type(self).__name__
return f"{clsname}(input_size={self.input_size}, hidden_size={self.hidden_size})"
forward(self, x, hidden=None)
¶
Defines the computation performed at every call.
Should be overridden by all subclasses.
.. note::
Although the recipe for forward pass needs to be defined within
this function, one should call the :class:Module instance afterwards
instead of this since the former takes care of running the
registered hooks while the latter silently ignores them.
Source code in evotorch/neuroevolution/net/layers.py
def forward(self, x: torch.Tensor, hidden=None) -> tuple:
sigm = torch.sigmoid
tanh = torch.tanh
if hidden is None:
h_prev = torch.zeros(self.hidden_size, dtype=x.dtype, device=x.device)
c_prev = torch.zeros(self.hidden_size, dtype=x.dtype, device=x.device)
else:
h_prev, c_prev = hidden
i_t = sigm(self.W_ii @ x + self.b_ii + self.W_hi @ h_prev + self.b_hi)
f_t = sigm(self.W_if @ x + self.b_if + self.W_hf @ h_prev + self.b_hf)
g_t = tanh(self.W_ig @ x + self.b_ig + self.W_hg @ h_prev + self.b_hg)
o_t = sigm(self.W_io @ x + self.b_io + self.W_ho @ h_prev + self.b_ho)
c_t = f_t * c_prev + i_t * g_t
h_t = o_t * tanh(c_t)
return h_t, (h_t, c_t)
LSTMNet (Module)
¶
Source code in evotorch/neuroevolution/net/layers.py
class LSTM(nn.Module):
def __init__(
self,
input_size: int,
hidden_size: int,
*,
dtype: torch.dtype = torch.float32,
device: Union[str, torch.device] = "cpu",
):
super().__init__()
input_size = int(input_size)
hidden_size = int(hidden_size)
self.input_size = input_size
self.hidden_size = hidden_size
def input_weight():
return nn.Parameter(torch.randn(self.hidden_size, self.input_size, dtype=dtype, device=device))
def weight():
return nn.Parameter(torch.randn(self.hidden_size, self.hidden_size, dtype=dtype, device=device))
def bias():
return nn.Parameter(torch.zeros(self.hidden_size, dtype=dtype, device=device))
self.W_ii = input_weight()
self.W_if = input_weight()
self.W_ig = input_weight()
self.W_io = input_weight()
self.W_hi = weight()
self.W_hf = weight()
self.W_hg = weight()
self.W_ho = weight()
self.b_ii = bias()
self.b_if = bias()
self.b_ig = bias()
self.b_io = bias()
self.b_hi = bias()
self.b_hf = bias()
self.b_hg = bias()
self.b_ho = bias()
def forward(self, x: torch.Tensor, hidden=None) -> tuple:
sigm = torch.sigmoid
tanh = torch.tanh
if hidden is None:
h_prev = torch.zeros(self.hidden_size, dtype=x.dtype, device=x.device)
c_prev = torch.zeros(self.hidden_size, dtype=x.dtype, device=x.device)
else:
h_prev, c_prev = hidden
i_t = sigm(self.W_ii @ x + self.b_ii + self.W_hi @ h_prev + self.b_hi)
f_t = sigm(self.W_if @ x + self.b_if + self.W_hf @ h_prev + self.b_hf)
g_t = tanh(self.W_ig @ x + self.b_ig + self.W_hg @ h_prev + self.b_hg)
o_t = sigm(self.W_io @ x + self.b_io + self.W_ho @ h_prev + self.b_ho)
c_t = f_t * c_prev + i_t * g_t
h_t = o_t * tanh(c_t)
return h_t, (h_t, c_t)
def __repr__(self) -> str:
clsname = type(self).__name__
return f"{clsname}(input_size={self.input_size}, hidden_size={self.hidden_size})"
forward(self, x, hidden=None)
¶
Defines the computation performed at every call.
Should be overridden by all subclasses.
.. note::
Although the recipe for forward pass needs to be defined within
this function, one should call the :class:Module instance afterwards
instead of this since the former takes care of running the
registered hooks while the latter silently ignores them.
Source code in evotorch/neuroevolution/net/layers.py
def forward(self, x: torch.Tensor, hidden=None) -> tuple:
sigm = torch.sigmoid
tanh = torch.tanh
if hidden is None:
h_prev = torch.zeros(self.hidden_size, dtype=x.dtype, device=x.device)
c_prev = torch.zeros(self.hidden_size, dtype=x.dtype, device=x.device)
else:
h_prev, c_prev = hidden
i_t = sigm(self.W_ii @ x + self.b_ii + self.W_hi @ h_prev + self.b_hi)
f_t = sigm(self.W_if @ x + self.b_if + self.W_hf @ h_prev + self.b_hf)
g_t = tanh(self.W_ig @ x + self.b_ig + self.W_hg @ h_prev + self.b_hg)
o_t = sigm(self.W_io @ x + self.b_io + self.W_ho @ h_prev + self.b_ho)
c_t = f_t * c_prev + i_t * g_t
h_t = o_t * tanh(c_t)
return h_t, (h_t, c_t)
LocomotorNet (Module)
¶
This is a control network which consists of two components: one linear, and one non-linear. The non-linear component is an input-independent set of sinusoidals waves whose amplitudes, frequencies and phases are trainable. Upon execution of a forward pass, the output of the non-linear component is the sum of all these sinusoidal waves. The linear component is a linear layer (optionally with bias) whose weights (and biases) are trainable. The final output of the LocomotorNet at the end of a forward pass is the sum of the linear and the non-linear components.
Note that this is a stateful network, where the only state
is the timestep t, which starts from 0 and gets incremented by 1
at the end of each forward pass. The reset() method resets
t back to 0.
Reference
Mario Srouji, Jian Zhang, Ruslan Salakhutdinov (2018). Structured Control Nets for Deep Reinforcement Learning.
Source code in evotorch/neuroevolution/net/layers.py
class LocomotorNet(nn.Module):
"""LocomotorNet: A locomotion-specific structured control net.
This is a control network which consists of two components:
one linear, and one non-linear. The non-linear component
is an input-independent set of sinusoidals waves whose
amplitudes, frequencies and phases are trainable.
Upon execution of a forward pass, the output of the non-linear
component is the sum of all these sinusoidal waves.
The linear component is a linear layer (optionally with bias)
whose weights (and biases) are trainable.
The final output of the LocomotorNet at the end of a forward pass
is the sum of the linear and the non-linear components.
Note that this is a stateful network, where the only state
is the timestep t, which starts from 0 and gets incremented by 1
at the end of each forward pass. The `reset()` method resets
t back to 0.
Reference:
Mario Srouji, Jian Zhang, Ruslan Salakhutdinov (2018).
Structured Control Nets for Deep Reinforcement Learning.
"""
def __init__(self, *, in_features: int, out_features: int, bias: bool = True, num_sinusoids=16):
"""`__init__(...)`: Initialize the LocomotorNet.
Args:
in_features: Length of the input vector
out_features: Length of the output vector
bias: Whether or not the linear component is to have a bias
num_sinusoids: Number of sinusoidal waves
"""
nn.Module.__init__(self)
self._in_features = in_features
self._out_features = out_features
self._bias = bias
self._num_sinusoids = num_sinusoids
self._linear_component = nn.Linear(
in_features=self._in_features, out_features=self._out_features, bias=self._bias
)
self._amplitudes = nn.ParameterList()
self._frequencies = nn.ParameterList()
self._phases = nn.ParameterList()
for _ in range(self._num_sinusoids):
for paramlist in (self._amplitudes, self._frequencies, self._phases):
paramlist.append(nn.Parameter(torch.randn(self._out_features, dtype=torch.float32)))
self.reset()
def reset(self):
"""Set the timestep t to 0"""
self._t = 0
@property
def t(self) -> int:
"""The current timestep t"""
return self._t
@property
def in_features(self) -> int:
"""Get the length of the input vector"""
return self._in_features
@property
def out_features(self) -> int:
"""Get the length of the output vector"""
return self._out_features
@property
def num_sinusoids(self) -> int:
"""Get the number of sinusoidal waves of the non-linear component"""
return self._num_sinusoids
@property
def bias(self) -> bool:
"""Get whether or not the linear component has bias"""
return self._bias
def forward(self, x: torch.Tensor) -> torch.Tensor:
"""Execute a forward pass"""
u_linear = self._linear_component(x)
t = self._t
u_nonlinear = torch.zeros(self._out_features)
for i in range(self._num_sinusoids):
A = self._amplitudes[i]
w = self._frequencies[i]
phi = self._phases[i]
u_nonlinear = u_nonlinear + (A * torch.sin(w * t + phi))
self._t += 1
return u_linear + u_nonlinear
bias: bool
property
readonly
¶
Get whether or not the linear component has bias
in_features: int
property
readonly
¶
Get the length of the input vector
num_sinusoids: int
property
readonly
¶
Get the number of sinusoidal waves of the non-linear component
out_features: int
property
readonly
¶
Get the length of the output vector
t: int
property
readonly
¶
The current timestep t
__init__(self, *, in_features, out_features, bias=True, num_sinusoids=16)
special
¶
__init__(...): Initialize the LocomotorNet.
Parameters:
| Name | Type | Description | Default |
|---|---|---|---|
in_features |
int |
Length of the input vector |
required |
out_features |
int |
Length of the output vector |
required |
bias |
bool |
Whether or not the linear component is to have a bias |
True |
num_sinusoids |
Number of sinusoidal waves |
16 |
Source code in evotorch/neuroevolution/net/layers.py
def __init__(self, *, in_features: int, out_features: int, bias: bool = True, num_sinusoids=16):
"""`__init__(...)`: Initialize the LocomotorNet.
Args:
in_features: Length of the input vector
out_features: Length of the output vector
bias: Whether or not the linear component is to have a bias
num_sinusoids: Number of sinusoidal waves
"""
nn.Module.__init__(self)
self._in_features = in_features
self._out_features = out_features
self._bias = bias
self._num_sinusoids = num_sinusoids
self._linear_component = nn.Linear(
in_features=self._in_features, out_features=self._out_features, bias=self._bias
)
self._amplitudes = nn.ParameterList()
self._frequencies = nn.ParameterList()
self._phases = nn.ParameterList()
for _ in range(self._num_sinusoids):
for paramlist in (self._amplitudes, self._frequencies, self._phases):
paramlist.append(nn.Parameter(torch.randn(self._out_features, dtype=torch.float32)))
self.reset()
forward(self, x)
¶
Execute a forward pass
Source code in evotorch/neuroevolution/net/layers.py
def forward(self, x: torch.Tensor) -> torch.Tensor:
"""Execute a forward pass"""
u_linear = self._linear_component(x)
t = self._t
u_nonlinear = torch.zeros(self._out_features)
for i in range(self._num_sinusoids):
A = self._amplitudes[i]
w = self._frequencies[i]
phi = self._phases[i]
u_nonlinear = u_nonlinear + (A * torch.sin(w * t + phi))
self._t += 1
return u_linear + u_nonlinear
reset(self)
¶
RNN (Module)
¶
Source code in evotorch/neuroevolution/net/layers.py
class RNN(nn.Module):
def __init__(
self,
input_size: int,
hidden_size: int,
nonlinearity: str = "tanh",
*,
dtype: torch.dtype = torch.float32,
device: Union[str, torch.device] = "cpu",
):
super().__init__()
input_size = int(input_size)
hidden_size = int(hidden_size)
nonlinearity = str(nonlinearity)
self.W1 = nn.Parameter(torch.randn(hidden_size, input_size, dtype=dtype, device=device))
self.W2 = nn.Parameter(torch.randn(hidden_size, hidden_size, dtype=dtype, device=device))
self.b1 = nn.Parameter(torch.zeros(hidden_size, dtype=dtype, device=device))
self.b2 = nn.Parameter(torch.zeros(hidden_size, dtype=dtype, device=device))
if nonlinearity == "tanh":
self.actfunc = torch.tanh
else:
self.actfunc = getattr(nnf, nonlinearity)
self.nonlinearity = nonlinearity
self.input_size = input_size
self.hidden_size = hidden_size
def forward(self, x: torch.Tensor, h: Optional[torch.Tensor] = None) -> tuple:
if h is None:
h = torch.zeros(self.hidden_size, dtype=x.dtype, device=x.device)
act = self.actfunc
W1 = self.W1
W2 = self.W2
b1 = self.b1.unsqueeze(-1)
b2 = self.b2.unsqueeze(-1)
x = x.unsqueeze(-1)
h = h.unsqueeze(-1)
y = act(((W1 @ x) + b1) + ((W2 @ h) + b2))
y = y.squeeze(-1)
return y, y
def __repr__(self) -> str:
clsname = type(self).__name__
return f"{clsname}(input_size={self.input_size}, hidden_size={self.hidden_size}, nonlinearity={repr(self.nonlinearity)})"
forward(self, x, h=None)
¶
Defines the computation performed at every call.
Should be overridden by all subclasses.
.. note::
Although the recipe for forward pass needs to be defined within
this function, one should call the :class:Module instance afterwards
instead of this since the former takes care of running the
registered hooks while the latter silently ignores them.
Source code in evotorch/neuroevolution/net/layers.py
def forward(self, x: torch.Tensor, h: Optional[torch.Tensor] = None) -> tuple:
if h is None:
h = torch.zeros(self.hidden_size, dtype=x.dtype, device=x.device)
act = self.actfunc
W1 = self.W1
W2 = self.W2
b1 = self.b1.unsqueeze(-1)
b2 = self.b2.unsqueeze(-1)
x = x.unsqueeze(-1)
h = h.unsqueeze(-1)
y = act(((W1 @ x) + b1) + ((W2 @ h) + b2))
y = y.squeeze(-1)
return y, y
RecurrentNet (Module)
¶
Source code in evotorch/neuroevolution/net/layers.py
class RNN(nn.Module):
def __init__(
self,
input_size: int,
hidden_size: int,
nonlinearity: str = "tanh",
*,
dtype: torch.dtype = torch.float32,
device: Union[str, torch.device] = "cpu",
):
super().__init__()
input_size = int(input_size)
hidden_size = int(hidden_size)
nonlinearity = str(nonlinearity)
self.W1 = nn.Parameter(torch.randn(hidden_size, input_size, dtype=dtype, device=device))
self.W2 = nn.Parameter(torch.randn(hidden_size, hidden_size, dtype=dtype, device=device))
self.b1 = nn.Parameter(torch.zeros(hidden_size, dtype=dtype, device=device))
self.b2 = nn.Parameter(torch.zeros(hidden_size, dtype=dtype, device=device))
if nonlinearity == "tanh":
self.actfunc = torch.tanh
else:
self.actfunc = getattr(nnf, nonlinearity)
self.nonlinearity = nonlinearity
self.input_size = input_size
self.hidden_size = hidden_size
def forward(self, x: torch.Tensor, h: Optional[torch.Tensor] = None) -> tuple:
if h is None:
h = torch.zeros(self.hidden_size, dtype=x.dtype, device=x.device)
act = self.actfunc
W1 = self.W1
W2 = self.W2
b1 = self.b1.unsqueeze(-1)
b2 = self.b2.unsqueeze(-1)
x = x.unsqueeze(-1)
h = h.unsqueeze(-1)
y = act(((W1 @ x) + b1) + ((W2 @ h) + b2))
y = y.squeeze(-1)
return y, y
def __repr__(self) -> str:
clsname = type(self).__name__
return f"{clsname}(input_size={self.input_size}, hidden_size={self.hidden_size}, nonlinearity={repr(self.nonlinearity)})"
forward(self, x, h=None)
¶
Defines the computation performed at every call.
Should be overridden by all subclasses.
.. note::
Although the recipe for forward pass needs to be defined within
this function, one should call the :class:Module instance afterwards
instead of this since the former takes care of running the
registered hooks while the latter silently ignores them.
Source code in evotorch/neuroevolution/net/layers.py
def forward(self, x: torch.Tensor, h: Optional[torch.Tensor] = None) -> tuple:
if h is None:
h = torch.zeros(self.hidden_size, dtype=x.dtype, device=x.device)
act = self.actfunc
W1 = self.W1
W2 = self.W2
b1 = self.b1.unsqueeze(-1)
b2 = self.b2.unsqueeze(-1)
x = x.unsqueeze(-1)
h = h.unsqueeze(-1)
y = act(((W1 @ x) + b1) + ((W2 @ h) + b2))
y = y.squeeze(-1)
return y, y
Round (Module)
¶
A small torch module for rounding the values of an input tensor
Source code in evotorch/neuroevolution/net/layers.py
class Round(nn.Module):
"""A small torch module for rounding the values of an input tensor"""
def __init__(self, ndigits: int = 0):
nn.Module.__init__(self)
self._ndigits = int(ndigits)
self._q = 10.0**self._ndigits
def forward(self, x):
x = x * self._q
x = torch.round(x)
x = x / self._q
return x
def extra_repr(self):
return "ndigits=" + str(self._ndigits)
extra_repr(self)
¶
Set the extra representation of the module
To print customized extra information, you should re-implement this method in your own modules. Both single-line and multi-line strings are acceptable.
forward(self, x)
¶
Defines the computation performed at every call.
Should be overridden by all subclasses.
.. note::
Although the recipe for forward pass needs to be defined within
this function, one should call the :class:Module instance afterwards
instead of this since the former takes care of running the
registered hooks while the latter silently ignores them.
Slice (Module)
¶
A small torch module for getting the slice of an input tensor
Source code in evotorch/neuroevolution/net/layers.py
class Slice(nn.Module):
"""A small torch module for getting the slice of an input tensor"""
def __init__(self, from_index: int, to_index: int):
"""`__init__(...)`: Initialize the Slice operator.
Args:
from_index: The index from which the slice begins.
to_index: The exclusive index at which the slice ends.
"""
nn.Module.__init__(self)
self._from_index = from_index
self._to_index = to_index
def forward(self, x):
return x[self._from_index : self._to_index]
def extra_repr(self):
return "from_index={}, to_index={}".format(self._from_index, self._to_index)
__init__(self, from_index, to_index)
special
¶
__init__(...): Initialize the Slice operator.
Parameters:
| Name | Type | Description | Default |
|---|---|---|---|
from_index |
int |
The index from which the slice begins. |
required |
to_index |
int |
The exclusive index at which the slice ends. |
required |
Source code in evotorch/neuroevolution/net/layers.py
extra_repr(self)
¶
Set the extra representation of the module
To print customized extra information, you should re-implement this method in your own modules. Both single-line and multi-line strings are acceptable.
forward(self, x)
¶
Defines the computation performed at every call.
Should be overridden by all subclasses.
.. note::
Although the recipe for forward pass needs to be defined within
this function, one should call the :class:Module instance afterwards
instead of this since the former takes care of running the
registered hooks while the latter silently ignores them.
StructuredControlNet (Module)
¶
Structured Control Net.
This is a control network consisting of two components: (i) a non-linear component, which is a feed-forward network; and (ii) a linear component, which is a linear layer. Both components take the input vector provided to the structured control network. The final output is the sum of the outputs of both components.
Reference
Mario Srouji, Jian Zhang, Ruslan Salakhutdinov (2018). Structured Control Nets for Deep Reinforcement Learning.
Source code in evotorch/neuroevolution/net/layers.py
class StructuredControlNet(nn.Module):
"""Structured Control Net.
This is a control network consisting of two components:
(i) a non-linear component, which is a feed-forward network; and
(ii) a linear component, which is a linear layer.
Both components take the input vector provided to the
structured control network.
The final output is the sum of the outputs of both components.
Reference:
Mario Srouji, Jian Zhang, Ruslan Salakhutdinov (2018).
Structured Control Nets for Deep Reinforcement Learning.
"""
def __init__(
self,
*,
in_features: int,
out_features: int,
num_layers: int,
hidden_size: int,
bias: bool = True,
nonlinearity: Union[str, Callable] = "tanh",
):
"""`__init__(...)`: Initialize the structured control net.
Args:
in_features: Length of the input vector
out_features: Length of the output vector
num_layers: Number of hidden layers for the non-linear component
hidden_size: Number of neurons in a hidden layer of the
non-linear component
bias: Whether or not the linear component is to have bias
nonlinearity: Activation function
"""
nn.Module.__init__(self)
self._in_features = in_features
self._out_features = out_features
self._num_layers = num_layers
self._hidden_size = hidden_size
self._bias = bias
self._nonlinearity = nonlinearity
self._linear_component = nn.Linear(
in_features=self._in_features, out_features=self._out_features, bias=self._bias
)
self._nonlinear_component = FeedForwardNet(
input_size=self._in_features,
layers=(
list((self._hidden_size, self._nonlinearity) for _ in range(self._num_layers))
+ [(self._out_features, self._nonlinearity)]
),
)
def forward(self, x: torch.Tensor) -> torch.Tensor:
"""TODO: documentation"""
return self._linear_component(x) + self._nonlinear_component(x)
@property
def in_features(self):
"""TODO: documentation"""
return self._in_features
@property
def out_features(self):
"""TODO: documentation"""
return self._out_features
@property
def num_layers(self):
"""TODO: documentation"""
return self._num_layers
@property
def hidden_size(self):
"""TODO: documentation"""
return self._hidden_size
@property
def bias(self):
"""TODO: documentation"""
return self._bias
@property
def nonlinearity(self):
"""TODO: documentation"""
return self._nonlinearity
bias
property
readonly
¶
hidden_size
property
readonly
¶
in_features
property
readonly
¶
nonlinearity
property
readonly
¶
num_layers
property
readonly
¶
out_features
property
readonly
¶
__init__(self, *, in_features, out_features, num_layers, hidden_size, bias=True, nonlinearity='tanh')
special
¶
__init__(...): Initialize the structured control net.
Parameters:
| Name | Type | Description | Default |
|---|---|---|---|
in_features |
int |
Length of the input vector |
required |
out_features |
int |
Length of the output vector |
required |
num_layers |
int |
Number of hidden layers for the non-linear component |
required |
hidden_size |
int |
Number of neurons in a hidden layer of the non-linear component |
required |
bias |
bool |
Whether or not the linear component is to have bias |
True |
nonlinearity |
Union[str, Callable] |
Activation function |
'tanh' |
Source code in evotorch/neuroevolution/net/layers.py
def __init__(
self,
*,
in_features: int,
out_features: int,
num_layers: int,
hidden_size: int,
bias: bool = True,
nonlinearity: Union[str, Callable] = "tanh",
):
"""`__init__(...)`: Initialize the structured control net.
Args:
in_features: Length of the input vector
out_features: Length of the output vector
num_layers: Number of hidden layers for the non-linear component
hidden_size: Number of neurons in a hidden layer of the
non-linear component
bias: Whether or not the linear component is to have bias
nonlinearity: Activation function
"""
nn.Module.__init__(self)
self._in_features = in_features
self._out_features = out_features
self._num_layers = num_layers
self._hidden_size = hidden_size
self._bias = bias
self._nonlinearity = nonlinearity
self._linear_component = nn.Linear(
in_features=self._in_features, out_features=self._out_features, bias=self._bias
)
self._nonlinear_component = FeedForwardNet(
input_size=self._in_features,
layers=(
list((self._hidden_size, self._nonlinearity) for _ in range(self._num_layers))
+ [(self._out_features, self._nonlinearity)]
),
)
forward(self, x)
¶
misc
¶
Utilities for reading and for writing neural network parameters
count_parameters(net)
¶
Get the number of parameters the network.
Parameters:
| Name | Type | Description | Default |
|---|---|---|---|
net |
Module |
The torch module whose parameters will be counted. |
required |
Returns:
| Type | Description |
|---|---|
int |
The number of parameters, as an integer. |
Source code in evotorch/neuroevolution/net/misc.py
device_of_module(m, default=None)
¶
Get the device in which the module exists.
This function looks at the first parameter of the module, and returns its device. This function is not meant to be used on modules whose parameters exist on different devices.
Parameters:
| Name | Type | Description | Default |
|---|---|---|---|
m |
Module |
The module whose device is being queried. |
required |
default |
Union[str, torch.device] |
The fallback device to return if the module has no parameters. If this is left as None, the fallback device is assumed to be "cpu". |
None |
Returns:
| Type | Description |
|---|---|
device |
The device of the module, determined from its first parameter. |
Source code in evotorch/neuroevolution/net/misc.py
def device_of_module(m: nn.Module, default: Optional[Union[str, torch.device]] = None) -> torch.device:
"""
Get the device in which the module exists.
This function looks at the first parameter of the module, and returns
its device. This function is not meant to be used on modules whose
parameters exist on different devices.
Args:
m: The module whose device is being queried.
default: The fallback device to return if the module has no
parameters. If this is left as None, the fallback device
is assumed to be "cpu".
Returns:
The device of the module, determined from its first parameter.
"""
if default is None:
default = torch.device("cpu")
device = default
for p in m.parameters():
device = p.device
break
return device
fill_parameters(net, vector)
¶
Fill the parameters of a torch module (net) from a vector.
No gradient information is kept.
The vector's length must be exactly the same with the number of parameters of the PyTorch module.
Parameters:
| Name | Type | Description | Default |
|---|---|---|---|
net |
Module |
The torch module whose parameter values will be filled. |
required |
vector |
Tensor |
A 1-D torch tensor which stores the parameter values. |
required |
Source code in evotorch/neuroevolution/net/misc.py
@torch.no_grad()
def fill_parameters(net: nn.Module, vector: torch.Tensor):
"""Fill the parameters of a torch module (net) from a vector.
No gradient information is kept.
The vector's length must be exactly the same with the number
of parameters of the PyTorch module.
Args:
net: The torch module whose parameter values will be filled.
vector: A 1-D torch tensor which stores the parameter values.
"""
address = 0
for p in net.parameters():
d = p.data.view(-1)
n = len(d)
d[:] = torch.as_tensor(vector[address : address + n], device=d.device)
address += n
if address != len(vector):
raise IndexError("The parameter vector is larger than expected")
parameter_vector(net, *, device=None)
¶
Get all the parameters of a torch module (net) into a vector
No gradient information is kept.
Parameters:
| Name | Type | Description | Default |
|---|---|---|---|
net |
Module |
The torch module whose parameters will be extracted. |
required |
device |
Union[str, torch.device] |
The device in which the parameter vector will be constructed. If the network has parameter across multiple devices, you can specify this argument so that concatenation of all the parameters will be successful. |
None |
Returns:
| Type | Description |
|---|---|
Tensor |
The parameters of the module in a 1-D tensor. |
Source code in evotorch/neuroevolution/net/misc.py
@torch.no_grad()
def parameter_vector(net: nn.Module, *, device: Optional[Device] = None) -> torch.Tensor:
"""Get all the parameters of a torch module (net) into a vector
No gradient information is kept.
Args:
net: The torch module whose parameters will be extracted.
device: The device in which the parameter vector will be constructed.
If the network has parameter across multiple devices,
you can specify this argument so that concatenation of all the
parameters will be successful.
Returns:
The parameters of the module in a 1-D tensor.
"""
dev_kwarg = {} if device is None else {"device": device}
all_vectors = []
for p in net.parameters():
all_vectors.append(torch.as_tensor(p.data.view(-1), **dev_kwarg))
return torch.cat(all_vectors)
multilayered
¶
MultiLayered (Module)
¶
Source code in evotorch/neuroevolution/net/multilayered.py
class MultiLayered(nn.Module):
def __init__(self, *layers: nn.Module):
super().__init__()
self._submodules = nn.ModuleList(layers)
def forward(self, x: torch.Tensor, h: Optional[dict] = None):
if h is None:
h = {}
new_h = {}
for i, layer in enumerate(self._submodules):
layer_h = h.get(i, None)
if layer_h is None:
layer_result = layer(x)
else:
layer_result = layer(x, h[i])
if isinstance(layer_result, tuple):
if len(layer_result) == 2:
x, layer_new_h = layer_result
else:
raise ValueError(
f"The layer number {i} returned a tuple of length {len(layer_result)}."
f" A tensor or a tuple of two elements was expected."
)
elif isinstance(layer_result, torch.Tensor):
x = layer_result
layer_new_h = None
else:
raise TypeError(
f"The layer number {i} returned an object of type {type(layer_result)}."
f" A tensor or a tuple of two elements was expected."
)
if layer_new_h is not None:
new_h[i] = layer_new_h
if len(new_h) == 0:
return x
else:
return x, new_h
def __iter__(self):
return self._submodules.__iter__()
def __getitem__(self, i):
return self._submodules[i]
def append(self, module: nn.Module):
self._submodules.append(module)
forward(self, x, h=None)
¶
Defines the computation performed at every call.
Should be overridden by all subclasses.
.. note::
Although the recipe for forward pass needs to be defined within
this function, one should call the :class:Module instance afterwards
instead of this since the former takes care of running the
registered hooks while the latter silently ignores them.
Source code in evotorch/neuroevolution/net/multilayered.py
def forward(self, x: torch.Tensor, h: Optional[dict] = None):
if h is None:
h = {}
new_h = {}
for i, layer in enumerate(self._submodules):
layer_h = h.get(i, None)
if layer_h is None:
layer_result = layer(x)
else:
layer_result = layer(x, h[i])
if isinstance(layer_result, tuple):
if len(layer_result) == 2:
x, layer_new_h = layer_result
else:
raise ValueError(
f"The layer number {i} returned a tuple of length {len(layer_result)}."
f" A tensor or a tuple of two elements was expected."
)
elif isinstance(layer_result, torch.Tensor):
x = layer_result
layer_new_h = None
else:
raise TypeError(
f"The layer number {i} returned an object of type {type(layer_result)}."
f" A tensor or a tuple of two elements was expected."
)
if layer_new_h is not None:
new_h[i] = layer_new_h
if len(new_h) == 0:
return x
else:
return x, new_h
parser
¶
Utilities for parsing string representations of neural net policies
NetParsingError (Exception)
¶
Representation of a parsing error
Source code in evotorch/neuroevolution/net/parser.py
class NetParsingError(Exception):
"""
Representation of a parsing error
"""
def __init__(
self,
message: str,
lineno: Optional[int] = None,
col_offset: Optional[int] = None,
original_error: Optional[Exception] = None,
):
"""
`__init__(...)`: Initialize the NetParsingError.
Args:
message: Error message, as string.
lineno: Erroneous line number in the string representation of the
neural network structure.
col_offset: Erroneous column number in the string representation
of the neural network structure.
original_error: If another error caused this parsing error,
that original error can be attached to this `NetParsingError`
instance via this argument.
"""
super().__init__()
self.message = message
self.lineno = lineno
self.col_offset = col_offset
self.original_error = original_error
def _to_string(self) -> str:
parts = []
parts.append(type(self).__name__)
if self.lineno is not None:
parts.append(" at line(")
parts.append(str(self.lineno - 1))
parts.append(")")
if self.col_offset is not None:
parts.append(" at column(")
parts.append(str(self.col_offset + 1))
parts.append(")")
parts.append(": ")
parts.append(self.message)
return "".join(parts)
def __str__(self) -> str:
return self._to_string()
def __repr__(self) -> str:
return self._to_string()
__init__(self, message, lineno=None, col_offset=None, original_error=None)
special
¶
__init__(...): Initialize the NetParsingError.
Parameters:
| Name | Type | Description | Default |
|---|---|---|---|
message |
str |
Error message, as string. |
required |
lineno |
Optional[int] |
Erroneous line number in the string representation of the neural network structure. |
None |
col_offset |
Optional[int] |
Erroneous column number in the string representation of the neural network structure. |
None |
original_error |
Optional[Exception] |
If another error caused this parsing error,
that original error can be attached to this |
None |
Source code in evotorch/neuroevolution/net/parser.py
def __init__(
self,
message: str,
lineno: Optional[int] = None,
col_offset: Optional[int] = None,
original_error: Optional[Exception] = None,
):
"""
`__init__(...)`: Initialize the NetParsingError.
Args:
message: Error message, as string.
lineno: Erroneous line number in the string representation of the
neural network structure.
col_offset: Erroneous column number in the string representation
of the neural network structure.
original_error: If another error caused this parsing error,
that original error can be attached to this `NetParsingError`
instance via this argument.
"""
super().__init__()
self.message = message
self.lineno = lineno
self.col_offset = col_offset
self.original_error = original_error
str_to_net(s, **constants)
¶
Read a string representation of a neural net structure,
and return a torch.nn.Module instance out of it.
Let us imagine that one wants to describe the following neural network structure:
from torch import nn
net = nn.Sequential(
nn.Linear(8, 16), nn.Tanh(), nn.Linear(16, 4, bias=False), nn.ReLU()
)
By using str_to_net(...) one can construct an equivalent
module via:
from evotorch.neuroevolution.net import str_to_net
net = str_to_net("Linear(8, 16) >> Tanh() >> Linear(16, 4, bias=False) >> ReLU()")
The string can also be multi-line:
One can also define constants for using them in strings:
net = str_to_net(
'''
Linear(input_size, hidden_size)
>> Tanh()
>> Linear(hidden_size, output_size, bias=False)
>> ReLU()
''',
input_size=8,
hidden_size=16,
output_size=4,
)
In the neural net structure string, when one refers to a module type,
say, Linear, first the name Linear is searched for in the namespace
evotorch.neuroevolution.net.layers, and then in the namespace torch.nn.
In the case of Linear, the searched name exists in torch.nn,
and therefore, the layer type to be instantiated is accepted as
torch.nn.Linear.
Instead of Linear, if one had used the name, say,
StructuredControlNet, then, the layer type to be instantiated
would be evotorch.neuroevolution.net.layers.StructuredControlNet.
The namespace evotorch.neuroevolution.net.layers contains its own
implementations for RNN and LSTM. These recurrent layer implementations
work similarly to their counterparts torch.nn.RNN and torch.nn.LSTM,
except that EvoTorch's implementations do not expect the data with extra
leftmost dimensions for batching and for timesteps. Instead, they expect
to receive a single input and a single current hidden state, and produce
a single output and a single new hidden state. These recurrent layer
implementations of EvoTorch can be used within a neural net structure
string. Therefore, the following examples are valid:
rnn1 = str_to_net("RNN(4, 8) >> Linear(8, 2)")
rnn2 = str_to_net(
'''
Linear(4, 10)
>> Tanh()
>> RNN(input_size=10, hidden_size=24, nonlinearity='tanh'
>> Linear(24, 2)
'''
)
lstm1 = str_to_net("LSTM(4, 32) >> Linear(32, 2)")
lstm2 = str_to_net("LSTM(input_size=4, hidden_size=32) >> Linear(32, 2)")
Notes regarding usage with evotorch.neuroevolution.GymNE
or with evotorch.neuroevolution.VecGymNE:
While instantiating a GymNE or a VecGymNE, one can specify a neural
net structure string as the policy. Therefore, while filling the policy
string for a GymNE, all these rules mentioned above apply. Additionally,
while using str_to_net(...) internally, GymNE and VecGymNE define
these extra constants:
obs_length (length of the observation vector),
act_length (length of the action vector for continuous-action
environments, or number of actions for discrete-action
environments), and
obs_shape (shape of the observation as a tuple, assuming that the
observation space is of type gym.spaces.Box, usable within the string
like obs_shape[0], obs_shape[1], etc., or simply obs_shape to refer
to the entire tuple).
Therefore, while instantiating a GymNE or a VecGymNE, one can define a
single-hidden-layered policy via this string:
In the policy string above, one might choose to omit the last Tanh(), as
GymNE and VecGymNE will clip the final output of the policy to conform
to the action boundaries defined by the target reinforcement learning
environment, and such a clipping operation might be seen as using an
activation function similar to hard-tanh anyway.
Parameters:
| Name | Type | Description | Default |
|---|---|---|---|
s |
str |
The string which expresses the neural net structure. |
required |
Returns:
| Type | Description |
|---|---|
Module |
The PyTorch module of the specified structure. |
Source code in evotorch/neuroevolution/net/parser.py
def str_to_net(s: str, **constants) -> nn.Module:
"""
Read a string representation of a neural net structure,
and return a `torch.nn.Module` instance out of it.
Let us imagine that one wants to describe the following
neural network structure:
```python
from torch import nn
net = nn.Sequential(
nn.Linear(8, 16), nn.Tanh(), nn.Linear(16, 4, bias=False), nn.ReLU()
)
```
By using `str_to_net(...)` one can construct an equivalent
module via:
```python
from evotorch.neuroevolution.net import str_to_net
net = str_to_net("Linear(8, 16) >> Tanh() >> Linear(16, 4, bias=False) >> ReLU()")
```
The string can also be multi-line:
```python
net = str_to_net(
'''
Linear(8, 16)
>> Tanh()
>> Linear(16, 4, bias=False)
>> ReLU()
'''
)
```
One can also define constants for using them in strings:
```python
net = str_to_net(
'''
Linear(input_size, hidden_size)
>> Tanh()
>> Linear(hidden_size, output_size, bias=False)
>> ReLU()
''',
input_size=8,
hidden_size=16,
output_size=4,
)
```
In the neural net structure string, when one refers to a module type,
say, `Linear`, first the name `Linear` is searched for in the namespace
`evotorch.neuroevolution.net.layers`, and then in the namespace `torch.nn`.
In the case of `Linear`, the searched name exists in `torch.nn`,
and therefore, the layer type to be instantiated is accepted as
`torch.nn.Linear`.
Instead of `Linear`, if one had used the name, say,
`StructuredControlNet`, then, the layer type to be instantiated
would be `evotorch.neuroevolution.net.layers.StructuredControlNet`.
The namespace `evotorch.neuroevolution.net.layers` contains its own
implementations for RNN and LSTM. These recurrent layer implementations
work similarly to their counterparts `torch.nn.RNN` and `torch.nn.LSTM`,
except that EvoTorch's implementations do not expect the data with extra
leftmost dimensions for batching and for timesteps. Instead, they expect
to receive a single input and a single current hidden state, and produce
a single output and a single new hidden state. These recurrent layer
implementations of EvoTorch can be used within a neural net structure
string. Therefore, the following examples are valid:
```python
rnn1 = str_to_net("RNN(4, 8) >> Linear(8, 2)")
rnn2 = str_to_net(
'''
Linear(4, 10)
>> Tanh()
>> RNN(input_size=10, hidden_size=24, nonlinearity='tanh'
>> Linear(24, 2)
'''
)
lstm1 = str_to_net("LSTM(4, 32) >> Linear(32, 2)")
lstm2 = str_to_net("LSTM(input_size=4, hidden_size=32) >> Linear(32, 2)")
```
**Notes regarding usage with `evotorch.neuroevolution.GymNE`
or with `evotorch.neuroevolution.VecGymNE`:**
While instantiating a `GymNE` or a `VecGymNE`, one can specify a neural
net structure string as the policy. Therefore, while filling the policy
string for a `GymNE`, all these rules mentioned above apply. Additionally,
while using `str_to_net(...)` internally, `GymNE` and `VecGymNE` define
these extra constants:
`obs_length` (length of the observation vector),
`act_length` (length of the action vector for continuous-action
environments, or number of actions for discrete-action
environments), and
`obs_shape` (shape of the observation as a tuple, assuming that the
observation space is of type `gym.spaces.Box`, usable within the string
like `obs_shape[0]`, `obs_shape[1]`, etc., or simply `obs_shape` to refer
to the entire tuple).
Therefore, while instantiating a `GymNE` or a `VecGymNE`, one can define a
single-hidden-layered policy via this string:
```
"Linear(obs_length, 16) >> Tanh() >> Linear(16, act_length) >> Tanh()"
```
In the policy string above, one might choose to omit the last `Tanh()`, as
`GymNE` and `VecGymNE` will clip the final output of the policy to conform
to the action boundaries defined by the target reinforcement learning
environment, and such a clipping operation might be seen as using an
activation function similar to hard-tanh anyway.
Args:
s: The string which expresses the neural net structure.
Returns:
The PyTorch module of the specified structure.
"""
s = f"(\n{s}\n)"
return _process_expr(ast.parse(s, mode="eval").body, constants=constants)
rl
¶
This namespace provides various reinforcement learning utilities.
ActClipWrapperModule (Module)
¶
Source code in evotorch/neuroevolution/net/rl.py
class ActClipWrapperModule(nn.Module):
def __init__(self, wrapped_module: nn.Module, obs_space: Box):
super().__init__()
device = device_of_module(wrapped_module)
if not isinstance(obs_space, Box):
raise TypeError(f"Unrecognized observation space: {obs_space}")
self.wrapped_module = wrapped_module
self.register_buffer("_low", torch.from_numpy(obs_space.low).to(device))
self.register_buffer("_high", torch.from_numpy(obs_space.high).to(device))
def forward(self, x: torch.Tensor, h: Any = None) -> Union[torch.Tensor, tuple]:
if h is None:
result = self.wrapped_module(x)
else:
result = self.wrapped_module(x, h)
if isinstance(result, tuple):
x, h = result
got_h = True
else:
x = result
h = None
got_h = False
x = torch.max(x, self._low)
x = torch.min(x, self._high)
if got_h:
return x, h
else:
return x
forward(self, x, h=None)
¶
Defines the computation performed at every call.
Should be overridden by all subclasses.
.. note::
Although the recipe for forward pass needs to be defined within
this function, one should call the :class:Module instance afterwards
instead of this since the former takes care of running the
registered hooks while the latter silently ignores them.
Source code in evotorch/neuroevolution/net/rl.py
def forward(self, x: torch.Tensor, h: Any = None) -> Union[torch.Tensor, tuple]:
if h is None:
result = self.wrapped_module(x)
else:
result = self.wrapped_module(x, h)
if isinstance(result, tuple):
x, h = result
got_h = True
else:
x = result
h = None
got_h = False
x = torch.max(x, self._low)
x = torch.min(x, self._high)
if got_h:
return x, h
else:
return x
AliveBonusScheduleWrapper (Wrapper)
¶
A Wrapper which awards the agent for being alive in a scheduled manner This wrapper is meant to be used for non-vectorized environments.
Source code in evotorch/neuroevolution/net/rl.py
class AliveBonusScheduleWrapper(gym.Wrapper):
"""
A Wrapper which awards the agent for being alive in a scheduled manner
This wrapper is meant to be used for non-vectorized environments.
"""
def __init__(self, env: gym.Env, alive_bonus_schedule: tuple, **kwargs):
"""
`__init__(...)`: Initialize the AliveBonusScheduleWrapper.
Args:
env: Environment to wrap.
alive_bonus_schedule: If given as a tuple `(t, b)`, an alive
bonus `b` will be added onto all the rewards beyond the
timestep `t`.
If given as a tuple `(t0, t1, b)`, a partial (linearly
increasing towards `b`) alive bonus will be added onto
all the rewards between the timesteps `t0` and `t1`,
and a full alive bonus (which equals to `b`) will be added
onto all the rewards beyond the timestep `t1`.
kwargs: Expected in the form of additional keyword arguments,
these will be passed to the initialization method of the
superclass.
"""
super().__init__(env, **kwargs)
self.__t: Optional[int] = None
if len(alive_bonus_schedule) == 3:
self.__t0, self.__t1, self.__bonus = (
int(alive_bonus_schedule[0]),
int(alive_bonus_schedule[1]),
float(alive_bonus_schedule[2]),
)
elif len(alive_bonus_schedule) == 2:
self.__t0, self.__t1, self.__bonus = (
int(alive_bonus_schedule[0]),
int(alive_bonus_schedule[0]),
float(alive_bonus_schedule[1]),
)
else:
raise ValueError(
f"The argument `alive_bonus_schedule` was expected to have 2 or 3 elements."
f" However, its value is {repr(alive_bonus_schedule)} (having {len(alive_bonus_schedule)} elements)."
)
if self.__t1 > self.__t0:
self.__gap = self.__t1 - self.__t0
else:
self.__gap = None
def reset(self, *args, **kwargs):
self.__t = 0
return self.env.reset(*args, **kwargs)
def step(self, action) -> tuple:
step_result = self.env.step(action)
self.__t += 1
observation = step_result[0]
reward = step_result[1]
rest = step_result[2:]
if self.__t >= self.__t1:
reward = reward + self.__bonus
elif (self.__gap is not None) and (self.__t >= self.__t0):
reward = reward + ((self.__t - self.__t0) / self.__gap) * self.__bonus
return (observation, reward) + rest
__init__(self, env, alive_bonus_schedule, **kwargs)
special
¶
__init__(...): Initialize the AliveBonusScheduleWrapper.
Parameters:
| Name | Type | Description | Default |
|---|---|---|---|
env |
Env |
Environment to wrap. |
required |
alive_bonus_schedule |
tuple |
If given as a tuple |
required |
kwargs |
Expected in the form of additional keyword arguments, these will be passed to the initialization method of the superclass. |
{} |
Source code in evotorch/neuroevolution/net/rl.py
def __init__(self, env: gym.Env, alive_bonus_schedule: tuple, **kwargs):
"""
`__init__(...)`: Initialize the AliveBonusScheduleWrapper.
Args:
env: Environment to wrap.
alive_bonus_schedule: If given as a tuple `(t, b)`, an alive
bonus `b` will be added onto all the rewards beyond the
timestep `t`.
If given as a tuple `(t0, t1, b)`, a partial (linearly
increasing towards `b`) alive bonus will be added onto
all the rewards between the timesteps `t0` and `t1`,
and a full alive bonus (which equals to `b`) will be added
onto all the rewards beyond the timestep `t1`.
kwargs: Expected in the form of additional keyword arguments,
these will be passed to the initialization method of the
superclass.
"""
super().__init__(env, **kwargs)
self.__t: Optional[int] = None
if len(alive_bonus_schedule) == 3:
self.__t0, self.__t1, self.__bonus = (
int(alive_bonus_schedule[0]),
int(alive_bonus_schedule[1]),
float(alive_bonus_schedule[2]),
)
elif len(alive_bonus_schedule) == 2:
self.__t0, self.__t1, self.__bonus = (
int(alive_bonus_schedule[0]),
int(alive_bonus_schedule[0]),
float(alive_bonus_schedule[1]),
)
else:
raise ValueError(
f"The argument `alive_bonus_schedule` was expected to have 2 or 3 elements."
f" However, its value is {repr(alive_bonus_schedule)} (having {len(alive_bonus_schedule)} elements)."
)
if self.__t1 > self.__t0:
self.__gap = self.__t1 - self.__t0
else:
self.__gap = None
reset(self, *args, **kwargs)
¶
step(self, action)
¶
Steps through the environment with action.
Source code in evotorch/neuroevolution/net/rl.py
def step(self, action) -> tuple:
step_result = self.env.step(action)
self.__t += 1
observation = step_result[0]
reward = step_result[1]
rest = step_result[2:]
if self.__t >= self.__t1:
reward = reward + self.__bonus
elif (self.__gap is not None) and (self.__t >= self.__t0):
reward = reward + ((self.__t - self.__t0) / self.__gap) * self.__bonus
return (observation, reward) + rest
ObsNormWrapperModule (Module)
¶
Source code in evotorch/neuroevolution/net/rl.py
class ObsNormWrapperModule(nn.Module):
def __init__(self, wrapped_module: nn.Module, rn: Union[RunningStat, RunningNorm]):
super().__init__()
device = device_of_module(wrapped_module)
self.wrapped_module = wrapped_module
with torch.no_grad():
normalizer = deepcopy(rn.to_layer()).to(device)
self.normalizer = normalizer
def forward(self, x: torch.Tensor, h: Any = None) -> Union[torch.Tensor, tuple]:
x = self.normalizer(x)
if h is None:
result = self.wrapped_module(x)
else:
result = self.wrapped_module(x, h)
if isinstance(result, tuple):
x, h = result
got_h = True
else:
x = result
h = None
got_h = False
if got_h:
return x, h
else:
return x
forward(self, x, h=None)
¶
Defines the computation performed at every call.
Should be overridden by all subclasses.
.. note::
Although the recipe for forward pass needs to be defined within
this function, one should call the :class:Module instance afterwards
instead of this since the former takes care of running the
registered hooks while the latter silently ignores them.
Source code in evotorch/neuroevolution/net/rl.py
def forward(self, x: torch.Tensor, h: Any = None) -> Union[torch.Tensor, tuple]:
x = self.normalizer(x)
if h is None:
result = self.wrapped_module(x)
else:
result = self.wrapped_module(x, h)
if isinstance(result, tuple):
x, h = result
got_h = True
else:
x = result
h = None
got_h = False
if got_h:
return x, h
else:
return x
reset_env(env)
¶
Reset a gym environment.
For gym 1.0, the plan is to have a reset(...) method which returns
a two-element tuple (observation, info) where info is an object
providing any additional information regarding the initial state of
the agent. However, the old (pre 1.0) gym API (and some environments
which were written with old gym compatibility in mind) has (or have)
a reset(...) method which returns a single object that is the
initial observation.
With the assumption that the observation space of the environment
is NOT tuple, this function can work with both pre-1.0 and (hopefully)
after-1.0 versions of gym, and always returns the initial observation.
Please do not use this function on environments whose observation
spaces or tuples, because then this function cannot distinguish between
environments whose reset(...) methods return a tuple and environments
whose reset(...) methods return a single observation object but that
observation object is a tuple.
Parameters:
| Name | Type | Description | Default |
|---|---|---|---|
env |
Env |
The gym environment which will be reset. |
required |
Returns:
| Type | Description |
|---|---|
Iterable |
The initial observation |
Source code in evotorch/neuroevolution/net/rl.py
def reset_env(env: gym.Env) -> Iterable:
"""
Reset a gym environment.
For gym 1.0, the plan is to have a `reset(...)` method which returns
a two-element tuple `(observation, info)` where `info` is an object
providing any additional information regarding the initial state of
the agent. However, the old (pre 1.0) gym API (and some environments
which were written with old gym compatibility in mind) has (or have)
a `reset(...)` method which returns a single object that is the
initial observation.
With the assumption that the observation space of the environment
is NOT tuple, this function can work with both pre-1.0 and (hopefully)
after-1.0 versions of gym, and always returns the initial observation.
Please do not use this function on environments whose observation
spaces or tuples, because then this function cannot distinguish between
environments whose `reset(...)` methods return a tuple and environments
whose `reset(...)` methods return a single observation object but that
observation object is a tuple.
Args:
env: The gym environment which will be reset.
Returns:
The initial observation
"""
result = env.reset()
if isinstance(result, tuple) and (len(result) == 2):
result = result[0]
return result
take_step_in_env(env, action)
¶
Take a step in the gym environment. Taking a step means performing the action provided via the arguments.
For gym 1.0, the plan is to have a step(...) method which returns a
5-elements tuple containing observation, reward, terminated,
truncated, info where terminated is a boolean indicating whether
or not the episode is terminated because of the actions taken within the
environment, and truncated is a boolean indicating whether or not the
episode is finished because the time limit is reached.
However, the old (pre 1.0) gym API (and some environments which were
written with old gym compatibility in mind) has (or have) a step(...)
method which returns 4 elements: observation, reward, done, info
where done is a boolean indicating whether or not the episode is
"done", either because of termination or because of truncation.
This function can work with both pre-1.0 and (hopefully) after-1.0
versions of gym, and always returns the 4-element tuple as its result.
Parameters:
| Name | Type | Description | Default |
|---|---|---|---|
env |
Env |
The gym environment in which the given action will be performed. |
required |
Returns:
| Type | Description |
|---|---|
tuple |
A tuple in the form |
Source code in evotorch/neuroevolution/net/rl.py
def take_step_in_env(env: gym.Env, action: Iterable) -> tuple:
"""
Take a step in the gym environment.
Taking a step means performing the action provided via the arguments.
For gym 1.0, the plan is to have a `step(...)` method which returns a
5-elements tuple containing `observation`, `reward`, `terminated`,
`truncated`, `info` where `terminated` is a boolean indicating whether
or not the episode is terminated because of the actions taken within the
environment, and `truncated` is a boolean indicating whether or not the
episode is finished because the time limit is reached.
However, the old (pre 1.0) gym API (and some environments which were
written with old gym compatibility in mind) has (or have) a `step(...)`
method which returns 4 elements: `observation`, `reward`, `done`, `info`
where `done` is a boolean indicating whether or not the episode is
"done", either because of termination or because of truncation.
This function can work with both pre-1.0 and (hopefully) after-1.0
versions of gym, and always returns the 4-element tuple as its result.
Args:
env: The gym environment in which the given action will be performed.
Returns:
A tuple in the form `(observation, reward, done, info)` where
`observation` is the observation received after performing the action,
`reward` is the amount of reward gained,
`done` is a boolean value indicating whether or not the episode has
ended, and
`info` is additional information (usually as a dictionary).
"""
result = env.step(action)
if isinstance(result, tuple):
n = len(result)
if n == 4:
observation, reward, done, info = result
elif n == 5:
observation, reward, terminated, truncated, info = result
done = terminated or truncated
else:
raise ValueError(
f"The result of the `step(...)` method of the gym environment"
f" was expected as a tuple of length 4 or 5."
f" However, the received result is {repr(result)}, which is"
f" of length {len(result)}."
)
else:
raise TypeError(
f"The result of the `step(...)` method of the gym environment"
f" was expected as a tuple of length 4 or 5."
f" However, the received result is {repr(result)}, which is"
f" of type {type(result)}."
)
return observation, reward, done, info
runningnorm
¶
CollectedStats (tuple)
¶
ObsNormLayer (Module)
¶
An observation normalizer which behaves as a PyTorch Module.
Source code in evotorch/neuroevolution/net/runningnorm.py
class ObsNormLayer(nn.Module):
"""
An observation normalizer which behaves as a PyTorch Module.
"""
def __init__(
self, mean: torch.Tensor, stdev: torch.Tensor, low: Optional[float] = None, high: Optional[float] = None
) -> None:
"""
`__init__(...)`: Initialize the ObsNormLayer.
Args:
mean: The mean according to which the observations are to be
normalized.
stdev: The standard deviation according to which the observations
are to be normalized.
low: Optionally a real number if the result of the normalization
is to be clipped. Represents the lower bound for the clipping
operation.
high: Optionally a real number if the result of the normalization
is to be clipped. Represents the upper bound for the clipping
operation.
"""
super().__init__()
self.register_buffer("_mean", mean)
self.register_buffer("_stdev", stdev)
self._lb = None if low is None else float(low)
self._ub = None if high is None else float(high)
def forward(self, x: torch.Tensor) -> torch.Tensor:
"""
Normalize an observation or a batch of observations.
Args:
x: The observation(s).
Returns:
The normalized counterpart of the observation(s).
"""
return _clamp((x - self._mean) / self._stdev, self._lb, self._ub)
__init__(self, mean, stdev, low=None, high=None)
special
¶
__init__(...): Initialize the ObsNormLayer.
Parameters:
| Name | Type | Description | Default |
|---|---|---|---|
mean |
Tensor |
The mean according to which the observations are to be normalized. |
required |
stdev |
Tensor |
The standard deviation according to which the observations are to be normalized. |
required |
low |
Optional[float] |
Optionally a real number if the result of the normalization is to be clipped. Represents the lower bound for the clipping operation. |
None |
high |
Optional[float] |
Optionally a real number if the result of the normalization is to be clipped. Represents the upper bound for the clipping operation. |
None |
Source code in evotorch/neuroevolution/net/runningnorm.py
def __init__(
self, mean: torch.Tensor, stdev: torch.Tensor, low: Optional[float] = None, high: Optional[float] = None
) -> None:
"""
`__init__(...)`: Initialize the ObsNormLayer.
Args:
mean: The mean according to which the observations are to be
normalized.
stdev: The standard deviation according to which the observations
are to be normalized.
low: Optionally a real number if the result of the normalization
is to be clipped. Represents the lower bound for the clipping
operation.
high: Optionally a real number if the result of the normalization
is to be clipped. Represents the upper bound for the clipping
operation.
"""
super().__init__()
self.register_buffer("_mean", mean)
self.register_buffer("_stdev", stdev)
self._lb = None if low is None else float(low)
self._ub = None if high is None else float(high)
forward(self, x)
¶
Normalize an observation or a batch of observations.
Parameters:
| Name | Type | Description | Default |
|---|---|---|---|
x |
Tensor |
The observation(s). |
required |
Returns:
| Type | Description |
|---|---|
Tensor |
The normalized counterpart of the observation(s). |
Source code in evotorch/neuroevolution/net/runningnorm.py
RunningNorm
¶
An online observation normalization tool
Source code in evotorch/neuroevolution/net/runningnorm.py
class RunningNorm:
"""
An online observation normalization tool
"""
def __init__(
self,
*,
shape: Union[tuple, int],
dtype: DType,
device: Optional[Device] = None,
min_variance: float = 1e-2,
clip: Optional[tuple] = None,
) -> None:
"""
`__init__(...)`: Initialize the RunningNorm
Args:
shape: Observation shape. Can be an integer or a tuple.
dtype: The dtype of the observations.
device: The device in which the observation stats are held.
If left as None, the device is assumed to be "cpu".
min_variance: A lower bound for the variance to be used in
the normalization computations.
In other words, if the computed variance according to the
collected observations ends up lower than `min_variance`,
this `min_variance` will be used instead (in an elementwise
manner) while computing the normalized observations.
As in Salimans et al. (2017), the default is 1e-2.
clip: Can be left as None (which is the default), or can be
given as a pair of real numbers.
This is used for clipping the observations after the
normalization operation.
In Salimans et al. (2017), (-5.0, +5.0) was used.
"""
# Make sure that the shape is stored as a torch.Size object.
if isinstance(shape, Iterable):
self._shape = torch.Size(shape)
else:
self._shape = torch.Size([int(shape)])
# Store the number of dimensions
self._ndim = len(self._shape)
# Store the dtype and the device
self._dtype = to_torch_dtype(dtype)
self._device = "cpu" if device is None else device
# Initialize the internally stored data as empty
self._sum: Optional[torch.Tensor] = None
self._sum_of_squares: Optional[torch.Tensor] = None
self._count: int = 0
# Store the minimum variance
self._min_variance = float(min_variance)
if clip is not None:
# If a clip tuple was provided, store the specified lower and upper bounds
lb, ub = clip
self._lb = float(lb)
self._ub = float(ub)
else:
# If a clip tuple was not provided the bounds are stored as None
self._lb = None
self._ub = None
def to(self, device: Device) -> "RunningNorm":
"""
If the target device is a different device, then make a copy of this
RunningNorm instance on the target device.
If the target device is the same with this RunningNorm's device, then
return this RunningNorm itself.
Args:
device: The target device.
Returns:
The RunningNorm on the target device. This can be a copy, or the
original RunningNorm instance itself.
"""
if torch.device(device) == torch.device(self.device):
return self
else:
new_running_norm = object.__new__(type(self))
already_handled = {"_sum", "_sum_of_squares", "_device"}
new_running_norm._sum = self._sum.to(device)
new_running_norm._sum_of_squares = self._sum_of_squares.to(device)
new_running_norm._device = device
for k, v in self.__dict__.items():
if k not in already_handled:
setattr(new_running_norm, k, deepcopy(v))
return new_running_norm
@property
def device(self) -> Device:
"""
The device in which the observation stats are held
"""
return self._device
@property
def dtype(self) -> DType:
"""
The dtype of the stored observation stats
"""
return self._dtype
@property
def shape(self) -> tuple:
"""
Observation shape
"""
return self._shape
@property
def min_variance(self) -> float:
"""
Minimum variance
"""
return self._min_variance
@property
def low(self) -> Optional[float]:
"""
The lower component of the bounds given in the `clip` tuple.
If `clip` was initialized as None, this is also None.
"""
return self._lb
@property
def high(self) -> Optional[float]:
"""
The higher (upper) component of the bounds given in the `clip` tuple.
If `clip` was initialized as None, this is also None.
"""
return self._ub
def _like_its_own(self, x: Iterable) -> torch.Tensor:
return torch.as_tensor(x, dtype=self._dtype, device=self._device)
def _verify(self, x: Iterable) -> torch.Tensor:
x = self._like_its_own(x)
if x.ndim == self._ndim:
if x.shape != self._shape:
raise ValueError(
f"This RunningNorm instance was initialized with shape: {self._shape}."
f" However, the provided tensor has an incompatible shape: {x._shape}."
)
elif x.ndim == (self._ndim + 1):
if x.shape[1:] != self._shape:
raise ValueError(
f"This RunningNorm instance was initialized with shape: {self._shape}."
f" The provided tensor is shaped {x.shape}."
f" Accepting the tensor's leftmost dimension as the batch size,"
f" the remaining shape is incompatible: {x.shape[1:]}"
)
else:
raise ValueError(
f"This RunningNorm instance was initialized with shape: {self._shape}."
f" The provided tensor is shaped {x.shape}."
f" The number of dimensions of the given tensor is incompatible."
)
return x
def _has_no_data(self) -> bool:
return (self._sum is None) and (self._sum_of_squares is None) and (self._count == 0)
def _has_data(self) -> bool:
return (self._sum is not None) and (self._sum_of_squares is not None) and (self._count > 0)
def reset(self):
"""
Remove all the collected observation data.
"""
self._sum = None
self._sum_of_squares = None
self._count = 0
@torch.no_grad()
def update(self, x: Union[Iterable, "RunningNorm"], mask: Optional[Iterable] = None, *, verify: bool = True):
"""
Update the stored stats with new observation data.
Args:
x: The new observation(s), as a PyTorch tensor, or any Iterable
that can be converted to a PyTorch tensor, or another
RunningNorm instance.
If given as a tensor or as an Iterable, the shape of `x` can
be the same with observation shape, or it can be augmented
with an extra leftmost dimension.
In the case of augmented dimension, `x` is interpreted not as
a single observation, but as a batch of observations.
If `x` is another RunningNorm instance, the stats stored by
this RunningNorm instance will be updated with all the data
stored by `x`.
mask: Can be given as a 1-dimensional Iterable of booleans ONLY
if `x` represents a batch of observations.
If a `mask` is provided, the i-th observation within the
observation batch `x` will be taken into account only if
the i-th item of the `mask` is True.
verify: Whether or not to verify the shape of the given Iterable
objects. The default is True.
"""
if isinstance(x, RunningNorm):
# If we are to update our stats according to another RunningNorm instance
if x._count > 0:
# We bother only if x is non-empty
if mask is not None:
# We were given another RunningNorm, not a batch of observations.
# So, we do not expect to receive a mask tensor.
# If a mask was provided, then this is an unexpected way of calling this function.
# We therefore raise an error.
raise ValueError(
"The `mask` argument is expected as None if the first argument is a RunningNorm."
" However, `mask` is found as something other than None."
)
if self._shape != x._shape:
# If the shapes of this RunningNorm and of the other RunningNorm
# do not match, then we cannot use `x` for updating our stats.
# It might be the case that `x` was initialized for another
# task, with differently sized observations.
# We therefore raise an error.
raise ValueError(
f"The RunningNorm to be updated has the shape {self._shape}"
f" The other RunningNorm has the shape {self._shape}"
f" These shapes are incompatible."
)
if self._has_no_data():
# If this RunningNorm has no data at all, then we clone the
# data of x.
self._sum = self._like_its_own(x._sum.clone())
self._sum_of_squares = self._like_its_own(x._sum_of_squares.clone())
self._count = x._count
elif self._has_data():
# If this RunningNorm has its own data, then we update the
# stored data with the data stored by x.
self._sum += self._like_its_own(x._sum)
self._sum_of_squares += self._like_its_own(x._sum_of_squares)
self._count += x._count
else:
assert False, "RunningNorm is in an invalid state! This might be a bug."
else:
# This is the case where the received argument x is not a
# RunningNorm object, but an Iterable.
if verify:
# If we have the `verify` flag, then we make sure that
# x is a tensor of the correct shape
x = self._verify(x)
if x.ndim == self._ndim:
# If the shape of x is exactly the same with the observation shape
# then we assume that x represents a single observation, and not a
# batch of observations.
if mask is not None:
# Since we are dealing with a single observation,
# we do not expect to receive a mask argument.
# If the mask argument was provided, then this is an unexpected
# usage of this function.
# We therefore raise an error.
raise ValueError(
"The `mask` argument is expected as None if the first argument is a single observation"
" (i.e. not a batch of observations, with an extra leftmost dimension)."
" However, `mask` is found as something other than None."
)
# Since x is a single observation,
# the sum of observations extracted from x is x itself,
# and the sum of squared observations extracted from x is
# the square of x itself.
sum_of_x = x
sum_of_x_squared = x.square()
# We extracted a single observation from x
n = 1
elif x.ndim == (self._ndim + 1):
# If the number of dimensions of x is one more than the number
# of dimensions of this RunningNorm, then we assume that x is a batch
# of observations.
if mask is not None:
# If a mask is provided, then we first make sure that it is a tensor
# of dtype bool in the correct device.
mask = torch.as_tensor(mask, dtype=torch.bool, device=self._device)
if mask.ndim != 1:
# We expect the mask to be 1-dimensional.
# If not, we raise an error.
raise ValueError(
f"The `mask` tensor was expected as a 1-dimensional tensor."
f" However, its shape is {mask.shape}."
)
if len(mask) != x.shape[0]:
# If the length of the mask is not the batch size of x,
# then there is a mismatch.
# We therefore raise an error.
raise ValueError(
f"The shape of the given tensor is {x.shape}."
f" Therefore, the batch size of observations is {x.shape[0]}."
f" However, the given `mask` tensor does not has an incompatible length: {len(mask)}."
)
# We compute how many True items we have in the mask.
# This integer gives us how many observations we extract from x.
n = int(torch.sum(torch.as_tensor(mask, dtype=torch.int64, device=self._device)))
# We now re-cast the mask as the observation dtype (so that True items turn to 1.0
# and False items turn to 0.0), and then increase its number of dimensions so that
# it can operate directly with x.
mask = self._like_its_own(mask).reshape(torch.Size([x.shape[0]] + ([1] * (x.ndim - 1))))
# Finally, we multiply x with the mask. This means that the observations with corresponding
# mask values as False are zeroed out.
x = x * mask
else:
# This is the case where we did not receive a mask.
# We can simply say that the number of observations to extract from x
# is the size of its leftmost dimension, i.e. the batch size.
n = x.shape[0]
# With or without a mask, we are now ready to extract the sum and sum of squares
# from x.
sum_of_x = torch.sum(x, dim=0)
sum_of_x_squared = torch.sum(x.square(), dim=0)
else:
# This is the case where the number of dimensions of x is unrecognized.
# This case is actually already checked by the _verify(...) method earlier.
# This defensive fallback case is only for when verify=False and it turned out
# that the ndim is invalid.
raise ValueError(f"Invalid shape: {x.shape}")
# At this point, we handled all the valid cases regarding the Iterable x,
# and we have our sum_of_x (sum of all observations), sum_of_squares
# (sum of all squared observations), and n (number of observations extracted
# from x).
if self._has_no_data():
# If our RunningNorm is empty, the observation data we extracted from x
# become our RunningNorm's new data.
self._sum = sum_of_x
self._sum_of_squares = sum_of_x_squared
self._count = n
elif self._has_data():
# If our RunningNorm is not empty, the stored data is updated with the
# data extracted from x.
self._sum += sum_of_x
self._sum_of_squares += sum_of_x_squared
self._count += n
else:
# This is an erroneous state where the internal data looks neither
# existent nor completely empty.
# This might be the result of a bug, or maybe this instance's
# protected variables were tempered with from the outside.
assert False, "RunningNorm is in an invalid state! This might be a bug."
@property
@torch.no_grad()
def stats(self) -> CollectedStats:
"""
The collected data's mean and standard deviation (stdev) in a tuple
"""
# Using the internally stored sum, sum_of_squares, and count,
# compute E[x] and E[x^2]
E_x = self._sum / self._count
E_x2 = self._sum_of_squares / self._count
# The mean is E[x]
mean = E_x
# The variance is E[x^2] - (E[x])^2, elementwise clipped such that
# it cannot go below min_variance
variance = _clamp(E_x2 - E_x.square(), self._min_variance, None)
# Standard deviation is finally computed as the square root of the variance
stdev = torch.sqrt(variance)
# Return the stats in a named tuple
return CollectedStats(mean=mean, stdev=stdev)
@property
def mean(self) -> torch.Tensor:
"""
The collected data's mean
"""
return self._sum / self._count
@property
def stdev(self) -> torch.Tensor:
"""
The collected data's standard deviation
"""
return self.stats.stdev
@property
def sum(self) -> torch.Tensor:
"""
The collected data's sum
"""
return self._sum
@property
def sum_of_squares(self) -> torch.Tensor:
"""
Sum of squares of the collected data
"""
return self._sum_of_squares
@property
def count(self) -> int:
"""
Number of observations encountered
"""
return self._count
@torch.no_grad()
def normalize(self, x: Iterable, *, result_as_numpy: Optional[bool] = None, verify: bool = True) -> Iterable:
"""
Normalize the given observation x.
Args:
x: The observation(s), as a PyTorch tensor, or any Iterable
that is convertable to a PyTorch tensor.
`x` can be a single observation, or it can be a batch
of observations (with an extra leftmost dimension).
result_as_numpy: Whether or not to return the normalized
observation as a numpy array.
If left as None (which is the default), then the returned
type depends on x: a PyTorch tensor is returned if x is a
PyTorch tensor, and a numpy array is returned otherwise.
If True, the result is always a numpy array.
If False, the result is always a PyTorch tensor.
verify: Whether or not to check the type and dimensions of x.
This is True by default.
Note that, if `verify` is False, this function will not
properly check the type of `x` and will assume that `x`
is a PyTorch tensor.
Returns:
The normalized observation, as a PyTorch tensor or a numpy array.
"""
if self._count == 0:
# If this RunningNorm instance has no data yet,
# then we do not know how to do the normalization.
# We therefore raise an error.
raise ValueError("Cannot do normalization because no data is collected yet.")
if verify:
# Here we verify the type and shape of x.
if result_as_numpy is None:
# If there is not an explicit request about the return type,
# we infer the return type from the type of x:
# if x is a tensor, we return a tensor;
# otherwise, we assume x to be a CPU-bound iterable, and
# therefore we return a numpy array.
result_as_numpy = not isinstance(x, torch.Tensor)
else:
result_as_numpy = bool(result_as_numpy)
# We call _verify() to make sure that x is of correct shape
# and is properly converted to a PyTorch tensor.
x = self._verify(x)
# We get the mean and stdev of the collected data
mean, stdev = self.stats
# Now we compute the normalized observation, clipped according to the
# lower and upper bounds expressed by the `clip` tuple, if exists.
result = _clamp((x - mean) / stdev, self._lb, self._ub)
if result_as_numpy:
# If we are to return the result as a numpy array, we do the
# necessary conversion.
result = result.cpu().numpy()
# Finally, return the result
return result
@torch.no_grad()
def update_and_normalize(self, x: Iterable, mask: Optional[Iterable] = None) -> Iterable:
"""
Update the observation stats according to x, then normalize x.
Args:
x: The observation(s), as a PyTorch tensor, or as an Iterable
which can be converted to a PyTorch tensor.
The shape of x can be the same with the observaiton shape,
or it can be augmented with an extra leftmost dimension
to express a batch of observations.
mask: Can be given as a 1-dimensional Iterable of booleans ONLY
if `x` represents a batch of observations.
If a `mask` is provided, the i-th observation within the
observation batch `x` will be taken into account only if
the the i-th item of the `mask` is True.
Returns:
The normalized counterpart of the observation(s) expressed by x.
"""
result_as_numpy = not isinstance(x, torch.Tensor)
x = self._verify(x)
self.update(x, mask, verify=False)
result = self.normalize(x, verify=False)
if result_as_numpy:
result = result.cpu().numpy()
return result
def to_layer(self) -> "ObsNormLayer":
"""
Make a PyTorch module which normalizes the its inputs.
Returns:
An ObsNormLayer instance.
"""
mean, stdev = self.stats
low = self.low
high = self.high
return ObsNormLayer(mean=mean, stdev=stdev, low=low, high=high)
def __repr__(self) -> str:
return f"<{self.__class__.__name__}, count: {self.count}>"
def __copy__(self) -> "RunningNorm":
return deepcopy(self)
count: int
property
readonly
¶
Number of observations encountered
device: Union[str, torch.device]
property
readonly
¶
The device in which the observation stats are held
dtype: Union[str, torch.dtype, numpy.dtype, Type]
property
readonly
¶
The dtype of the stored observation stats
high: Optional[float]
property
readonly
¶
The higher (upper) component of the bounds given in the clip tuple.
If clip was initialized as None, this is also None.
low: Optional[float]
property
readonly
¶
The lower component of the bounds given in the clip tuple.
If clip was initialized as None, this is also None.
mean: Tensor
property
readonly
¶
The collected data's mean
min_variance: float
property
readonly
¶
Minimum variance
shape: tuple
property
readonly
¶
Observation shape
stats: CollectedStats
property
readonly
¶
The collected data's mean and standard deviation (stdev) in a tuple
stdev: Tensor
property
readonly
¶
The collected data's standard deviation
sum: Tensor
property
readonly
¶
The collected data's sum
sum_of_squares: Tensor
property
readonly
¶
Sum of squares of the collected data
__init__(self, *, shape, dtype, device=None, min_variance=0.01, clip=None)
special
¶
__init__(...): Initialize the RunningNorm
Parameters:
| Name | Type | Description | Default |
|---|---|---|---|
shape |
Union[tuple, int] |
Observation shape. Can be an integer or a tuple. |
required |
dtype |
Union[str, torch.dtype, numpy.dtype, Type] |
The dtype of the observations. |
required |
device |
Union[str, torch.device] |
The device in which the observation stats are held. If left as None, the device is assumed to be "cpu". |
None |
min_variance |
float |
A lower bound for the variance to be used in
the normalization computations.
In other words, if the computed variance according to the
collected observations ends up lower than |
0.01 |
clip |
Optional[tuple] |
Can be left as None (which is the default), or can be given as a pair of real numbers. This is used for clipping the observations after the normalization operation. In Salimans et al. (2017), (-5.0, +5.0) was used. |
None |
Source code in evotorch/neuroevolution/net/runningnorm.py
def __init__(
self,
*,
shape: Union[tuple, int],
dtype: DType,
device: Optional[Device] = None,
min_variance: float = 1e-2,
clip: Optional[tuple] = None,
) -> None:
"""
`__init__(...)`: Initialize the RunningNorm
Args:
shape: Observation shape. Can be an integer or a tuple.
dtype: The dtype of the observations.
device: The device in which the observation stats are held.
If left as None, the device is assumed to be "cpu".
min_variance: A lower bound for the variance to be used in
the normalization computations.
In other words, if the computed variance according to the
collected observations ends up lower than `min_variance`,
this `min_variance` will be used instead (in an elementwise
manner) while computing the normalized observations.
As in Salimans et al. (2017), the default is 1e-2.
clip: Can be left as None (which is the default), or can be
given as a pair of real numbers.
This is used for clipping the observations after the
normalization operation.
In Salimans et al. (2017), (-5.0, +5.0) was used.
"""
# Make sure that the shape is stored as a torch.Size object.
if isinstance(shape, Iterable):
self._shape = torch.Size(shape)
else:
self._shape = torch.Size([int(shape)])
# Store the number of dimensions
self._ndim = len(self._shape)
# Store the dtype and the device
self._dtype = to_torch_dtype(dtype)
self._device = "cpu" if device is None else device
# Initialize the internally stored data as empty
self._sum: Optional[torch.Tensor] = None
self._sum_of_squares: Optional[torch.Tensor] = None
self._count: int = 0
# Store the minimum variance
self._min_variance = float(min_variance)
if clip is not None:
# If a clip tuple was provided, store the specified lower and upper bounds
lb, ub = clip
self._lb = float(lb)
self._ub = float(ub)
else:
# If a clip tuple was not provided the bounds are stored as None
self._lb = None
self._ub = None
normalize(self, x, *, result_as_numpy=None, verify=True)
¶
Normalize the given observation x.
Parameters:
| Name | Type | Description | Default |
|---|---|---|---|
x |
Iterable |
The observation(s), as a PyTorch tensor, or any Iterable
that is convertable to a PyTorch tensor.
|
required |
result_as_numpy |
Optional[bool] |
Whether or not to return the normalized observation as a numpy array. If left as None (which is the default), then the returned type depends on x: a PyTorch tensor is returned if x is a PyTorch tensor, and a numpy array is returned otherwise. If True, the result is always a numpy array. If False, the result is always a PyTorch tensor. |
None |
verify |
bool |
Whether or not to check the type and dimensions of x.
This is True by default.
Note that, if |
True |
Returns:
| Type | Description |
|---|---|
Iterable |
The normalized observation, as a PyTorch tensor or a numpy array. |
Source code in evotorch/neuroevolution/net/runningnorm.py
@torch.no_grad()
def normalize(self, x: Iterable, *, result_as_numpy: Optional[bool] = None, verify: bool = True) -> Iterable:
"""
Normalize the given observation x.
Args:
x: The observation(s), as a PyTorch tensor, or any Iterable
that is convertable to a PyTorch tensor.
`x` can be a single observation, or it can be a batch
of observations (with an extra leftmost dimension).
result_as_numpy: Whether or not to return the normalized
observation as a numpy array.
If left as None (which is the default), then the returned
type depends on x: a PyTorch tensor is returned if x is a
PyTorch tensor, and a numpy array is returned otherwise.
If True, the result is always a numpy array.
If False, the result is always a PyTorch tensor.
verify: Whether or not to check the type and dimensions of x.
This is True by default.
Note that, if `verify` is False, this function will not
properly check the type of `x` and will assume that `x`
is a PyTorch tensor.
Returns:
The normalized observation, as a PyTorch tensor or a numpy array.
"""
if self._count == 0:
# If this RunningNorm instance has no data yet,
# then we do not know how to do the normalization.
# We therefore raise an error.
raise ValueError("Cannot do normalization because no data is collected yet.")
if verify:
# Here we verify the type and shape of x.
if result_as_numpy is None:
# If there is not an explicit request about the return type,
# we infer the return type from the type of x:
# if x is a tensor, we return a tensor;
# otherwise, we assume x to be a CPU-bound iterable, and
# therefore we return a numpy array.
result_as_numpy = not isinstance(x, torch.Tensor)
else:
result_as_numpy = bool(result_as_numpy)
# We call _verify() to make sure that x is of correct shape
# and is properly converted to a PyTorch tensor.
x = self._verify(x)
# We get the mean and stdev of the collected data
mean, stdev = self.stats
# Now we compute the normalized observation, clipped according to the
# lower and upper bounds expressed by the `clip` tuple, if exists.
result = _clamp((x - mean) / stdev, self._lb, self._ub)
if result_as_numpy:
# If we are to return the result as a numpy array, we do the
# necessary conversion.
result = result.cpu().numpy()
# Finally, return the result
return result
reset(self)
¶
to(self, device)
¶
If the target device is a different device, then make a copy of this RunningNorm instance on the target device. If the target device is the same with this RunningNorm's device, then return this RunningNorm itself.
Parameters:
| Name | Type | Description | Default |
|---|---|---|---|
device |
Union[str, torch.device] |
The target device. |
required |
Returns:
| Type | Description |
|---|---|
RunningNorm |
The RunningNorm on the target device. This can be a copy, or the original RunningNorm instance itself. |
Source code in evotorch/neuroevolution/net/runningnorm.py
def to(self, device: Device) -> "RunningNorm":
"""
If the target device is a different device, then make a copy of this
RunningNorm instance on the target device.
If the target device is the same with this RunningNorm's device, then
return this RunningNorm itself.
Args:
device: The target device.
Returns:
The RunningNorm on the target device. This can be a copy, or the
original RunningNorm instance itself.
"""
if torch.device(device) == torch.device(self.device):
return self
else:
new_running_norm = object.__new__(type(self))
already_handled = {"_sum", "_sum_of_squares", "_device"}
new_running_norm._sum = self._sum.to(device)
new_running_norm._sum_of_squares = self._sum_of_squares.to(device)
new_running_norm._device = device
for k, v in self.__dict__.items():
if k not in already_handled:
setattr(new_running_norm, k, deepcopy(v))
return new_running_norm
to_layer(self)
¶
Make a PyTorch module which normalizes the its inputs.
Returns:
| Type | Description |
|---|---|
ObsNormLayer |
An ObsNormLayer instance. |
Source code in evotorch/neuroevolution/net/runningnorm.py
update(self, x, mask=None, *, verify=True)
¶
Update the stored stats with new observation data.
Parameters:
| Name | Type | Description | Default |
|---|---|---|---|
x |
Union[Iterable, RunningNorm] |
The new observation(s), as a PyTorch tensor, or any Iterable
that can be converted to a PyTorch tensor, or another
RunningNorm instance.
If given as a tensor or as an Iterable, the shape of |
required |
mask |
Optional[Iterable] |
Can be given as a 1-dimensional Iterable of booleans ONLY
if |
None |
verify |
bool |
Whether or not to verify the shape of the given Iterable objects. The default is True. |
True |
Source code in evotorch/neuroevolution/net/runningnorm.py
@torch.no_grad()
def update(self, x: Union[Iterable, "RunningNorm"], mask: Optional[Iterable] = None, *, verify: bool = True):
"""
Update the stored stats with new observation data.
Args:
x: The new observation(s), as a PyTorch tensor, or any Iterable
that can be converted to a PyTorch tensor, or another
RunningNorm instance.
If given as a tensor or as an Iterable, the shape of `x` can
be the same with observation shape, or it can be augmented
with an extra leftmost dimension.
In the case of augmented dimension, `x` is interpreted not as
a single observation, but as a batch of observations.
If `x` is another RunningNorm instance, the stats stored by
this RunningNorm instance will be updated with all the data
stored by `x`.
mask: Can be given as a 1-dimensional Iterable of booleans ONLY
if `x` represents a batch of observations.
If a `mask` is provided, the i-th observation within the
observation batch `x` will be taken into account only if
the i-th item of the `mask` is True.
verify: Whether or not to verify the shape of the given Iterable
objects. The default is True.
"""
if isinstance(x, RunningNorm):
# If we are to update our stats according to another RunningNorm instance
if x._count > 0:
# We bother only if x is non-empty
if mask is not None:
# We were given another RunningNorm, not a batch of observations.
# So, we do not expect to receive a mask tensor.
# If a mask was provided, then this is an unexpected way of calling this function.
# We therefore raise an error.
raise ValueError(
"The `mask` argument is expected as None if the first argument is a RunningNorm."
" However, `mask` is found as something other than None."
)
if self._shape != x._shape:
# If the shapes of this RunningNorm and of the other RunningNorm
# do not match, then we cannot use `x` for updating our stats.
# It might be the case that `x` was initialized for another
# task, with differently sized observations.
# We therefore raise an error.
raise ValueError(
f"The RunningNorm to be updated has the shape {self._shape}"
f" The other RunningNorm has the shape {self._shape}"
f" These shapes are incompatible."
)
if self._has_no_data():
# If this RunningNorm has no data at all, then we clone the
# data of x.
self._sum = self._like_its_own(x._sum.clone())
self._sum_of_squares = self._like_its_own(x._sum_of_squares.clone())
self._count = x._count
elif self._has_data():
# If this RunningNorm has its own data, then we update the
# stored data with the data stored by x.
self._sum += self._like_its_own(x._sum)
self._sum_of_squares += self._like_its_own(x._sum_of_squares)
self._count += x._count
else:
assert False, "RunningNorm is in an invalid state! This might be a bug."
else:
# This is the case where the received argument x is not a
# RunningNorm object, but an Iterable.
if verify:
# If we have the `verify` flag, then we make sure that
# x is a tensor of the correct shape
x = self._verify(x)
if x.ndim == self._ndim:
# If the shape of x is exactly the same with the observation shape
# then we assume that x represents a single observation, and not a
# batch of observations.
if mask is not None:
# Since we are dealing with a single observation,
# we do not expect to receive a mask argument.
# If the mask argument was provided, then this is an unexpected
# usage of this function.
# We therefore raise an error.
raise ValueError(
"The `mask` argument is expected as None if the first argument is a single observation"
" (i.e. not a batch of observations, with an extra leftmost dimension)."
" However, `mask` is found as something other than None."
)
# Since x is a single observation,
# the sum of observations extracted from x is x itself,
# and the sum of squared observations extracted from x is
# the square of x itself.
sum_of_x = x
sum_of_x_squared = x.square()
# We extracted a single observation from x
n = 1
elif x.ndim == (self._ndim + 1):
# If the number of dimensions of x is one more than the number
# of dimensions of this RunningNorm, then we assume that x is a batch
# of observations.
if mask is not None:
# If a mask is provided, then we first make sure that it is a tensor
# of dtype bool in the correct device.
mask = torch.as_tensor(mask, dtype=torch.bool, device=self._device)
if mask.ndim != 1:
# We expect the mask to be 1-dimensional.
# If not, we raise an error.
raise ValueError(
f"The `mask` tensor was expected as a 1-dimensional tensor."
f" However, its shape is {mask.shape}."
)
if len(mask) != x.shape[0]:
# If the length of the mask is not the batch size of x,
# then there is a mismatch.
# We therefore raise an error.
raise ValueError(
f"The shape of the given tensor is {x.shape}."
f" Therefore, the batch size of observations is {x.shape[0]}."
f" However, the given `mask` tensor does not has an incompatible length: {len(mask)}."
)
# We compute how many True items we have in the mask.
# This integer gives us how many observations we extract from x.
n = int(torch.sum(torch.as_tensor(mask, dtype=torch.int64, device=self._device)))
# We now re-cast the mask as the observation dtype (so that True items turn to 1.0
# and False items turn to 0.0), and then increase its number of dimensions so that
# it can operate directly with x.
mask = self._like_its_own(mask).reshape(torch.Size([x.shape[0]] + ([1] * (x.ndim - 1))))
# Finally, we multiply x with the mask. This means that the observations with corresponding
# mask values as False are zeroed out.
x = x * mask
else:
# This is the case where we did not receive a mask.
# We can simply say that the number of observations to extract from x
# is the size of its leftmost dimension, i.e. the batch size.
n = x.shape[0]
# With or without a mask, we are now ready to extract the sum and sum of squares
# from x.
sum_of_x = torch.sum(x, dim=0)
sum_of_x_squared = torch.sum(x.square(), dim=0)
else:
# This is the case where the number of dimensions of x is unrecognized.
# This case is actually already checked by the _verify(...) method earlier.
# This defensive fallback case is only for when verify=False and it turned out
# that the ndim is invalid.
raise ValueError(f"Invalid shape: {x.shape}")
# At this point, we handled all the valid cases regarding the Iterable x,
# and we have our sum_of_x (sum of all observations), sum_of_squares
# (sum of all squared observations), and n (number of observations extracted
# from x).
if self._has_no_data():
# If our RunningNorm is empty, the observation data we extracted from x
# become our RunningNorm's new data.
self._sum = sum_of_x
self._sum_of_squares = sum_of_x_squared
self._count = n
elif self._has_data():
# If our RunningNorm is not empty, the stored data is updated with the
# data extracted from x.
self._sum += sum_of_x
self._sum_of_squares += sum_of_x_squared
self._count += n
else:
# This is an erroneous state where the internal data looks neither
# existent nor completely empty.
# This might be the result of a bug, or maybe this instance's
# protected variables were tempered with from the outside.
assert False, "RunningNorm is in an invalid state! This might be a bug."
update_and_normalize(self, x, mask=None)
¶
Update the observation stats according to x, then normalize x.
Parameters:
| Name | Type | Description | Default |
|---|---|---|---|
x |
Iterable |
The observation(s), as a PyTorch tensor, or as an Iterable which can be converted to a PyTorch tensor. The shape of x can be the same with the observaiton shape, or it can be augmented with an extra leftmost dimension to express a batch of observations. |
required |
mask |
Optional[Iterable] |
Can be given as a 1-dimensional Iterable of booleans ONLY
if |
None |
Returns:
| Type | Description |
|---|---|
Iterable |
The normalized counterpart of the observation(s) expressed by x. |
Source code in evotorch/neuroevolution/net/runningnorm.py
@torch.no_grad()
def update_and_normalize(self, x: Iterable, mask: Optional[Iterable] = None) -> Iterable:
"""
Update the observation stats according to x, then normalize x.
Args:
x: The observation(s), as a PyTorch tensor, or as an Iterable
which can be converted to a PyTorch tensor.
The shape of x can be the same with the observaiton shape,
or it can be augmented with an extra leftmost dimension
to express a batch of observations.
mask: Can be given as a 1-dimensional Iterable of booleans ONLY
if `x` represents a batch of observations.
If a `mask` is provided, the i-th observation within the
observation batch `x` will be taken into account only if
the the i-th item of the `mask` is True.
Returns:
The normalized counterpart of the observation(s) expressed by x.
"""
result_as_numpy = not isinstance(x, torch.Tensor)
x = self._verify(x)
self.update(x, mask, verify=False)
result = self.normalize(x, verify=False)
if result_as_numpy:
result = result.cpu().numpy()
return result
runningstat
¶
RunningStat
¶
Tool for efficiently computing the mean and stdev of arrays. The arrays themselves are not stored separately, instead, they are accumulated.
This RunningStat is implemented as a wrapper around RunningNorm. The difference is that the interface of RunningStat is simplified to expect only numpy arrays, and expect only non-vectorized observations. With this simplified interface, RunningStat is meant to be used by GymNE, on classical non-vectorized gym tasks.
Source code in evotorch/neuroevolution/net/runningstat.py
class RunningStat:
"""
Tool for efficiently computing the mean and stdev of arrays.
The arrays themselves are not stored separately,
instead, they are accumulated.
This RunningStat is implemented as a wrapper around RunningNorm.
The difference is that the interface of RunningStat is simplified
to expect only numpy arrays, and expect only non-vectorized
observations.
With this simplified interface, RunningStat is meant to be used
by GymNE, on classical non-vectorized gym tasks.
"""
def __init__(self):
"""
`__init__(...)`: Initialize the RunningStat.
"""
self._rn: Optional[RunningNorm] = None
self.reset()
def reset(self):
"""
Reset the RunningStat to its initial state.
"""
self._rn = None
@property
def count(self) -> int:
"""
Get the number of arrays accumulated.
"""
if self._rn is None:
return 0
else:
return self._rn.count
@property
def sum(self) -> np.ndarray:
"""
Get the sum of all accumulated arrays.
"""
return self._rn.sum.numpy()
@property
def sum_of_squares(self) -> np.ndarray:
"""
Get the sum of squares of all accumulated arrays.
"""
return self._rn.sum_of_squares.numpy()
@property
def mean(self) -> np.ndarray:
"""
Get the mean of all accumulated arrays.
"""
return self._rn.mean.numpy()
@property
def stdev(self) -> np.ndarray:
"""
Get the standard deviation of all accumulated arrays.
"""
return self._rn.stdev.numpy()
def update(self, x: Union[np.ndarray, "RunningStat"]):
"""
Accumulate more data into the RunningStat object.
If the argument is an array, that array is added
as one more data element.
If the argument is another RunningStat instance,
all the stats accumulated by that RunningStat object
are added into this RunningStat object.
"""
if isinstance(x, RunningStat):
if x.count > 0:
if self._rn is None:
self._rn = deepcopy(x._rn)
else:
self._rn.update(x._rn)
else:
if self._rn is None:
x = np.array(x, dtype="float32")
self._rn = RunningNorm(shape=x.shape, dtype="float32", device="cpu")
self._rn.update(x)
def normalize(self, x: Union[np.ndarray, list]) -> np.ndarray:
"""
Normalize the array x according to the accumulated stats.
"""
if self._rn is None:
return x
else:
x = np.array(x, dtype="float32")
return self._rn.normalize(x)
def __copy__(self):
return deepcopy(self)
def __repr__(self) -> str:
return f"<{self.__class__.__name__}, count: {self.count}>"
def to(self, device: Union[str, torch.device]) -> "RunningStat":
"""
If the target device is cpu, return this RunningStat instance itself.
A RunningStat object is meant to work with numpy arrays. Therefore,
any device other than the cpu will trigger an error.
Args:
device: The target device. Only cpu is supported.
Returns:
The original RunningStat.
"""
if torch.device(device) == torch.device("cpu"):
return self
else:
raise ValueError(
f"The received target device is {repr(device)}. However, RunningStat can only work on a cpu."
)
def to_layer(self) -> nn.Module:
"""
Make a PyTorch module which normalizes the its inputs.
Returns:
An ObsNormLayer instance.
"""
return self._rn.to_layer()
count: int
property
readonly
¶
Get the number of arrays accumulated.
mean: ndarray
property
readonly
¶
Get the mean of all accumulated arrays.
stdev: ndarray
property
readonly
¶
Get the standard deviation of all accumulated arrays.
sum: ndarray
property
readonly
¶
Get the sum of all accumulated arrays.
sum_of_squares: ndarray
property
readonly
¶
Get the sum of squares of all accumulated arrays.
__init__(self)
special
¶
normalize(self, x)
¶
Normalize the array x according to the accumulated stats.
reset(self)
¶
to(self, device)
¶
If the target device is cpu, return this RunningStat instance itself. A RunningStat object is meant to work with numpy arrays. Therefore, any device other than the cpu will trigger an error.
Parameters:
| Name | Type | Description | Default |
|---|---|---|---|
device |
Union[str, torch.device] |
The target device. Only cpu is supported. |
required |
Returns:
| Type | Description |
|---|---|
RunningStat |
The original RunningStat. |
Source code in evotorch/neuroevolution/net/runningstat.py
def to(self, device: Union[str, torch.device]) -> "RunningStat":
"""
If the target device is cpu, return this RunningStat instance itself.
A RunningStat object is meant to work with numpy arrays. Therefore,
any device other than the cpu will trigger an error.
Args:
device: The target device. Only cpu is supported.
Returns:
The original RunningStat.
"""
if torch.device(device) == torch.device("cpu"):
return self
else:
raise ValueError(
f"The received target device is {repr(device)}. However, RunningStat can only work on a cpu."
)
to_layer(self)
¶
Make a PyTorch module which normalizes the its inputs.
Returns:
| Type | Description |
|---|---|
Module |
An ObsNormLayer instance. |
update(self, x)
¶
Accumulate more data into the RunningStat object. If the argument is an array, that array is added as one more data element. If the argument is another RunningStat instance, all the stats accumulated by that RunningStat object are added into this RunningStat object.
Source code in evotorch/neuroevolution/net/runningstat.py
def update(self, x: Union[np.ndarray, "RunningStat"]):
"""
Accumulate more data into the RunningStat object.
If the argument is an array, that array is added
as one more data element.
If the argument is another RunningStat instance,
all the stats accumulated by that RunningStat object
are added into this RunningStat object.
"""
if isinstance(x, RunningStat):
if x.count > 0:
if self._rn is None:
self._rn = deepcopy(x._rn)
else:
self._rn.update(x._rn)
else:
if self._rn is None:
x = np.array(x, dtype="float32")
self._rn = RunningNorm(shape=x.shape, dtype="float32", device="cpu")
self._rn.update(x)
statefulmodule
¶
StatefulModule (Module)
¶
A wrapper that provides a stateful interface for recurrent torch modules.
If the torch module to be wrapped is non-recurrent and its forward method has a single input (the input tensor) and a single output (the output tensor), then this wrapper module acts as a no-op wrapper.
If the torch module to be wrapped is recurrent and its forward method has
two inputs (the input tensor and an optional second argument for the hidden
state) and two outputs (the output tensor and the new hidden state), then
this wrapper brings a new forward-passing interface. In this new interface,
the forward method has a single input (the input tensor) and a single
output (the output tensor). The hidden states, instead of being
explicitly requested via a second argument and returned as a second
result, are stored and used by the wrapper.
When a new series of inputs is to be used, one has to call the reset()
method of this wrapper.
Source code in evotorch/neuroevolution/net/statefulmodule.py
class StatefulModule(nn.Module):
"""
A wrapper that provides a stateful interface for recurrent torch modules.
If the torch module to be wrapped is non-recurrent and its forward method
has a single input (the input tensor) and a single output (the output
tensor), then this wrapper module acts as a no-op wrapper.
If the torch module to be wrapped is recurrent and its forward method has
two inputs (the input tensor and an optional second argument for the hidden
state) and two outputs (the output tensor and the new hidden state), then
this wrapper brings a new forward-passing interface. In this new interface,
the forward method has a single input (the input tensor) and a single
output (the output tensor). The hidden states, instead of being
explicitly requested via a second argument and returned as a second
result, are stored and used by the wrapper.
When a new series of inputs is to be used, one has to call the `reset()`
method of this wrapper.
"""
def __init__(self, wrapped_module: nn.Module):
"""
`__init__(...)`: Initialize the StatefulModule.
Args:
wrapped_module: The `torch.nn.Module` instance to wrap.
"""
super().__init__()
# Declare the variable that will store the hidden state of wrapped_module, if any.
self._hidden: Any = None
# Store the module that is wrapped.
self.wrapped_module = wrapped_module
def forward(self, x: torch.Tensor) -> torch.Tensor:
if self._hidden is None:
# If there is no stored hidden state, then only pass the input tensor to the wrapped module.
out = self.wrapped_module(x)
else:
# If there is a hidden state saved from the previous call to this `forward(...)` method, then pass the
# input tensor and this stored hidden state.
out = self.wrapped_module(x, self._hidden)
if isinstance(out, tuple):
# If the result of the wrapped module is a tuple, then we assume that the wrapped module returned an
# output tensor and a hidden state. We assume the first element of this tuple as the output tensor,
# and the second element as the new hidden state.
# We set the variable y to the output tensor, and we store the new hidden state via the attribute
# `_hidden`.
y, self._hidden = out
else:
# If the result of the wrapped module is not a tuple, then we assume that the wrapped module returned
# only the output tensor. We set the variable y to the output tensor, and set the attribute `_hidden`
# as None to indicate that there was no hidden state received.
y = out
self._hidden = None
# We return y, which stores the output received by the wrapped module.
return y
def reset(self):
"""
Reset the hidden state, if any.
"""
self._hidden = None
__init__(self, wrapped_module)
special
¶
__init__(...): Initialize the StatefulModule.
Parameters:
| Name | Type | Description | Default |
|---|---|---|---|
wrapped_module |
Module |
The |
required |
Source code in evotorch/neuroevolution/net/statefulmodule.py
def __init__(self, wrapped_module: nn.Module):
"""
`__init__(...)`: Initialize the StatefulModule.
Args:
wrapped_module: The `torch.nn.Module` instance to wrap.
"""
super().__init__()
# Declare the variable that will store the hidden state of wrapped_module, if any.
self._hidden: Any = None
# Store the module that is wrapped.
self.wrapped_module = wrapped_module
forward(self, x)
¶
Defines the computation performed at every call.
Should be overridden by all subclasses.
.. note::
Although the recipe for forward pass needs to be defined within
this function, one should call the :class:Module instance afterwards
instead of this since the former takes care of running the
registered hooks while the latter silently ignores them.
Source code in evotorch/neuroevolution/net/statefulmodule.py
def forward(self, x: torch.Tensor) -> torch.Tensor:
if self._hidden is None:
# If there is no stored hidden state, then only pass the input tensor to the wrapped module.
out = self.wrapped_module(x)
else:
# If there is a hidden state saved from the previous call to this `forward(...)` method, then pass the
# input tensor and this stored hidden state.
out = self.wrapped_module(x, self._hidden)
if isinstance(out, tuple):
# If the result of the wrapped module is a tuple, then we assume that the wrapped module returned an
# output tensor and a hidden state. We assume the first element of this tuple as the output tensor,
# and the second element as the new hidden state.
# We set the variable y to the output tensor, and we store the new hidden state via the attribute
# `_hidden`.
y, self._hidden = out
else:
# If the result of the wrapped module is not a tuple, then we assume that the wrapped module returned
# only the output tensor. We set the variable y to the output tensor, and set the attribute `_hidden`
# as None to indicate that there was no hidden state received.
y = out
self._hidden = None
# We return y, which stores the output received by the wrapped module.
return y
reset(self)
¶
ensure_stateful(net)
¶
Ensure that a module is wrapped by StatefulModule.
If the given module is already wrapped by StatefulModule, then the module itself is returned. If the given module is not wrapped by StatefulModule, then this function first wraps the module via a new StatefulModule instance, and then this new wrapper is returned.
Parameters:
| Name | Type | Description | Default |
|---|---|---|---|
net |
Module |
The |
required |
Returns:
| Type | Description |
|---|---|
The module |
Source code in evotorch/neuroevolution/net/statefulmodule.py
def ensure_stateful(net: nn.Module):
"""
Ensure that a module is wrapped by StatefulModule.
If the given module is already wrapped by StatefulModule, then the
module itself is returned.
If the given module is not wrapped by StatefulModule, then this function
first wraps the module via a new StatefulModule instance, and then this
new wrapper is returned.
Args:
net: The `torch.nn.Module` to be wrapped by StatefulModule (if it is
not already wrapped by it).
Returns:
The module `net`, wrapped by StatefulModule.
"""
if not isinstance(net, StatefulModule):
return StatefulModule(net)
vecrl
¶
This namespace provides various vectorized reinforcement learning utilities.
Policy
¶
A Policy for deciding the actions for a reinforcement learning environment.
This can be seen as a stateful wrapper around a PyTorch module.
Let us assume that we have the following PyTorch module:
which has 48 parameters (when all the parameters are flattened).
Let us randomly generate a parameter vector for our module net:
We can now prepare a policy:
If we generate a random observation:
We can receive our action as follows:
If the PyTorch module that we wish to wrap is a recurrent network (i.e. a network which expects an optional second argument for the hidden state, and returns a second value which represents the updated hidden state), then, the hidden state is automatically managed by the Policy instance.
Let us assume that we have a recurrent network named recnet.
In this case, because the hidden state of the network is internally managed, the usage is still the same with our previous non-recurrent
Examples:
When using a recurrent module on multiple episodes, it is important to reset the hidden state of the network. This is achieved by the reset method:
policy.reset()
action1 = policy(observation1)
# action2 will be computed with the hidden state generated by the
# previous forward-pass.
action2 = policy(observation2)
policy.reset()
# action3 will be computed according to the renewed hidden state.
action3 = policy(observation3)
Both for non-recurrent and recurrent networks, it is possible to perform vectorized operations. For now, let us return to our first non-recurrent example:
Instead of generating only one parameter vector, we now generate a batch of parameter vectors. Let us say that our batch size is 10:
Like we did in the non-batched examples, we can do:
Because we are now in the batched mode, policy now expects a batch
of observations and will return a batch of actions:
When doing vectorized reinforcement learning with a recurrent module,
it can be the case that only some of the environments are finished,
and therefore it is necessary to reset the hidden states associated
with those environments only. The reset(...) method of Policy
has a second argument to specify which of the recurrent network
instances are to be reset. For example, if the episodes of the
environments with indices 2 and 5 are about to restart (and therefore
we wish to reset the states of the networks with indices 2 and 5),
then, we can do:
Source code in evotorch/neuroevolution/net/vecrl.py
class Policy:
"""
A Policy for deciding the actions for a reinforcement learning environment.
This can be seen as a stateful wrapper around a PyTorch module.
Let us assume that we have the following PyTorch module:
```python
from torch import nn
net = nn.Linear(5, 8)
```
which has 48 parameters (when all the parameters are flattened).
Let us randomly generate a parameter vector for our module `net`:
```python
parameters = torch.randn(48)
```
We can now prepare a policy:
```python
policy = Policy(net)
policy.set_parameters(parameters)
```
If we generate a random observation:
```python
observation = torch.randn(5)
```
We can receive our action as follows:
```python
action = policy(observation)
```
If the PyTorch module that we wish to wrap is a recurrent network (i.e.
a network which expects an optional second argument for the hidden state,
and returns a second value which represents the updated hidden state),
then, the hidden state is automatically managed by the Policy instance.
Let us assume that we have a recurrent network named `recnet`.
```python
policy = Policy(recnet)
policy.set_parameters(parameters_of_recnet)
```
In this case, because the hidden state of the network is internally
managed, the usage is still the same with our previous non-recurrent
example:
```python
action = policy(observation)
```
When using a recurrent module on multiple episodes, it is important
to reset the hidden state of the network. This is achieved by the
reset method:
```python
policy.reset()
action1 = policy(observation1)
# action2 will be computed with the hidden state generated by the
# previous forward-pass.
action2 = policy(observation2)
policy.reset()
# action3 will be computed according to the renewed hidden state.
action3 = policy(observation3)
```
Both for non-recurrent and recurrent networks, it is possible to
perform vectorized operations. For now, let us return to our
first non-recurrent example:
```python
net = nn.Linear(5, 8)
```
Instead of generating only one parameter vector, we now generate
a batch of parameter vectors. Let us say that our batch size is 10:
```python
batch_of_parameters = torch.randn(10, 48)
```
Like we did in the non-batched examples, we can do:
```python
policy = Policy(net)
policy.set_parameters(batch_of_parameters)
```
Because we are now in the batched mode, `policy` now expects a batch
of observations and will return a batch of actions:
```python
batch_of_observations = torch.randn(10, 5)
batch_of_actions = policy(batch_of_observations)
```
When doing vectorized reinforcement learning with a recurrent module,
it can be the case that only some of the environments are finished,
and therefore it is necessary to reset the hidden states associated
with those environments only. The `reset(...)` method of Policy
has a second argument to specify which of the recurrent network
instances are to be reset. For example, if the episodes of the
environments with indices 2 and 5 are about to restart (and therefore
we wish to reset the states of the networks with indices 2 and 5),
then, we can do:
```python
policy.reset(torch.tensor([2, 5]))
```
"""
def __init__(self, net: Union[str, Callable, nn.Module], **kwargs):
"""
`__init__(...)`: Initialize the Policy.
Args:
net: The network to be wrapped by the Policy object.
This can be a string, a Callable (e.g. a `torch.nn.Module`
subclass), or a `torch.nn.Module` instance.
When this argument is a string, the network will be
created with the help of the function
`evotorch.neuroevolution.net.str_to_net(...)` and then
wrapped. Please see the `str_to_net(...)` function's
documentation for details regarding how a network structure
can be expressed via strings.
kwargs: Expected in the form of additional keyword arguments,
these keyword arguments will be passed to the provided
Callable object (if the argument `net` is a Callable)
or to `str_to_net(...)` (if the argument `net` is a string)
at the moment of generating the network.
If the argument `net` is a `torch.nn.Module` instance,
having any additional keyword arguments will trigger an
error, because the network is already instantiated and
therefore, it is not possible to pass these keyword arguments.
"""
from ..net import str_to_net
from ..net.functional import ModuleExpectingFlatParameters, make_functional_module
if isinstance(net, str):
self.__module = str_to_net(net, **kwargs)
elif isinstance(net, nn.Module):
if len(kwargs) > 0:
raise ValueError(
f"When the network is given as an `nn.Module` instance, extra network arguments cannot be used"
f" (because the network is already instantiated)."
f" However, these extra keyword arguments were received: {kwargs}."
)
self.__module = net
elif isinstance(net, Callable):
self.__module = net(**kwargs)
else:
raise TypeError(
f"The class `Policy` expected a string or an `nn.Module` instance, or a Callable, but received {net}"
f" (whose type is {type(net)})."
)
self.__fmodule: ModuleExpectingFlatParameters = make_functional_module(self.__module)
self.__state: Any = None
self.__parameters: Optional[torch.Tensor] = None
def set_parameters(self, parameters: torch.Tensor, indices: Optional[MaskOrIndices] = None, *, reset: bool = True):
"""
Set the parameters of the policy.
Args:
parameters: A 1-dimensional or a 2-dimensional tensor containing
the flattened parameters to be used with the neural network.
If the given parameters are two-dimensional, then, given that
the leftmost size of the parameter tensor is `n`, the
observations will be expected in a batch with leftmost size
`n`, and the returned actions will also be in a batch,
again with the leftmost size `n`.
indices: For when the parameters were previously given via a
2-dimensional tensor, provide this argument if you would like
to change only some rows of the previously given parameters.
For example, if `indices` is given as `torch.tensor([2, 4])`
and the argument `parameters` is given as a 2-dimensional
tensor with leftmost size 2, then the rows with indices
2 and 4 will be replaced by these new parameters provided
via the argument `parameters`.
reset: If given as True, the hidden states of the networks whose
parameters just changed will be reset. If `indices` was not
provided at all, then this means that the parameters of all
networks are modified, in which case, all the hidden states
will be reset.
If given as False, no such resetting will be done.
"""
if self.__parameters is None:
if indices is not None:
raise ValueError(
"The argument `indices` can be used only if network parameters were previously specified."
" However, it seems that the method `set_parameters(...)` was not called before."
)
self.__parameters = parameters
else:
if indices is None:
self.__parameters = parameters
else:
self.__parameters[indices] = parameters
if reset:
self.reset(indices)
def __call__(self, x: torch.Tensor) -> torch.Tensor:
"""
Pass the given observations through the network.
Args:
x: The observations, as a PyTorch tensor.
If the parameters were given (via the method
`set_parameters(...)`) as a 1-dimensional tensor, then this
argument is expected to store a single observation.
If the parameters were given as a 2-dimensional tensor,
then, this argument is expected to store a batch of
observations, and the leftmost size of this observation
tensor must match with the leftmost size of the parameter
tensor.
Returns:
The output tensor, which represents the action to take.
"""
if self.__parameters is None:
raise ValueError("Please use the method `set_parameters(...)` before calling the policy.")
if self.__state is None:
further_args = (x,)
else:
further_args = (x, self.__state)
parameters = self.__parameters
ndim = parameters.ndim
if ndim == 1:
result = self.__fmodule(parameters, *further_args)
elif ndim == 2:
vmapped = vmap(self.__fmodule)
result = vmapped(parameters, *further_args)
else:
raise ValueError(
f"Expected the parameters as a 1 or 2 dimensional tensor."
f" However, the received parameters tensor has {ndim} dimensions."
)
if isinstance(result, torch.Tensor):
return result
elif isinstance(result, tuple):
result, state = result
self.__state = state
return result
else:
raise TypeError(f"The torch module used by the Policy returned an unexpected object: {result}")
def reset(self, indices: Optional[MaskOrIndices] = None, *, copy: bool = True):
"""
Reset the hidden states, if the contained module is a recurrent network.
Args:
indices: Optionally a sequence of integers or a sequence of
booleans, specifying which networks' states will be
reset. If left as None, then the states of all the networks
will be reset.
copy: When `indices` is given as something other than None,
if `copy` is given as True, then the resetting will NOT
be done in-place. Instead, a new copy of the hidden state
will first be created, and then the specified regions
of this new copy will be cleared, and then finally this
modified copy will be declared as the new hidden state.
It is a common practice for recurrent neural network
implementations to return the same tensor both as its
output and as (part of) its hidden state. With `copy=False`,
the resetting would be done in-place, and the action
tensor could be involuntarily reset as well.
This in-place modification could cause silent bugs
if the unintended modification on the action tensor
happens BEFORE the action is sent to the reinforcement
learning environment.
To prevent such situations, the default value for the argument
`copy` is True.
"""
if indices is None:
self.__state = None
else:
if self.__state is not None:
with torch.no_grad():
if copy:
self.__state = deepcopy(self.__state)
reset_tensors(self.__state, indices)
@property
def parameters(self) -> torch.Tensor:
"""
The currently used parameters.
"""
return self.__parameters
@property
def h(self) -> Optional[torch.Tensor]:
"""
The hidden state of the contained recurrent network, if any.
If the contained recurrent network did not generate a hidden state
yet, or if the contained network is not recurrent, then the result
will be None.
"""
return self.__state
@property
def parameter_length(self) -> int:
"""
Length of the parameter tensor.
"""
return self.__fmodule.parameter_length
@property
def wrapped_module(self) -> nn.Module:
"""
The wrapped `torch.nn.Module` instance.
"""
return self.__module
def to_torch_module(self, parameter_vector: torch.Tensor) -> nn.Module:
"""
Get a copy of the contained network, parameterized as specified.
Args:
parameter_vector: The parameters to be used by the new network.
Returns:
Copy of the contained network, as a `torch.nn.Module` instance.
"""
with torch.no_grad():
net = deepcopy(self.__module).to(parameter_vector.device)
nnu.vector_to_parameters(parameter_vector, net.parameters())
return net
h: Optional[torch.Tensor]
property
readonly
¶
The hidden state of the contained recurrent network, if any.
If the contained recurrent network did not generate a hidden state yet, or if the contained network is not recurrent, then the result will be None.
parameter_length: int
property
readonly
¶
Length of the parameter tensor.
parameters: Tensor
property
readonly
¶
The currently used parameters.
wrapped_module: Module
property
readonly
¶
The wrapped torch.nn.Module instance.
__call__(self, x)
special
¶
Pass the given observations through the network.
Parameters:
| Name | Type | Description | Default |
|---|---|---|---|
x |
Tensor |
The observations, as a PyTorch tensor.
If the parameters were given (via the method
|
required |
Returns:
| Type | Description |
|---|---|
Tensor |
The output tensor, which represents the action to take. |
Source code in evotorch/neuroevolution/net/vecrl.py
def __call__(self, x: torch.Tensor) -> torch.Tensor:
"""
Pass the given observations through the network.
Args:
x: The observations, as a PyTorch tensor.
If the parameters were given (via the method
`set_parameters(...)`) as a 1-dimensional tensor, then this
argument is expected to store a single observation.
If the parameters were given as a 2-dimensional tensor,
then, this argument is expected to store a batch of
observations, and the leftmost size of this observation
tensor must match with the leftmost size of the parameter
tensor.
Returns:
The output tensor, which represents the action to take.
"""
if self.__parameters is None:
raise ValueError("Please use the method `set_parameters(...)` before calling the policy.")
if self.__state is None:
further_args = (x,)
else:
further_args = (x, self.__state)
parameters = self.__parameters
ndim = parameters.ndim
if ndim == 1:
result = self.__fmodule(parameters, *further_args)
elif ndim == 2:
vmapped = vmap(self.__fmodule)
result = vmapped(parameters, *further_args)
else:
raise ValueError(
f"Expected the parameters as a 1 or 2 dimensional tensor."
f" However, the received parameters tensor has {ndim} dimensions."
)
if isinstance(result, torch.Tensor):
return result
elif isinstance(result, tuple):
result, state = result
self.__state = state
return result
else:
raise TypeError(f"The torch module used by the Policy returned an unexpected object: {result}")
__init__(self, net, **kwargs)
special
¶
__init__(...): Initialize the Policy.
Parameters:
| Name | Type | Description | Default |
|---|---|---|---|
net |
Union[str, Callable, torch.nn.modules.module.Module] |
The network to be wrapped by the Policy object.
This can be a string, a Callable (e.g. a |
required |
kwargs |
Expected in the form of additional keyword arguments,
these keyword arguments will be passed to the provided
Callable object (if the argument |
{} |
Source code in evotorch/neuroevolution/net/vecrl.py
def __init__(self, net: Union[str, Callable, nn.Module], **kwargs):
"""
`__init__(...)`: Initialize the Policy.
Args:
net: The network to be wrapped by the Policy object.
This can be a string, a Callable (e.g. a `torch.nn.Module`
subclass), or a `torch.nn.Module` instance.
When this argument is a string, the network will be
created with the help of the function
`evotorch.neuroevolution.net.str_to_net(...)` and then
wrapped. Please see the `str_to_net(...)` function's
documentation for details regarding how a network structure
can be expressed via strings.
kwargs: Expected in the form of additional keyword arguments,
these keyword arguments will be passed to the provided
Callable object (if the argument `net` is a Callable)
or to `str_to_net(...)` (if the argument `net` is a string)
at the moment of generating the network.
If the argument `net` is a `torch.nn.Module` instance,
having any additional keyword arguments will trigger an
error, because the network is already instantiated and
therefore, it is not possible to pass these keyword arguments.
"""
from ..net import str_to_net
from ..net.functional import ModuleExpectingFlatParameters, make_functional_module
if isinstance(net, str):
self.__module = str_to_net(net, **kwargs)
elif isinstance(net, nn.Module):
if len(kwargs) > 0:
raise ValueError(
f"When the network is given as an `nn.Module` instance, extra network arguments cannot be used"
f" (because the network is already instantiated)."
f" However, these extra keyword arguments were received: {kwargs}."
)
self.__module = net
elif isinstance(net, Callable):
self.__module = net(**kwargs)
else:
raise TypeError(
f"The class `Policy` expected a string or an `nn.Module` instance, or a Callable, but received {net}"
f" (whose type is {type(net)})."
)
self.__fmodule: ModuleExpectingFlatParameters = make_functional_module(self.__module)
self.__state: Any = None
self.__parameters: Optional[torch.Tensor] = None
reset(self, indices=None, *, copy=True)
¶
Reset the hidden states, if the contained module is a recurrent network.
Parameters:
| Name | Type | Description | Default |
|---|---|---|---|
indices |
Union[int, Iterable] |
Optionally a sequence of integers or a sequence of booleans, specifying which networks' states will be reset. If left as None, then the states of all the networks will be reset. |
None |
copy |
bool |
When |
True |
Source code in evotorch/neuroevolution/net/vecrl.py
def reset(self, indices: Optional[MaskOrIndices] = None, *, copy: bool = True):
"""
Reset the hidden states, if the contained module is a recurrent network.
Args:
indices: Optionally a sequence of integers or a sequence of
booleans, specifying which networks' states will be
reset. If left as None, then the states of all the networks
will be reset.
copy: When `indices` is given as something other than None,
if `copy` is given as True, then the resetting will NOT
be done in-place. Instead, a new copy of the hidden state
will first be created, and then the specified regions
of this new copy will be cleared, and then finally this
modified copy will be declared as the new hidden state.
It is a common practice for recurrent neural network
implementations to return the same tensor both as its
output and as (part of) its hidden state. With `copy=False`,
the resetting would be done in-place, and the action
tensor could be involuntarily reset as well.
This in-place modification could cause silent bugs
if the unintended modification on the action tensor
happens BEFORE the action is sent to the reinforcement
learning environment.
To prevent such situations, the default value for the argument
`copy` is True.
"""
if indices is None:
self.__state = None
else:
if self.__state is not None:
with torch.no_grad():
if copy:
self.__state = deepcopy(self.__state)
reset_tensors(self.__state, indices)
set_parameters(self, parameters, indices=None, *, reset=True)
¶
Set the parameters of the policy.
Parameters:
| Name | Type | Description | Default |
|---|---|---|---|
parameters |
Tensor |
A 1-dimensional or a 2-dimensional tensor containing
the flattened parameters to be used with the neural network.
If the given parameters are two-dimensional, then, given that
the leftmost size of the parameter tensor is |
required |
indices |
Union[int, Iterable] |
For when the parameters were previously given via a
2-dimensional tensor, provide this argument if you would like
to change only some rows of the previously given parameters.
For example, if |
None |
reset |
bool |
If given as True, the hidden states of the networks whose
parameters just changed will be reset. If |
True |
Source code in evotorch/neuroevolution/net/vecrl.py
def set_parameters(self, parameters: torch.Tensor, indices: Optional[MaskOrIndices] = None, *, reset: bool = True):
"""
Set the parameters of the policy.
Args:
parameters: A 1-dimensional or a 2-dimensional tensor containing
the flattened parameters to be used with the neural network.
If the given parameters are two-dimensional, then, given that
the leftmost size of the parameter tensor is `n`, the
observations will be expected in a batch with leftmost size
`n`, and the returned actions will also be in a batch,
again with the leftmost size `n`.
indices: For when the parameters were previously given via a
2-dimensional tensor, provide this argument if you would like
to change only some rows of the previously given parameters.
For example, if `indices` is given as `torch.tensor([2, 4])`
and the argument `parameters` is given as a 2-dimensional
tensor with leftmost size 2, then the rows with indices
2 and 4 will be replaced by these new parameters provided
via the argument `parameters`.
reset: If given as True, the hidden states of the networks whose
parameters just changed will be reset. If `indices` was not
provided at all, then this means that the parameters of all
networks are modified, in which case, all the hidden states
will be reset.
If given as False, no such resetting will be done.
"""
if self.__parameters is None:
if indices is not None:
raise ValueError(
"The argument `indices` can be used only if network parameters were previously specified."
" However, it seems that the method `set_parameters(...)` was not called before."
)
self.__parameters = parameters
else:
if indices is None:
self.__parameters = parameters
else:
self.__parameters[indices] = parameters
if reset:
self.reset(indices)
to_torch_module(self, parameter_vector)
¶
Get a copy of the contained network, parameterized as specified.
Parameters:
| Name | Type | Description | Default |
|---|---|---|---|
parameter_vector |
Tensor |
The parameters to be used by the new network. |
required |
Returns:
| Type | Description |
|---|---|
Module |
Copy of the contained network, as a |
Source code in evotorch/neuroevolution/net/vecrl.py
def to_torch_module(self, parameter_vector: torch.Tensor) -> nn.Module:
"""
Get a copy of the contained network, parameterized as specified.
Args:
parameter_vector: The parameters to be used by the new network.
Returns:
Copy of the contained network, as a `torch.nn.Module` instance.
"""
with torch.no_grad():
net = deepcopy(self.__module).to(parameter_vector.device)
nnu.vector_to_parameters(parameter_vector, net.parameters())
return net
TorchWrapper (Wrapper)
¶
A gym wrapper which ensures that the actions, observations, rewards, and the 'done' values are expressed as PyTorch tensors.
Source code in evotorch/neuroevolution/net/vecrl.py
class TorchWrapper(gym.Wrapper):
"""
A gym wrapper which ensures that the actions, observations, rewards, and
the 'done' values are expressed as PyTorch tensors.
"""
def __init__(
self,
env: Union[gym.Env],
*,
force_classic_api: bool = False,
discrete_to_continuous_act: bool = False,
clip_actions: bool = False,
**kwargs,
):
"""
`__init__(...)`: Initialize the TorchWrapper.
Args:
env: The gym environment to be wrapped.
force_classic_api: Set this as True if you would like to enable
the classic API. In the classic API, the `reset(...)` method
returns only the observation and the `step(...)` method
returns 4 elements (not 5).
discrete_to_continuous_act: When this is set as True and the
wrapped environment has a Discrete action space, this wrapper
will transform the action space to Box. A Discrete-action
environment with `n` actions will be converted to a Box-action
environment where the action length is `n`.
The index of the largest value within the action vector will
be applied to the underlying environment.
clip_actions: Set this as True if you would like to clip the given
actions so that they conform to the declared boundaries of the
action space.
kwargs: Expected in the form of additional keyword arguments.
These additional keyword arguments are passed to the
superclass.
"""
super().__init__(env, **kwargs)
# Declare the variable that will store the array type of the underlying environment.
self.__array_type: Optional[str] = None
if hasattr(env, "single_observation_space"):
# If the underlying environment has the attribute "single_observation_space",
# then this is a vectorized environment.
self.__vectorized = True
# Get the observation and action spaces.
obs_space = env.single_observation_space
act_space = env.single_action_space
else:
# If the underlying environment has the attribute "single_observation_space",
# then this is a non-vectorized environment.
self.__vectorized = False
# Get the observation and action spaces.
obs_space = env.observation_space
act_space = env.action_space
# Ensure that the observation and action spaces are supported.
_must_be_supported_space(obs_space)
_must_be_supported_space(act_space)
# Store the choice of the user regarding "force_classic_api".
self.__force_classic_api = bool(force_classic_api)
if isinstance(act_space, Discrete) and discrete_to_continuous_act:
# The underlying action space is Discrete and `discrete_to_continuous_act` is given as True.
# Therefore, we convert the action space to continuous (to Box).
# Take the shape and the dtype of the discrete action space.