Asymmetric diluted Hopfield network
The asymmetric diluted Hopfield network, which recalls a stored pattern below its memory capacity.
- Name:
diluted_hopfield - Field:
computer_science/neural_networks - Formalism: discrete map
- References: Derrida, Gardner, and Zippelius 1987; Bertin 2021
Factory
@register_model(
field='computer_science/neural_networks',
tags=('neural network', 'associative memory', 'spin glass', 'discrete map'),
references=('Derrida, Gardner, and Zippelius 1987', 'Bertin 2021'),
defaults={'until': 20, 'dt': 1},
observables=lambda system: [ObservableSpec('overlap', _overlap, group='neurons')],
)
def diluted_hopfield(
n_neurons: int = 4000,
*,
connectivity: float = 50.0,
load: float = 0.2,
temperature_scale: float = 3.0,
overlap: float = 0.3,
random_state: int | np.random.Generator | None = 0,
) -> System:
"""Return the asymmetric diluted Hopfield network, which recalls a stored pattern below its memory capacity.
The network stores `p = load K` random patterns in the couplings `J_ij = C_ij Σ_μ ξ_i ξ_j`, where every `C_ij` is
1 with probability `K / N` independently of `C_ji`, so the couplings are sparse and asymmetric. All neurons update
together at the inverse temperature `β = temperature_scale / K`. Started with an overlap `m` with the first pattern,
the overlap follows `m(t + 1) = f(m(t))` for large `K`, which keeps a nonzero fixed point below the critical load
`2 / π`.
Args:
n_neurons:
The number of neurons `N`.
connectivity:
The mean number `K` of inputs of a neuron.
load:
The ratio `p / K` of the number of patterns to the connectivity.
temperature_scale:
The product `β K` of the inverse temperature and the connectivity.
overlap:
The initial overlap with the first pattern.
random_state:
The seed or generator of the patterns, the dilution, and the initial values.
Returns:
system:
The model, with the values `neurons.spin` and the observable `overlap`.
Examples:
>>> from skcomplex.models.computer_science.neural_networks import diluted_hopfield
>>> diluted_hopfield(200, connectivity=10).elements[0].check().prepare().size_
200
"""
rng = np.random.default_rng(random_state)
n_patterns = max(1, round(load * connectivity))
patterns = rng.choice([-1.0, 1.0], size=(n_patterns, n_neurons))
spins = np.where(rng.random(n_neurons) < (1 + overlap) / 2, patterns[0], -patterns[0])
neurons = GroupElement(
'neurons',
element=DilutedNeuronElement,
size=n_neurons,
values={
'spin': spins,
'pattern': patterns[0],
'beta': temperature_scale / connectivity,
},
)
glauber = ChoiceInteraction(
'glauber',
target='neurons',
state='spin',
options=(-1.0, 1.0),
utility=functools.partial(_alignment, couplings=_couplings(patterns, connectivity, rng)),
)
return System(elements=[neurons], interactions=[glauber])
Components
class DilutedNeuronElement(Element):
"""Binary neuron of an asymmetric diluted Hopfield network, updated in parallel with the others at every step.
The logit choice of `diluted_hopfield` draws the value `s = ±1` of the neuron with probability
`1 / (1 + exp(-2 s β h))`, where `h = Σ_j J_ij s_j` is its local field, so its mean is `tanh(β h)`.
Args:
label:
The label of the neuron or the group.
visual:
How the element is drawn, or `None` for `default_visual_`.
spin:
The initial value, -1 or 1.
pattern:
The value of the first stored pattern at the neuron, whose overlap with the network is recorded.
beta:
The inverse temperature `β`.
Examples:
>>> from skcomplex.models.computer_science.neural_networks import DilutedNeuronElement
>>> DilutedNeuronElement('neuron').beta
0.06
"""
def __init__(
self,
label: str,
visual: VisualSpec | None = None,
*,
spin: Annotated[float, StateSpec(domain=ZZ)] = 1.0,
pattern: Annotated[float, ParameterSpec()] = 1.0,
beta: Annotated[float, ParameterSpec(domain=RealSet.closed_open(0, oo))] = 0.06,
) -> None:
super().__init__(label, visual=visual)
self.spin = spin
self.pattern = pattern
self.beta = beta
class ChoiceInteraction(Interaction):
"""Choice of every member of a group among options, drawn every step from the utilities of the options by a rule.
The rule `'logit'`, which is the heat-bath rule, draws option `k` of utility `u_k` with probability
`exp(beta u_k) / sum_j exp(beta u_j)`, where `beta` is the intensity of choice. The rules `'metropolis'` and
`'fermi'` propose one of the other options uniformly, or with the proposal `'all'` one of all the options
including the current one, and accept it with probability `min(1, exp(beta (u_new - u_old)))` or
`1 / (1 + exp(-beta (u_new - u_old)))`, so the rule `'fermi'` with the proposal `'all'` is the heat-bath rule of a
uniform proposal. The rule `'best'` draws uniformly among the options of largest utility under the ties
`'random'`, and takes the first of them under `'first'`. Every step, each member revises its choice with probability
`activity` and keeps it otherwise.
In the `'symbolic'` form, `utility(namespace, option)` returns the utility as a symbolic expression, where
`namespace.target` holds the states and parameters of the group, and every member draws at once. In the `'agent'`
form, the group is an agent group and `utility(agent, model, option)` returns the utility as a number, so it can
read the agent, its neighbours, and the history and storage of the model view. Its members draw at once, from the
choices of the last step, under the `'synchronous'` schedule, one after the other in their order or in a new random
order, each seeing the choices made before it, under the `'sequential'` or `'random-sequential'` schedule, and each
in its own activation under the `'member'` schedule, together with the rules of its element and the other member
interactions, in the order the system lists them. An option of utility `-inf` is never drawn in the agent form.
Args:
label:
The label of the interaction, unique among the parts of its system.
target:
The path of the element or group whose members choose.
state:
The name of the discrete state of the members that holds the chosen option.
options:
The values the state can take, at least two.
utility:
The function `utility(namespace, option)` in the symbolic form, or `utility(agent, model, option)` in the
agent form, that returns the utility of an option.
rule:
The rule of the draw, `'logit'`, `'metropolis'`, `'fermi'`, or `'best'`.
intensity:
The intensity of choice `beta`, zero for uniform choice.
schedule:
The order in which the members revise, `'synchronous'` or, in the agent form, `'sequential'`,
`'random-sequential'`, or `'member'`.
activity:
The probability that a member revises its choice in a step.
form:
`'symbolic'` for update laws of the states of a group, or `'agent'` for a rule over an agent group.
phase:
The phase of the step the agent form runs in, or `None` for the default phase.
proposal:
The options the rules `'metropolis'` and `'fermi'` propose from, `'others'` for the options other than the
current one, or `'all'` for every option.
ties:
The option the rule `'best'` takes among options of equal largest utility, `'random'` for one drawn
uniformly or `'first'` for the first in the order of the options.
Raises:
ValidationError:
The rule, the schedule, the form, the proposal, or the ties are unknown, the schedule is not synchronous in
the symbolic form, there are fewer than two options, or the utility is not callable.
Examples:
>>> from skcomplex.core.interactions import ChoiceInteraction
>>> from skcomplex.core.laws.utils import mean
>>> choice = ChoiceInteraction(
... 'choice', target='spins', state='spin', options=(-1, 1), utility=lambda s, o: o * mean(s.target.spin)
... )
>>> choice.check().rule
'logit'
>>> agents = ChoiceInteraction(
... 'choice', target='traders', state='plan', options=(0, 1), utility=lambda a, m, o: o, form='agent'
... )
>>> type(agents.check().prepare().run(SimpleNamespace()).laws_[0]).__name__
'RuleLaw'
>>> ChoiceInteraction(
... 'heat-bath', target='spins', state='spin', options=(-1, 0, 1), utility=lambda s, o: o, rule='fermi',
... proposal='all'
... ).check().proposal
'all'
"""
def __init__(
self,
label: str,
*,
target: Annotated[str, RoleSpec()],
state: str,
options: Sequence[float],
utility: Callable[..., float | Expression],
rule: str = 'logit',
intensity: Annotated[float, ParameterSpec(domain=RealSet.closed_open(0, oo))] = 1.0,
schedule: str = 'synchronous',
activity: Annotated[float, ParameterSpec(domain=RealSet.closed(0, 1))] = 1.0,
form: str = 'symbolic',
phase: str | None = None,
proposal: str = 'others',
ties: str = 'random',
) -> None:
super().__init__(label)
self.target = target
self.state = state
self.options = options
self.utility = utility
self.rule = rule
self.intensity = intensity
self.schedule = schedule
self.activity = activity
self.form = form
self.phase = phase
self.proposal = proposal
self.ties = ties
def _check_variants(self) -> None:
"""Raise an error for an unknown variant, or a symbolic choice that is not synchronous."""
self._check_variant('rule', self.rule, RULES, (BaseSpec,))
for name, value, variants in (
('schedule', self.schedule, SCHEDULES),
('form', self.form, FORMS),
('proposal', self.proposal, PROPOSALS),
('ties', self.ties, TIES),
):
if value not in variants:
error_msg = f'Unknown {name} {value!r} of {self.label!r}. Use one of {variants}.'
raise ValidationError(error_msg)
if self.form == 'symbolic' and self.schedule != 'synchronous':
error_msg = (
f'{self.label!r} has the schedule {self.schedule!r}, which only the agent form runs. Use '
"form='agent' or schedule='synchronous'."
)
raise ValidationError(error_msg)
def _check(self) -> None:
"""Raise an error for an unknown variant, fewer than two options, or a utility that cannot be called.
Raises:
ValidationError:
The type declares states, the rule, the schedule, the form, the proposal, or the ties are unknown, the
schedule is not synchronous in the symbolic form, there are fewer than two options, or the utility is
not callable.
"""
super()._check()
self._check_variants()
if len(self.options) < MIN_OPTIONS:
error_msg = f'{self.label!r} chooses among {list(self.options)}. Give at least {MIN_OPTIONS} options.'
raise ValidationError(error_msg)
if not callable(self.utility):
error_msg = f'The utility of {self.label!r} must be callable, but it is {self.utility!r}.'
raise ValidationError(error_msg)
def _select(self, index: Expression, values: Sequence[Expression]) -> Expression:
"""Return the value at a symbolic index."""
return sum((where(index == position, value, 0) for position, value in enumerate(values)), SR(0))
def _draw_softmax(self, state: Expression, utilities: Sequence[Expression], intensity: Expression) -> Expression:
"""Return the option drawn with probabilities proportional to `exp(beta u_k)`."""
index = categorical(f'{state}_option', [exp(intensity * utility) for utility in utilities])
return self._select(index, [SR(option) for option in self.options])
def _draw_best(self, state: Expression, utilities: Sequence[Expression]) -> Expression:
"""Return an option drawn uniformly among the options of largest utility."""
largest = utilities[0]
for utility in utilities[1:]:
largest = where(utility > largest, utility, largest)
weights = [where(utility >= largest, 1, 0) for utility in utilities]
if self.ties == 'first':
weights = [
weight * where(sum(weights[:position], SR(0)) > 0, 0, 1) for position, weight in enumerate(weights)
]
index = categorical(f'{state}_best', weights)
return self._select(index, [SR(option) for option in self.options])
def _draw_proposal(self, state: Expression, utilities: Sequence[Expression], intensity: Expression) -> Expression:
"""Return a uniform proposal of an option, accepted by the Metropolis or the Fermi probability."""
weights = [SR(1) if self.proposal == 'all' else where(state == option, 0, 1) for option in self.options]
index = categorical(f'{state}_proposal', weights)
proposal = self._select(index, [SR(option) for option in self.options])
gain = self._select(index, utilities) - sum(
(where(state == option, utility, 0) for option, utility in zip(self.options, utilities, strict=True)),
SR(0),
)
if self.rule == 'metropolis':
acceptance = where(gain >= 0, 1, exp(intensity * gain))
else:
acceptance = 1 / (1 + exp(-intensity * gain))
return where(uniform(f'{state}_acceptance') < acceptance, proposal, state)
def _draw_symbolic(self, namespace: SimpleNamespace) -> Expression:
"""Return the option a member draws, as a symbolic expression of the namespace."""
state = getattr(self._get_member(namespace, 'target'), self.state)
utilities = [SR(self.utility(namespace, option)) for option in self.options]
if self.rule == 'logit':
return self._draw_softmax(state, utilities, self._get_parameter_symbol('intensity', namespace))
if self.rule == 'best':
return self._draw_best(state, utilities)
if self.rule in ('metropolis', 'fermi'):
return self._draw_proposal(state, utilities, self._get_parameter_symbol('intensity', namespace))
raise self._get_variant_error(self.rule, 'rule', '_draw_symbolic')
def _get_utilities(self, agent: AgentViewProtocol, model: ModelViewProtocol) -> np.ndarray:
"""Return the utility of every option for an agent."""
return np.array([float(self.utility(agent, model, option)) for option in self.options])
def _draw_agent_proposal(self, agent: AgentViewProtocol, model: ModelViewProtocol, intensity: float) -> float:
"""Return a uniform proposal of an option of an agent if accepted, or its current option."""
current = float(getattr(agent, self.state))
others = [option for option in self.options if self.proposal == 'all' or option != current]
proposal = float(others[int(model.rng.integers(len(others)))])
before = float(self.utility(agent, model, current)) if current in self.options else 0.0
gain = float(self.utility(agent, model, proposal)) - before
if self.rule == 'metropolis':
acceptance = 1.0 if gain >= 0 else math.exp(intensity * gain)
else:
acceptance = float(expit(intensity * gain))
return proposal if model.rng.random() < acceptance else current
def _draw_agent(self, agent: AgentViewProtocol, model: ModelViewProtocol, intensity: float) -> float:
"""Return the option an agent draws by the rule."""
if self.rule in ('metropolis', 'fermi'):
return self._draw_agent_proposal(agent, model, intensity)
utilities = self._get_utilities(agent, model)
if self.rule == 'best':
return float(self.options[self._pick_best(utilities, self.ties, model)])
if self.rule != 'logit':
raise self._get_variant_error(self.rule, 'rule', '_draw_agent')
scaled = np.where(np.isneginf(utilities), -np.inf, intensity * utilities)
probabilities = softmax(scaled)
return float(self.options[int(model.rng.choice(len(self.options), p=probabilities))])
def _is_active(self, model: ModelViewProtocol, activity: float) -> bool:
"""Return whether a member revises its choice in this step."""
return activity >= 1 or bool(model.rng.random() < activity)
def _get_choice(self, agent: AgentViewProtocol, model: ModelViewProtocol) -> float | None:
"""Return the option an agent draws if it revises in this step, or `None` to keep its own."""
if not self._is_active(model, self._get_parameter_value('activity', model)):
return None
return self._draw_agent(agent, model, self._get_parameter_value('intensity', model))
def _set_choice(self, agent: AgentViewProtocol, option: float | None) -> None:
"""Set the state of an agent to the option it draws, if any."""
if option is not None:
setattr(agent, self.state, option)
def _choose_member(self, agent: AgentViewProtocol, model: ModelViewProtocol) -> None:
"""Set the state of one agent to the option it draws if it revises, in its own activation."""
if agent.alive:
self._set_choice(agent, self._get_choice(agent, model))
def choose(self, model: ModelViewProtocol) -> None:
"""Set the state of every agent that revises in this step to the option it draws, by the schedule.
Args:
model:
The model view of the interaction.
"""
self._update_agents(model, self.schedule, self._get_choice, self._set_choice)
def _get_laws(self, namespace: SimpleNamespace) -> list[BasePart]:
"""Return the update of the state of every member that revises, or the rule of the agent form.
Args:
namespace:
The namespace of the symbols the laws can use, with `namespace.target`, `namespace.intensity`, and
`namespace.activity`.
Returns:
laws:
The update law of the state, or the rule that runs the choice of the agents.
"""
if self.form == 'agent' and self.schedule == 'member':
return [RuleLaw(self._choose_member, phase=self.phase, members=self._get_role('target'))]
if self.form == 'agent':
return [RuleLaw(self.choose, phase=self.phase)]
state = getattr(self._get_member(namespace, 'target'), self.state)
revised = where(
uniform(f'{state}_activity') < self._get_parameter_symbol('activity', namespace),
self._draw_symbolic(namespace),
state,
)
return [UpdateLaw(state, revised)]
Equations
neurons.spin[i]: spin_neurons(t + 1) == where(uniform_spin_neurons_activity_glauber < activity_glauber, where(where(categorical_spin_neurons_option_glauber*(e^(1.0*_reduced_0*beta_neurons*intensity_glauber) + e^(-1.0*_reduced_0*beta_neurons*intensity_glauber)) >= e^(-1.0*_reduced_0*beta_neurons*intensity_glauber), 1, 0) == 0, -1.0, 0) + where(where(categorical_spin_neurons_option_glauber*(e^(1.0*_reduced_0*beta_neurons*intensity_glauber) + e^(-1.0*_reduced_0*beta_neurons*intensity_glauber)) >= e^(-1.0*_reduced_0*beta_neurons*intensity_glauber), 1, 0) == 1, 1.0, 0), spin_neurons(t))