Elman recurrent network trained by backpropagation
Elman's simple recurrent network, which learns to predict the next bit of a temporal XOR stream.
- Name:
elman_network - Field:
computer_science/neural_networks - Formalism: learning by gradient steps
- References: Gros 2015 §10.5.1; Elman 1990
Factory
@register_model(
field='computer_science/neural_networks',
tags=('neural network', 'recurrent network', 'learning', 'backpropagation through time', 'discrete map'),
references=('Gros 2015 §10.5.1', 'Elman 1990'),
defaults={'until': 200, 'dt': 1},
)
def elman_network(
n_hidden: int = 4,
*,
n_bits: int = 3000,
window: int = 12,
batch_size: int = 16,
learning_rate: float = 0.05,
truncation: int | None = None,
random_state: int | np.random.Generator | None = 0,
) -> System:
"""Return Elman's simple recurrent network, which learns to predict the next bit of a temporal XOR stream.
The nested discrete map `net` holds one input unit, `n_hidden` hidden units, and one output unit. The input sets
the drive `u = D x` of the hidden units, the hidden units take `s <- sigmoid(A s + u + b)` (Gros 2015 Eq. 10.18),
and the output predicts `y <- sigmoid(W s + c)`, in three phases of every step, so the prediction of a step reads
its input. The training interaction `training` trains `D`, `A`, `W`, `b`, and `c` with Adam on the squared error
of predicting the next input from the inputs so far, over windows of `window` bits that start at every third bit of
a stream in which every third bit is the exclusive or of the two before it. Only those bits can be predicted, so
the metric `xor_error` falls toward 0 while `random_error`, the error on the other bits, stays near 0.25. One step
of the system is one epoch.
Args:
n_hidden:
The number of hidden units.
n_bits:
The length of the stream.
window:
The number of bits of a sequence.
batch_size:
The number of sequences of a batch.
learning_rate:
The learning rate of Adam.
truncation:
The number of steps the gradient flows back through the hidden activities, 1 for Elman's original rule
that holds the previous activities as constant context, or `None` for full backpropagation through time.
random_state:
The seed or generator of the stream and of the initial weights.
Returns:
system:
The model, with the nested system `net` and the training interaction `training`.
Examples:
>>> from skcomplex.models.computer_science.neural_networks import elman_network
>>> [item.label for item in elman_network().systems]
['net']
"""
rng = np.random.default_rng(random_state)
data = _dataset(_temporal_xor(n_bits, rng), window)
net = System(
'net',
elements=[
GroupElement('inputs', element=InputUnitElement, size=1),
GroupElement('hidden', element=HiddenUnitElement, size=n_hidden),
GroupElement('outputs', element=OutputUnitElement, size=1),
],
interactions=[
InputCouplingInteraction(
'input',
target='hidden',
source='inputs',
D=rng.normal(0.0, WEIGHT_SCALE, (n_hidden, 1)),
),
RecurrenceCouplingInteraction(
'recurrence',
target='hidden',
A=rng.normal(0.0, WEIGHT_SCALE, (n_hidden, n_hidden)),
),
ReadoutCouplingInteraction(
'readout',
target='outputs',
source='hidden',
W=rng.normal(0.0, WEIGHT_SCALE, (1, n_hidden)),
),
],
time=DiscreteTime(schedule=PhasesScheduleSpec(PHASES)),
)
training = TrainingInteraction(
'training',
model='net',
trained=TRAINED,
loss=ObservableSpec('error', group='outputs', reduce='sum', expression=_squared_error),
data=data,
learning_rate=learning_rate,
batch_size=batch_size,
truncation=truncation,
metrics=(
ObservableSpec('xor_error', group='outputs', reduce='sum', expression=_squared_error),
ObservableSpec('random_error', group='outputs', reduce='sum', expression=_squared_error),
),
)
return System('elman', systems=[net], interactions=[training], time=DiscreteTime())
Components
class TrainingInteraction(Interaction):
"""Training of the adaptive parameters of a nested system by gradient steps on a loss over unrolled runs.
The role `model` binds the path of a nested system, which the training unrolls over the sequences of a dataset
from its initial states, inside one step of the discrete time of the system that holds the interaction. That step
is one epoch of shuffled batches, each followed by one update of the optimizer, and the interaction returns the
`GradientStepLaw` that takes it. The hyperparameters of the optimizer are parameters of the interaction, so a
schedule or `Simulation.update_parameters` changes them between epochs. `'sgd'` reads the learning rate,
`'momentum'` also the momentum, `'rmsprop'` the learning rate and `beta2`, the decay of its mean squared gradient,
and `'adam'` the learning rate, `beta1`, and `beta2`. It differs from `GradientInteraction`, which moves states
down the gradient of a potential in the system's own time.
Args:
label:
The label of the interaction, unique among the parts of its system.
model:
The path of the nested system whose parameters the interaction trains.
trained:
The dotted paths of the adaptive parameters of the nested system to train, such as `readout.W`.
loss:
The observable of the nested system whose weighted mean over the steps of a batch is minimized.
data:
The sequences of the parameters of the nested system that set its inputs and targets at every step.
optimizer:
The name of the optimizer, one of `OPTIMIZERS`.
learning_rate:
The learning rate of the optimizer.
momentum:
The momentum of `'momentum'`.
beta1:
The decay of the mean gradient of `'adam'`.
beta2:
The decay of the mean squared gradient of `'adam'` and `'rmsprop'`.
batch_size:
The number of sequences of a batch.
truncation:
The number of inner steps the gradient flows back through the inner states, or `None` for the whole run.
A truncation of 1 holds the previous inner states as constant inputs, as Elman's original training does.
metrics:
The observables of the nested system recorded at the end of every epoch.
Raises:
ValidationError:
The optimizer is unknown, the batch size or the truncation is not a positive integer, no parameter is
trained, or the loss, the data, or a metric has the wrong type or label.
Examples:
>>> import numpy as np
>>> from skcomplex.core.interactions import TrainingInteraction
>>> from skcomplex.specs import DatasetSpec, ObservableSpec
>>> data = DatasetSpec({'inputs.x': np.zeros((4, 3, 1))})
>>> training = TrainingInteraction(
... 'training', model='net', trained=['readout.W'], loss=ObservableSpec('error', 0.0), data=data
... )
>>> training.check().optimizer
'adam'
"""
def __init__(
self,
label: str,
*,
model: Annotated[str, RoleSpec()],
trained: Sequence[str],
loss: ObservableSpec,
data: DatasetSpec,
optimizer: str = 'adam',
learning_rate: Annotated[float, ParameterSpec(domain=RealSet.closed_open(0, oo))] = 1e-3,
momentum: Annotated[float, ParameterSpec(domain=RealSet.closed_open(0, 1))] = 0.9,
beta1: Annotated[float, ParameterSpec(domain=RealSet.closed_open(0, 1))] = 0.9,
beta2: Annotated[float, ParameterSpec(domain=RealSet.closed_open(0, 1))] = 0.999,
batch_size: int = 32,
truncation: int | None = None,
metrics: Sequence[ObservableSpec] = (),
) -> None:
super().__init__(label)
self.model = model
self.trained = trained
self.loss = loss
self.data = data
self.optimizer = optimizer
self.learning_rate = learning_rate
self.momentum = momentum
self.beta1 = beta1
self.beta2 = beta2
self.batch_size = batch_size
self.truncation = truncation
self.metrics = metrics
def _check_counts(self) -> None:
"""Raise an error for an unknown optimizer, or a batch size or truncation that is not a positive integer."""
if self.optimizer not in OPTIMIZERS:
error_msg = (
f'The optimizer of {self.label!r} is {self.optimizer!r}, which is not supported. Choose one of '
f'{", ".join(map(repr, OPTIMIZERS))}.'
)
raise ValidationError(error_msg)
for name in ('batch_size', 'truncation'):
value = getattr(self, name)
if name == 'truncation' and value is None:
continue
if not isinstance(value, int) or isinstance(value, bool) or value < 1:
error_msg = f'The {name} of {self.label!r} is {value!r}. Give a positive integer.'
raise ValidationError(error_msg)
def _check_observables(self) -> None:
"""Raise an error for no trained parameter, a loss, data, or metric of the wrong type, or a repeated label."""
if isinstance(self.trained, str) or not self.trained or not all(isinstance(p, str) for p in self.trained):
error_msg = f'{self.label!r} trains {self.trained!r}. Give the dotted paths of adaptive parameters.'
raise ValidationError(error_msg)
if not isinstance(self.data, DatasetSpec):
error_msg = f'The data of {self.label!r} is a {type(self.data).__name__}. Give a `DatasetSpec`.'
raise ValidationError(error_msg)
observables = [self.loss, *self.metrics]
if not all(isinstance(item, ObservableSpec) for item in observables):
error_msg = f'The loss and the metrics of {self.label!r} are observables. Give `ObservableSpec`s.'
raise ValidationError(error_msg)
labels = ['loss', *(item.label for item in self.metrics)]
if len(set(labels)) < len(labels):
error_msg = (
f'The metrics of {self.label!r} are labelled {labels[1:]}. Give distinct labels other than loss.'
)
raise ValidationError(error_msg)
def _check(self) -> None:
"""Raise an error for an optimizer, counts, trained parameters, or observables the training cannot take.
Raises:
ValidationError:
The optimizer is unknown, the batch size or the truncation is not a positive integer, no parameter is
trained, or the loss, the data, or a metric has the wrong type or label.
"""
super()._check()
self._check_counts()
self._check_observables()
self.data.check()
def _get_laws(self, namespace: SimpleNamespace) -> Sequence[BasePart]:
"""Return the gradient-step law of the training."""
del namespace
law = GradientStepLaw(
self.trained,
self.loss,
self.data,
self.optimizer,
self.batch_size,
truncation=self.truncation,
metrics=self.metrics,
label=self.label,
)
return [law]
class InputUnitElement(Element):
"""Input unit of a simple recurrent network, whose value a dataset or a schedule sets at every step.
Args:
label:
The label of the element or group.
visual:
How the element is drawn, or `None` for `default_visual_`.
x:
The input.
Examples:
>>> from skcomplex.models.computer_science.neural_networks import InputUnitElement
>>> InputUnitElement('bit').x
0.0
"""
def __init__(
self,
label: str,
visual: VisualSpec | None = None,
*,
x: Annotated[float, ParameterSpec()] = 0.0,
) -> None:
super().__init__(label, visual=visual)
self.x = x
class HiddenUnitElement(Element):
"""Hidden unit of a simple recurrent network, whose activity is a sigmoid of its recurrent input and its drive.
The drive `u` holds the weighted input of the previous step, and the bias `b` is adaptive, so training changes it.
Args:
label:
The label of the element or group.
visual:
How the element is drawn, or `None` for `default_visual_`.
s:
The activity in `(0, 1)`.
u:
The weighted input of the previous step.
b:
The bias, an adaptive parameter.
Examples:
>>> from skcomplex.models.computer_science.neural_networks import HiddenUnitElement
>>> HiddenUnitElement('unit').s
0.5
"""
def __init__(
self,
label: str,
visual: VisualSpec | None = None,
*,
s: Annotated[float, StateSpec()] = 0.5,
u: Annotated[float, StateSpec()] = 0.0,
b: Annotated[float, ParameterSpec(adaptive=True)] = 0.0,
) -> None:
super().__init__(label, visual=visual)
self.s = s
self.u = u
self.b = b
class OutputUnitElement(Element):
"""Output unit of a simple recurrent network, whose activity predicts a target.
The bias `c` is adaptive, so training changes it, and the target is a parameter a dataset or a schedule sets.
Args:
label:
The label of the element or group.
visual:
How the element is drawn, or `None` for `default_visual_`.
y:
The prediction in `(0, 1)`.
c:
The bias, an adaptive parameter.
target:
The value the prediction is compared with.
Examples:
>>> from skcomplex.models.computer_science.neural_networks import OutputUnitElement
>>> OutputUnitElement('prediction').target
0.0
"""
def __init__(
self,
label: str,
visual: VisualSpec | None = None,
*,
y: Annotated[float, StateSpec()] = 0.5,
c: Annotated[float, ParameterSpec(adaptive=True)] = 0.0,
target: Annotated[float, ParameterSpec()] = 0.0,
) -> None:
super().__init__(label, visual=visual)
self.y = y
self.c = c
self.target = target
class InputCouplingInteraction(CouplingInteraction):
"""Weighted input of a simple recurrent network, which sets the drive `u = D x` of every hidden unit each step.
The weights `D` are an adaptive matrix with one row per hidden unit and one column per input unit, so training
changes them. The coupling runs in the phase `drive` of the step, which the system declares before the phases
`hidden` of the recurrence and `readout` of the readout, so the hidden units read the drive of the same step.
Args:
label:
The label of the interaction, unique among the parts of its system.
target:
The path of the group of hidden units.
source:
The path of the group of input units.
D:
The input weights, a matrix of shape `(hidden units, input units)` or a number for every entry.
Examples:
>>> from skcomplex.models.computer_science.neural_networks import InputCouplingInteraction
>>> InputCouplingInteraction('input', target='hidden', source='inputs').write
'u'
"""
def __init__(
self,
label: str,
*,
target: Annotated[str, RoleSpec()],
source: Annotated[str, RoleSpec()],
D: Annotated[object, ParameterSpec(shape=('target', 'source'), adaptive=True)] = 0.0,
) -> None:
super().__init__(
label,
target=target,
source=source,
read='x',
write='u',
function=self._input,
mode='group',
weights='D',
form='map',
phase='drive',
)
self.D = D
def _input(self, namespace: SimpleNamespace) -> object:
"""Return the input of the other unit."""
return namespace.other.x
class RecurrenceCouplingInteraction(CouplingInteraction):
"""Recurrence of a simple recurrent network, `s_i <- sigmoid(sum_j A_ij s_j + u_i + b_i)` with the logistic sigmoid.
Every hidden unit reads the activities of every hidden unit, itself included, through the adaptive matrix `A`, adds
its drive `u` and its bias `b`, and takes the sigmoid of the sum as its next activity (Gros 2015 Eq. 10.18).
Args:
label:
The label of the interaction, unique among the parts of its system.
target:
The path of the group of hidden units.
A:
The recurrent weights, a square matrix with one row and one column per hidden unit, or a number for every
entry.
Examples:
>>> from skcomplex.models.computer_science.neural_networks import RecurrenceCouplingInteraction
>>> RecurrenceCouplingInteraction('recurrence', target='hidden').include_self
True
"""
def __init__(
self,
label: str,
*,
target: Annotated[str, RoleSpec()],
A: Annotated[object, ParameterSpec(shape=('target', 'target'), adaptive=True)] = 0.0,
) -> None:
super().__init__(
label,
target=target,
read='s',
write='s',
function=self._activity,
mode='group',
weights='A',
include_self=True,
form='map',
phase='hidden',
combine=self._activate,
)
self.A = A
def _activity(self, namespace: SimpleNamespace) -> object:
"""Return the activity of the other unit."""
return namespace.other.s
def _activate(self, namespace: SimpleNamespace, reduction: object) -> object:
"""Return the sigmoid of the recurrent input, the drive, and the bias of the unit."""
return 1 / (1 + exp(-(reduction + namespace.member.u + namespace.member.b)))
class ReadoutCouplingInteraction(CouplingInteraction):
"""Readout of a simple recurrent network, `y_i <- sigmoid(sum_j W_ij s_j + c_i)` with the logistic sigmoid.
Every output unit reads the activities of the hidden units through the adaptive matrix `W`, adds its bias `c`, and
takes the sigmoid of the sum as its next prediction.
Args:
label:
The label of the interaction, unique among the parts of its system.
target:
The path of the group of output units.
source:
The path of the group of hidden units.
W:
The readout weights, a matrix of shape `(output units, hidden units)` or a number for every entry.
Examples:
>>> from skcomplex.models.computer_science.neural_networks import ReadoutCouplingInteraction
>>> ReadoutCouplingInteraction('readout', target='outputs', source='hidden').write
'y'
"""
def __init__(
self,
label: str,
*,
target: Annotated[str, RoleSpec()],
source: Annotated[str, RoleSpec()],
W: Annotated[object, ParameterSpec(shape=('target', 'source'), adaptive=True)] = 0.0,
) -> None:
super().__init__(
label,
target=target,
source=source,
read='s',
write='y',
function=self._activity,
mode='group',
weights='W',
form='map',
phase='readout',
combine=self._activate,
)
self.W = W
def _activity(self, namespace: SimpleNamespace) -> object:
"""Return the activity of the hidden unit."""
return namespace.other.s
def _activate(self, namespace: SimpleNamespace, reduction: object) -> object:
"""Return the sigmoid of the weighted activities and the bias of the output unit."""
return 1 / (1 + exp(-(reduction + namespace.member.c)))