Index
Utility classes and functions for neural networks
functional
¶
ModuleExpectingFlatParameters
¶
A wrapper which brings a functional interface around a torch module.
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 1dimensional (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.
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 1dimensional (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)
```
"""
@torch.no_grad()
def __init__(self, net: nn.Module, *, disable_autograd_tracking: bool = False):
"""
`__init__(...)`: Initialize the `ModuleExpectingFlatParameters` instance.
Args:
net: The module that is to be wrapped by a functional interface.
disable_autograd_tracking: If given as True, all operations
regarding the wrapped module will be performed in the context
`torch.no_grad()`, forcefully disabling the autograd.
If given as False, autograd will not be affected.
The default is False.
"""
# Declare the variables which will store information regarding the parameters of the module.
self.__param_names = []
self.__param_shapes = []
self.__param_length = 0
self.__param_slices = []
self.__num_params = 0
# Iterate over the parameters of the module and fill the related information.
i = 0
j = 0
for pname, p in net.named_parameters():
self.__param_names.append(pname)
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.__buffer_dict = {bname: b.clone() for bname, b in net.named_buffers()}
self.__net = deepcopy(net)
self.__net.to("meta")
self.__disable_autograd_tracking = bool(disable_autograd_tracking)
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.
"""
for bname in self.__buffer_dict.keys():
self.__buffer_dict[bname] = torch.as_tensor(self.__buffer_dict[bname], device=x.device)
@property
def buffers(self) > tuple:
"""Get the stored buffers"""
return tuple(self.__buffer_dict)
@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 1dimensional 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_and_buffers = {}
for i, pname in enumerate(self.__param_names):
param_slice = self.__param_slices[i]
param_shape = self.__param_shapes[i]
param = parameter_vector[param_slice].reshape(param_shape)
params_and_buffers[pname] = param
# Make sure that the buffer tensors are in the same device with x
self.__transfer_buffers(x)
# Add the buffer tensors to the dictionary `params_and_buffers`
params_and_buffers.update(self.__buffer_dict)
# Prepare the nogradient context if gradient tracking is disabled
context = torch.no_grad() if self.__disable_autograd_tracking else nullcontext()
# Run the module and return the results
with context:
return functional_call(self.__net, params_and_buffers, tuple([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 1dimensional 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 1dimensional 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_and_buffers = {}
for i, pname in enumerate(self.__param_names):
param_slice = self.__param_slices[i]
param_shape = self.__param_shapes[i]
param = parameter_vector[param_slice].reshape(param_shape)
params_and_buffers[pname] = param
# Make sure that the buffer tensors are in the same device with x
self.__transfer_buffers(x)
# Add the buffer tensors to the dictionary `params_and_buffers`
params_and_buffers.update(self.__buffer_dict)
# Prepare the nogradient context if gradient tracking is disabled
context = torch.no_grad() if self.__disable_autograd_tracking else nullcontext()
# Run the module and return the results
with context:
return functional_call(self.__net, params_and_buffers, tuple([x, *state_args]))
__init__(self, net, *, disable_autograd_tracking=False)
special
¶
__init__(...)
: Initialize the ModuleExpectingFlatParameters
instance.
Parameters:
Name  Type  Description  Default 

net 
Module 
The module that is to be wrapped by a functional interface. 
required 
disable_autograd_tracking 
bool 
If given as True, all operations
regarding the wrapped module will be performed in the context

False 
Source code in evotorch/neuroevolution/net/functional.py
@torch.no_grad()
def __init__(self, net: nn.Module, *, disable_autograd_tracking: bool = False):
"""
`__init__(...)`: Initialize the `ModuleExpectingFlatParameters` instance.
Args:
net: The module that is to be wrapped by a functional interface.
disable_autograd_tracking: If given as True, all operations
regarding the wrapped module will be performed in the context
`torch.no_grad()`, forcefully disabling the autograd.
If given as False, autograd will not be affected.
The default is False.
"""
# Declare the variables which will store information regarding the parameters of the module.
self.__param_names = []
self.__param_shapes = []
self.__param_length = 0
self.__param_slices = []
self.__num_params = 0
# Iterate over the parameters of the module and fill the related information.
i = 0
j = 0
for pname, p in net.named_parameters():
self.__param_names.append(pname)
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.__buffer_dict = {bname: b.clone() for bname, b in net.named_buffers()}
self.__net = deepcopy(net)
self.__net.to("meta")
self.__disable_autograd_tracking = bool(disable_autograd_tracking)
make_functional_module(net, *, disable_autograd_tracking=False)
¶
Wrap a torch module so that it has a functional interface.
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 1dimensional (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 
disable_autograd_tracking 
bool 
If given as True, all operations
regarding the wrapped module will be performed in the context

False 
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, *, disable_autograd_tracking: bool = False) > ModuleExpectingFlatParameters:
"""
Wrap a torch module so that it has a functional interface.
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 1dimensional (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.
disable_autograd_tracking: If given as True, all operations
regarding the wrapped module will be performed in the context
`torch.no_grad()`, forcefully disabling the autograd.
If given as False, autograd will not be affected.
The default is False.
Returns:
The functional wrapper, as an instance of
`evotorch.neuroevolution.net.ModuleExpectingFlatParameters`.
"""
return ModuleExpectingFlatParameters(net, disable_autograd_tracking=disable_autograd_tracking)
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 rightargument of the operation
(the leftargument 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 rightargument of the operation (the leftargument 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 rightargument of the operation
(the leftargument 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 reimplement this method in your own modules. Both singleline and multiline 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 reimplement this method in your own modules. Both singleline and multiline 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 reimplement this method in your own modules. Both singleline and multiline 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 tanhactivated 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 tanhactivated 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 nonlinear. The nonlinear component is an inputindependent set of sinusoidals waves whose amplitudes, frequencies and phases are trainable. Upon execution of a forward pass, the output of the nonlinear 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 nonlinear 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 locomotionspecific structured control net.
This is a control network which consists of two components:
one linear, and one nonlinear. The nonlinear component
is an inputindependent set of sinusoidals waves whose
amplitudes, frequencies and phases are trainable.
Upon execution of a forward pass, the output of the nonlinear
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 nonlinear 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 nonlinear 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 nonlinear 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 reimplement this method in your own modules. Both singleline and multiline 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 reimplement this method in your own modules. Both singleline and multiline 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 nonlinear component, which is a feedforward 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 nonlinear component, which is a feedforward 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 nonlinear component
hidden_size: Number of neurons in a hidden layer of the
nonlinear 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 nonlinear component 
required 
hidden_size 
int 
Number of neurons in a hidden layer of the nonlinear 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 nonlinear component
hidden_size: Number of neurons in a hidden layer of the
nonlinear 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 1D 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 1D 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 1D 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 1D 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 __len__(self):
return len(self._submodules)
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
from evotorch.neuroevolution.net import MultiLayered
net = MultiLayered(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 multiline:
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 continuousaction
environments, or number of actions for discreteaction
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
singlehiddenlayered 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 hardtanh 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
from evotorch.neuroevolution.net import MultiLayered
net = MultiLayered(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 multiline:
```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 continuousaction
environments, or number of actions for discreteaction
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
singlehiddenlayered 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 hardtanh 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 nonvectorized 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 nonvectorized 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)
¶
Uses the :meth:step
of the :attr:env
that can be overwritten to change the returned data.
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 gymnasium environment.
Even though the gymnasium
library switched to a new API where the
reset()
method returns a tuple (observation, info)
, this function
follows the conventions of the classical gym
library and returns
only the observation of the newly reset environment.
Parameters:
Name  Type  Description  Default 

env 
Env 
The gymnasium 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 gymnasium environment.
Even though the `gymnasium` library switched to a new API where the
`reset()` method returns a tuple `(observation, info)`, this function
follows the conventions of the classical `gym` library and returns
only the observation of the newly reset environment.
Args:
env: The gymnasium 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 gymnasium environment. Taking a step means performing the action provided via the arguments.
Even though the gymnasium
library switched to a new API where the
step()
method returns a 5element tuple of the form
(observation, reward, terminated, truncated, info)
, this function
follows the conventions of the classical gym
library and returns
a 4element tuple (observation, reward, done, info)
.
Parameters:
Name  Type  Description  Default 

env 
Env 
The gymnasium environment in which the action will be performed. 
required 
action 
Iterable 
The action to 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 gymnasium environment.
Taking a step means performing the action provided via the arguments.
Even though the `gymnasium` library switched to a new API where the
`step()` method returns a 5element tuple of the form
`(observation, reward, terminated, truncated, info)`, this function
follows the conventions of the classical `gym` library and returns
a 4element tuple `(observation, reward, done, info)`.
Args:
env: The gymnasium environment in which the action will be performed.
action: The action to 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 = 1e2,
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 1e2.
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 1dimensional Iterable of booleans ONLY
if `x` represents a batch of observations.
If a `mask` is provided, the ith observation within the
observation batch `x` will be taken into account only if
the ith 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 nonempty
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 1dimensional.
# If not, we raise an error.
raise ValueError(
f"The `mask` tensor was expected as a 1dimensional 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 recast 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 CPUbound 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 1dimensional Iterable of booleans ONLY
if `x` represents a batch of observations.
If a `mask` is provided, the ith observation within the
observation batch `x` will be taken into account only if
the the ith 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 = 1e2,
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 1e2.
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 CPUbound 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 1dimensional 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 1dimensional Iterable of booleans ONLY
if `x` represents a batch of observations.
If a `mask` is provided, the ith observation within the
observation batch `x` will be taken into account only if
the ith 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 nonempty
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 1dimensional.
# If not, we raise an error.
raise ValueError(
f"The `mask` tensor was expected as a 1dimensional 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 recast 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 1dimensional 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 1dimensional Iterable of booleans ONLY
if `x` represents a batch of observations.
If a `mask` is provided, the ith observation within the
observation batch `x` will be taken into account only if
the the ith 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 nonvectorized observations. With this simplified interface, RunningStat is meant to be used by GymNE, on classical nonvectorized 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 nonvectorized
observations.
With this simplified interface, RunningStat is meant to be used
by GymNE, on classical nonvectorized 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 nonrecurrent 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 noop 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 forwardpassing 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 nonrecurrent 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 noop 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 forwardpassing 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 

StatefulModule 
The module 
Source code in evotorch/neuroevolution/net/statefulmodule.py
def ensure_stateful(net: nn.Module) > StatefulModule:
"""
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)
return 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 nonrecurrent
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 forwardpass.
action2 = policy(observation2)
policy.reset()
# action3 will be computed according to the renewed hidden state.
action3 = policy(observation3)
Both for nonrecurrent and recurrent networks, it is possible to perform vectorized operations. For now, let us return to our first nonrecurrent 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 nonbatched 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 nonrecurrent
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 forwardpass.
action2 = policy(observation2)
policy.reset()
# action3 will be computed according to the renewed hidden state.
action3 = policy(observation3)
```
Both for nonrecurrent and recurrent networks, it is possible to
perform vectorized operations. For now, let us return to our
first nonrecurrent 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 nonbatched 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 1dimensional or a 2dimensional tensor containing
the flattened parameters to be used with the neural network.
If the given parameters are twodimensional, 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
2dimensional 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 2dimensional
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 1dimensional tensor, then this
argument is expected to store a single observation.
If the parameters were given as a 2dimensional 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 inplace. 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 inplace, and the action
tensor could be involuntarily reset as well.
This inplace 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 1dimensional tensor, then this
argument is expected to store a single observation.
If the parameters were given as a 2dimensional 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 inplace. 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 inplace, and the action
tensor could be involuntarily reset as well.
This inplace 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 1dimensional or a 2dimensional tensor containing
the flattened parameters to be used with the neural network.
If the given parameters are twodimensional, 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 2dimensional tensor, provide this argument if you would like to change only some rows of the previously given parameters 