omnibias.torch.confluent_bank
PyTorch activation derivatives, trainable operator layers, network jets, and direct differential-operator contractions.
import omnibias.torch.confluent_bankModule contract & mathematical conventions
Opt-in preallocated confluent refinement bank and atomic structural commits.
Legacy AdaptivePackBank behavior is preserved. This subclass adds stable IDs, representation modes, and optimizer-aware transactions without replacing any Parameter object. Call commit only between optimizer steps, after zero_grad.
Shared formulas do not imply universal floating-point identity. Verify dtype, shape, derivative-order support, and the stated numerical assumptions for the path you use.
CLASS
ConfluentPackBank
omnibias.torch.confluent_bank.ConfluentPackBankSource ↗class ConfluentPackBank(AdaptivePackBank)Ordinary (0), centered-pair (1), and moment/derivative (2) slots.
A pair uses weights=m0, odd_moments=m1, rho=h**2 and eta=v*x+c. A moment slot uses Taylor-normalized moment_coefficients. Legacy state dictionaries migrate additively to ordinary mode. Slot IDs never recycle; activating a formerly inactive slot resets its optimizer moments.
Parameters
| Name | Type & default | Contract |
|---|---|---|
packspositional-or-keyword | Sequence[RefinedPack]Required | See the operation contract and implementation below. |
max_packskeyword-only | intDefault: 16 | See the operation contract and implementation below. |
max_orderkeyword-only | intDefault: 8 | See the operation contract and implementation below. |
basekeyword-only | strDefault: 'sigmoid' | See the operation contract and implementation below. |
dtypekeyword-only | torch.dtype | NoneDefault: None | See the operation contract and implementation below. |
Regression example · one registered tower per slot and correct confluent norms
This example belongs to the pinned upstream test suite. It may use fixtures and imports from its full file. Run it from that checkout with its test dependencies installed.
def test_one_registered_tower_per_slot_and_correct_confluent_norms(monkeypatch) -> None:
torch_bank, bank = pair_banks()
torch_bank.modes[0] = 1
torch_bank.modes[1] = 2
torch_bank.moment_coefficients.data[1, 1] = 0.7
bank = replace(bank, modes=bank.modes.at[:2].set(jnp.array([1, 2])),
moment_coefficients=bank.moment_coefficients.at[1, 1].set(0.7))
spec = torch_bank.act_spec
calls = []
def tower(z, n):
calls.append(n)
return spec.tower(z, n)
torch_bank.act_spec = replace(spec, tower=tower)
x = torch.tensor([-0.3, 0.2], dtype=torch.float64)
terms = torch_bank.slot_terms(x)
assert calls == [11] * torch_bank.max_packs
np.testing.assert_allclose(np.asarray(bank.slot_terms(jnp.asarray(x.numpy()))), terms.detach().numpy(), atol=3e-16)
expected = [float(terms[i].detach().square().sum().sqrt()) for i in range(3)]
assert torch_bank.pack_term_norms(x) == expected
with pytest.raises(RuntimeError, match="RefinedPack"):
torch_bank.active_packs()
with pytest.raises(RuntimeError, match="transactional"):
torch_bank.sync_from([])
jspec = jb.get_activation(bank.base)
jcalls = []
def jtower(z, n):
jcalls.append(n)
return jspec.tower(z, n)
monkeypatch.setattr(jb, "get_activation", lambda _name: replace(jspec, tower=jtower))
bank.forward(jnp.asarray(x.numpy()))
assert jcalls == [11] * bank.max_packspython -m pytest tests/test_confluent_bank_parity.py -k test_one_registered_tower_per_slot_and_correct_confluent_norms -qRegression example · cluster row transition and error gated derivative
This example belongs to the pinned upstream test suite. It may use fixtures and imports from its full file. Run it from that checkout with its test dependencies installed.
def test_cluster_row_transition_and_error_gated_derivative() -> None:
packs = [RefinedPack(0, v, w) for v, w in [(-0.01, 0.7), (0.0, -0.2), (0.02, 0.5)]]
bank = jb.init_confluent_bank(packs, max_packs=4, max_order=4)
torch_bank = tb.ConfluentPackBank(packs, max_packs=4, max_order=4, dtype=torch.float64)
proposal = bank.propose_cluster((0, 1, 2), order=3, error_budget=1e-7)
assert proposal == torch_bank.propose_cluster((0, 1, 2), order=3, error_budget=1e-7)
assert TransitionProposal.from_json(proposal.to_json()) == proposal
candidate, _, accepted = jb.commit_transition(bank, proposal, moments(bank), between_steps=True,
acceptance=lambda _b, _p: True)
assert accepted and int(candidate.modes[0]) == 2
assert candidate.moment_coefficients.shape == (4, 5)
assert not bool(candidate.active[1]) and not bool(candidate.active[2])
np.testing.assert_allclose(np.asarray(candidate.forward(jnp.array([-0.2, 0.4]))),
np.asarray(bank.forward(jnp.array([-0.2, 0.4]))), atol=2e-9)
_, pair = pair_banks()
pair, state, _ = jb.commit_transition(pair, pair.propose_pair(0, 1), moments(pair), between_steps=True,
acceptance=lambda _b, _p: True)
derivative = pair.propose_derivative(0, error_budget=0)
same, same_state, accepted = jb.commit_transition(pair, derivative, state, between_steps=True,
acceptance=lambda _b, p: p.error_budget > 0)
assert not accepted and same is pair and same_state is state
assert float(pair.rho[0]) > 0python -m pytest tests/test_confluent_bank_parity.py -k test_cluster_row_transition_and_error_gated_derivative -qFields & methods
rhofield
rho: TensorDeclared Tensor field on ConfluentPackBank. A value must be provided when constructing the configuration, unless an inherited constructor supplies it.
Regression example · zero output birth reuses slot and resets all moments
This example belongs to the pinned upstream test suite. It may use fixtures and imports from its full file. Run it from that checkout with its test dependencies installed.
def test_zero_output_birth_reuses_slot_and_resets_all_moments() -> None:
torch_bank, bank = pair_banks()
# Simulate an inactive slot retaining an obsolete collision representation.
torch_bank.modes[3] = 2
torch_bank.moment_coefficients.data[3] = 9
torch_bank.rho.data[3] = 0.2
bank = replace(bank, modes=bank.modes.at[3].set(2), rho=bank.rho.at[3].set(0.2),
moment_coefficients=bank.moment_coefficients.at[3].set(9))
pack = RefinedPack(2, 0.4, 0, 1.3)
proposal = bank.propose_birth(3, pack)
assert proposal == torch_bank.propose_birth(3, pack)
state = moments(bank)
candidate, next_state, accepted = jb.commit_transition(bank, proposal, state, between_steps=True,
acceptance=lambda _b, _p: True)
optimizer = torch.optim.Adam(torch_bank.parameters())
for parameter in torch_bank.parameters():
optimizer.state[parameter] = {"step": torch.tensor(7.0), "exp_avg": torch.ones_like(parameter),
"exp_avg_sq": 2 * torch.ones_like(parameter)}
assert tb.commit_transition(torch_bank, proposal, optimizer, between_steps=True,
acceptance=lambda _b, _p: True)
assert accepted and candidate.portable_snapshot() == torch_bank.portable_snapshot()
assert int(candidate.slot_ids[3]) == int(bank.slot_ids[3])
assert bool(candidate.active[3]) and int(candidate.modes[3]) == 0
assert bool(jnp.all(candidate.moment_coefficients[3] == 0))
for entries in next_state.moments.values():
assert all(bool(jnp.all(value[3] == 0)) for value in entries.values())
x = jnp.array([-0.2, 0.3])
np.testing.assert_array_equal(np.asarray(candidate.forward(x)), np.asarray(bank.forward(x)))
with pytest.raises(ValueError, match="inactive"):
candidate.propose_birth(3, pack)
with pytest.raises(ValueError, match="zero output"):
bank.propose_birth(3, RefinedPack(1, 0.2, 1))python -m pytest tests/test_confluent_bank_parity.py -k test_zero_output_birth_reuses_slot_and_resets_all_moments -qodd_momentsfield
odd_moments: TensorDeclared Tensor field on ConfluentPackBank. A value must be provided when constructing the configuration, unless an inherited constructor supplies it.
View source ↗directionsfield
directions: TensorDeclared Tensor field on ConfluentPackBank. A value must be provided when constructing the configuration, unless an inherited constructor supplies it.
View source ↗direction_biasfield
direction_bias: TensorDeclared Tensor field on ConfluentPackBank. A value must be provided when constructing the configuration, unless an inherited constructor supplies it.
View source ↗moment_coefficientsfield
moment_coefficients: TensorDeclared Tensor field on ConfluentPackBank. A value must be provided when constructing the configuration, unless an inherited constructor supplies it.
Regression example · one registered tower per slot and correct confluent norms
This example belongs to the pinned upstream test suite. It may use fixtures and imports from its full file. Run it from that checkout with its test dependencies installed.
def test_one_registered_tower_per_slot_and_correct_confluent_norms(monkeypatch) -> None:
torch_bank, bank = pair_banks()
torch_bank.modes[0] = 1
torch_bank.modes[1] = 2
torch_bank.moment_coefficients.data[1, 1] = 0.7
bank = replace(bank, modes=bank.modes.at[:2].set(jnp.array([1, 2])),
moment_coefficients=bank.moment_coefficients.at[1, 1].set(0.7))
spec = torch_bank.act_spec
calls = []
def tower(z, n):
calls.append(n)
return spec.tower(z, n)
torch_bank.act_spec = replace(spec, tower=tower)
x = torch.tensor([-0.3, 0.2], dtype=torch.float64)
terms = torch_bank.slot_terms(x)
assert calls == [11] * torch_bank.max_packs
np.testing.assert_allclose(np.asarray(bank.slot_terms(jnp.asarray(x.numpy()))), terms.detach().numpy(), atol=3e-16)
expected = [float(terms[i].detach().square().sum().sqrt()) for i in range(3)]
assert torch_bank.pack_term_norms(x) == expected
with pytest.raises(RuntimeError, match="RefinedPack"):
torch_bank.active_packs()
with pytest.raises(RuntimeError, match="transactional"):
torch_bank.sync_from([])
jspec = jb.get_activation(bank.base)
jcalls = []
def jtower(z, n):
jcalls.append(n)
return jspec.tower(z, n)
monkeypatch.setattr(jb, "get_activation", lambda _name: replace(jspec, tower=jtower))
bank.forward(jnp.asarray(x.numpy()))
assert jcalls == [11] * bank.max_packspython -m pytest tests/test_confluent_bank_parity.py -k test_one_registered_tower_per_slot_and_correct_confluent_norms -qmodesfield
modes: TensorDeclared Tensor field on ConfluentPackBank. A value must be provided when constructing the configuration, unless an inherited constructor supplies it.
Regression example · one registered tower per slot and correct confluent norms
This example belongs to the pinned upstream test suite. It may use fixtures and imports from its full file. Run it from that checkout with its test dependencies installed.
def test_one_registered_tower_per_slot_and_correct_confluent_norms(monkeypatch) -> None:
torch_bank, bank = pair_banks()
torch_bank.modes[0] = 1
torch_bank.modes[1] = 2
torch_bank.moment_coefficients.data[1, 1] = 0.7
bank = replace(bank, modes=bank.modes.at[:2].set(jnp.array([1, 2])),
moment_coefficients=bank.moment_coefficients.at[1, 1].set(0.7))
spec = torch_bank.act_spec
calls = []
def tower(z, n):
calls.append(n)
return spec.tower(z, n)
torch_bank.act_spec = replace(spec, tower=tower)
x = torch.tensor([-0.3, 0.2], dtype=torch.float64)
terms = torch_bank.slot_terms(x)
assert calls == [11] * torch_bank.max_packs
np.testing.assert_allclose(np.asarray(bank.slot_terms(jnp.asarray(x.numpy()))), terms.detach().numpy(), atol=3e-16)
expected = [float(terms[i].detach().square().sum().sqrt()) for i in range(3)]
assert torch_bank.pack_term_norms(x) == expected
with pytest.raises(RuntimeError, match="RefinedPack"):
torch_bank.active_packs()
with pytest.raises(RuntimeError, match="transactional"):
torch_bank.sync_from([])
jspec = jb.get_activation(bank.base)
jcalls = []
def jtower(z, n):
jcalls.append(n)
return jspec.tower(z, n)
monkeypatch.setattr(jb, "get_activation", lambda _name: replace(jspec, tower=jtower))
bank.forward(jnp.asarray(x.numpy()))
assert jcalls == [11] * bank.max_packspython -m pytest tests/test_confluent_bank_parity.py -k test_one_registered_tower_per_slot_and_correct_confluent_norms -qslot_idsfield
slot_ids: TensorDeclared Tensor field on ConfluentPackBank. A value must be provided when constructing the configuration, unless an inherited constructor supplies it.
View source ↗transition_versionfield
transition_version: TensorDeclared Tensor field on ConfluentPackBank. A value must be provided when constructing the configuration, unless an inherited constructor supplies it.
View source ↗__init__method
__init__(self, packs: Sequence[RefinedPack], *, max_packs: int=16, max_order: int=8, base: str='sigmoid', dtype: torch.dtype | None=None) -> NoneSee the class contract above and the exact method implementation below.
Parameters
| Name | Type & default | Contract |
|---|---|---|
packspositional-or-keyword | Sequence[RefinedPack]Required | See the operation contract and implementation below. |
max_packskeyword-only | intDefault: 16 | See the operation contract and implementation below. |
max_orderkeyword-only | intDefault: 8 | See the operation contract and implementation below. |
basekeyword-only | strDefault: 'sigmoid' | See the operation contract and implementation below. |
dtypekeyword-only | torch.dtype | NoneDefault: None | See the operation contract and implementation below. |
Returns: None
Validation & failure cases
- When
base not in ('sigmoid', 'tanh'), raisesValueError('the confluent bank supports sigmoid and tanh').
Method implementation
def __init__(self, packs: Sequence[RefinedPack], *, max_packs: int = 16,
max_order: int = 8, base: str = 'sigmoid',
dtype: torch.dtype | None = None) -> None:
if base not in ('sigmoid', 'tanh'):
raise ValueError('the confluent bank supports sigmoid and tanh')
super().__init__(packs, max_packs=max_packs, max_order=max_order, base=base, dtype=dtype)
dt = self.centers.dtype
self.rho = nn.Parameter(torch.zeros(max_packs, dtype=dt))
self.odd_moments = nn.Parameter(torch.zeros(max_packs, dtype=dt))
self.directions = nn.Parameter(torch.zeros(max_packs, dtype=dt))
self.direction_bias = nn.Parameter(torch.ones(max_packs, dtype=dt))
self.moment_coefficients = nn.Parameter(torch.zeros((max_packs, max_order+1), dtype=dt))
self.register_buffer('modes', torch.zeros(max_packs, dtype=torch.long))
self.register_buffer('slot_ids', torch.arange(max_packs, dtype=torch.long))
self.register_buffer('transition_version', torch.tensor(0, dtype=torch.long))forwardmethod
forward(self, z: Tensor) -> TensorSee the class contract above and the exact method implementation below.
Parameters
| Name | Type & default | Contract |
|---|---|---|
zpositional-or-keyword | TensorRequired | See the operation contract and implementation below. |
Returns: Tensor
Method implementation
def forward(self, z: Tensor) -> Tensor:
terms = self.slot_terms(z)
result = torch.zeros_like(z)
for i in range(self.max_packs):
result = result + terms[i]
return resultslot_termsmethod
slot_terms(self, z: Tensor) -> TensorActual ordinary/pair/moment contributions, including zero inactive rows.
One registered tower per allocated slot serves all three branches. The pair's stable finite formula separately evaluates three sigmoids.
Parameters
| Name | Type & default | Contract |
|---|---|---|
zpositional-or-keyword | TensorRequired | See the operation contract and implementation below. |
Returns: Tensor
Regression example · one registered tower per slot and correct confluent norms
This example belongs to the pinned upstream test suite. It may use fixtures and imports from its full file. Run it from that checkout with its test dependencies installed.
def test_one_registered_tower_per_slot_and_correct_confluent_norms(monkeypatch) -> None:
torch_bank, bank = pair_banks()
torch_bank.modes[0] = 1
torch_bank.modes[1] = 2
torch_bank.moment_coefficients.data[1, 1] = 0.7
bank = replace(bank, modes=bank.modes.at[:2].set(jnp.array([1, 2])),
moment_coefficients=bank.moment_coefficients.at[1, 1].set(0.7))
spec = torch_bank.act_spec
calls = []
def tower(z, n):
calls.append(n)
return spec.tower(z, n)
torch_bank.act_spec = replace(spec, tower=tower)
x = torch.tensor([-0.3, 0.2], dtype=torch.float64)
terms = torch_bank.slot_terms(x)
assert calls == [11] * torch_bank.max_packs
np.testing.assert_allclose(np.asarray(bank.slot_terms(jnp.asarray(x.numpy()))), terms.detach().numpy(), atol=3e-16)
expected = [float(terms[i].detach().square().sum().sqrt()) for i in range(3)]
assert torch_bank.pack_term_norms(x) == expected
with pytest.raises(RuntimeError, match="RefinedPack"):
torch_bank.active_packs()
with pytest.raises(RuntimeError, match="transactional"):
torch_bank.sync_from([])
jspec = jb.get_activation(bank.base)
jcalls = []
def jtower(z, n):
jcalls.append(n)
return jspec.tower(z, n)
monkeypatch.setattr(jb, "get_activation", lambda _name: replace(jspec, tower=jtower))
bank.forward(jnp.asarray(x.numpy()))
assert jcalls == [11] * bank.max_packspython -m pytest tests/test_confluent_bank_parity.py -k test_one_registered_tower_per_slot_and_correct_confluent_norms -qMethod implementation
def slot_terms(self, z: Tensor) -> Tensor:
"""Actual ordinary/pair/moment contributions, including zero inactive rows.
One registered tower per allocated slot serves all three branches.
The pair's stable finite formula separately evaluates three sigmoids.
"""
terms = []
for i in range(self.max_packs):
argument = self.scales[i] * (z-self.centers[i])
tower = _sigma_tower(self.act_spec, argument, max(self.max_order, 11))
ordinary = torch.zeros_like(z)
for order in range(self.max_order+1):
value = self.weights[i] * (self.scales[i] ** order) * tower[order]
ordinary = ordinary + torch.where(self.orders[i] == order, value, torch.zeros_like(z))
pair = collision_atom(argument, self.rho[i], self.directions[i]*z+self.direction_bias[i],
self.weights[i], self.odd_moments[i], activation=self.act_spec.name,
tower=tower)
moments = moment_atom(argument, self.moment_coefficients[i], activation=self.act_spec.name,
tower=tower)
value = torch.where(self.modes[i] == 0, ordinary,
torch.where(self.modes[i] == 1, pair, moments))
terms.append(torch.where(self.active[i], value, torch.zeros_like(z)))
return torch.stack(terms)pack_term_normsmethod
pack_term_norms(self, z: Tensor) -> list[float]Host L2 diagnostics of actual live contributions in slot order.
Parameters
| Name | Type & default | Contract |
|---|---|---|
zpositional-or-keyword | TensorRequired | See the operation contract and implementation below. |
Returns: list[float]
Regression example · one registered tower per slot and correct confluent norms
This example belongs to the pinned upstream test suite. It may use fixtures and imports from its full file. Run it from that checkout with its test dependencies installed.
def test_one_registered_tower_per_slot_and_correct_confluent_norms(monkeypatch) -> None:
torch_bank, bank = pair_banks()
torch_bank.modes[0] = 1
torch_bank.modes[1] = 2
torch_bank.moment_coefficients.data[1, 1] = 0.7
bank = replace(bank, modes=bank.modes.at[:2].set(jnp.array([1, 2])),
moment_coefficients=bank.moment_coefficients.at[1, 1].set(0.7))
spec = torch_bank.act_spec
calls = []
def tower(z, n):
calls.append(n)
return spec.tower(z, n)
torch_bank.act_spec = replace(spec, tower=tower)
x = torch.tensor([-0.3, 0.2], dtype=torch.float64)
terms = torch_bank.slot_terms(x)
assert calls == [11] * torch_bank.max_packs
np.testing.assert_allclose(np.asarray(bank.slot_terms(jnp.asarray(x.numpy()))), terms.detach().numpy(), atol=3e-16)
expected = [float(terms[i].detach().square().sum().sqrt()) for i in range(3)]
assert torch_bank.pack_term_norms(x) == expected
with pytest.raises(RuntimeError, match="RefinedPack"):
torch_bank.active_packs()
with pytest.raises(RuntimeError, match="transactional"):
torch_bank.sync_from([])
jspec = jb.get_activation(bank.base)
jcalls = []
def jtower(z, n):
jcalls.append(n)
return jspec.tower(z, n)
monkeypatch.setattr(jb, "get_activation", lambda _name: replace(jspec, tower=jtower))
bank.forward(jnp.asarray(x.numpy()))
assert jcalls == [11] * bank.max_packspython -m pytest tests/test_confluent_bank_parity.py -k test_one_registered_tower_per_slot_and_correct_confluent_norms -qMethod implementation
def pack_term_norms(self, z: Tensor) -> list[float]:
"""Host L2 diagnostics of actual live contributions in slot order."""
terms = self.slot_terms(z).detach()
return [float(terms[i].square().sum().sqrt()) for i in range(self.max_packs) if bool(self.active[i])]active_packsmethod
active_packs(self) -> list[RefinedPack]See the class contract above and the exact method implementation below.
Returns: list[RefinedPack]
Validation & failure cases
- When
bool(((self.modes != 0) & self.active).any()), raisesRuntimeError('confluent modes cannot be represented as RefinedPack; use slot_terms or portable_snapshot').
Regression example · one registered tower per slot and correct confluent norms
This example belongs to the pinned upstream test suite. It may use fixtures and imports from its full file. Run it from that checkout with its test dependencies installed.
def test_one_registered_tower_per_slot_and_correct_confluent_norms(monkeypatch) -> None:
torch_bank, bank = pair_banks()
torch_bank.modes[0] = 1
torch_bank.modes[1] = 2
torch_bank.moment_coefficients.data[1, 1] = 0.7
bank = replace(bank, modes=bank.modes.at[:2].set(jnp.array([1, 2])),
moment_coefficients=bank.moment_coefficients.at[1, 1].set(0.7))
spec = torch_bank.act_spec
calls = []
def tower(z, n):
calls.append(n)
return spec.tower(z, n)
torch_bank.act_spec = replace(spec, tower=tower)
x = torch.tensor([-0.3, 0.2], dtype=torch.float64)
terms = torch_bank.slot_terms(x)
assert calls == [11] * torch_bank.max_packs
np.testing.assert_allclose(np.asarray(bank.slot_terms(jnp.asarray(x.numpy()))), terms.detach().numpy(), atol=3e-16)
expected = [float(terms[i].detach().square().sum().sqrt()) for i in range(3)]
assert torch_bank.pack_term_norms(x) == expected
with pytest.raises(RuntimeError, match="RefinedPack"):
torch_bank.active_packs()
with pytest.raises(RuntimeError, match="transactional"):
torch_bank.sync_from([])
jspec = jb.get_activation(bank.base)
jcalls = []
def jtower(z, n):
jcalls.append(n)
return jspec.tower(z, n)
monkeypatch.setattr(jb, "get_activation", lambda _name: replace(jspec, tower=jtower))
bank.forward(jnp.asarray(x.numpy()))
assert jcalls == [11] * bank.max_packspython -m pytest tests/test_confluent_bank_parity.py -k test_one_registered_tower_per_slot_and_correct_confluent_norms -qMethod implementation
def active_packs(self) -> list[RefinedPack]:
if bool(((self.modes != 0) & self.active).any()):
raise RuntimeError('confluent modes cannot be represented as RefinedPack; use slot_terms or portable_snapshot')
return super().active_packs()sync_frommethod
sync_from(self, packs: Sequence[RefinedPack]) -> NoneSee the class contract above and the exact method implementation below.
Parameters
| Name | Type & default | Contract |
|---|---|---|
packspositional-or-keyword | Sequence[RefinedPack]Required | See the operation contract and implementation below. |
Returns: None
Regression example · one registered tower per slot and correct confluent norms
This example belongs to the pinned upstream test suite. It may use fixtures and imports from its full file. Run it from that checkout with its test dependencies installed.
def test_one_registered_tower_per_slot_and_correct_confluent_norms(monkeypatch) -> None:
torch_bank, bank = pair_banks()
torch_bank.modes[0] = 1
torch_bank.modes[1] = 2
torch_bank.moment_coefficients.data[1, 1] = 0.7
bank = replace(bank, modes=bank.modes.at[:2].set(jnp.array([1, 2])),
moment_coefficients=bank.moment_coefficients.at[1, 1].set(0.7))
spec = torch_bank.act_spec
calls = []
def tower(z, n):
calls.append(n)
return spec.tower(z, n)
torch_bank.act_spec = replace(spec, tower=tower)
x = torch.tensor([-0.3, 0.2], dtype=torch.float64)
terms = torch_bank.slot_terms(x)
assert calls == [11] * torch_bank.max_packs
np.testing.assert_allclose(np.asarray(bank.slot_terms(jnp.asarray(x.numpy()))), terms.detach().numpy(), atol=3e-16)
expected = [float(terms[i].detach().square().sum().sqrt()) for i in range(3)]
assert torch_bank.pack_term_norms(x) == expected
with pytest.raises(RuntimeError, match="RefinedPack"):
torch_bank.active_packs()
with pytest.raises(RuntimeError, match="transactional"):
torch_bank.sync_from([])
jspec = jb.get_activation(bank.base)
jcalls = []
def jtower(z, n):
jcalls.append(n)
return jspec.tower(z, n)
monkeypatch.setattr(jb, "get_activation", lambda _name: replace(jspec, tower=jtower))
bank.forward(jnp.asarray(x.numpy()))
assert jcalls == [11] * bank.max_packspython -m pytest tests/test_confluent_bank_parity.py -k test_one_registered_tower_per_slot_and_correct_confluent_norms -qMethod implementation
def sync_from(self, packs: Sequence[RefinedPack]) -> None:
raise RuntimeError('confluent banks require transactional proposals and commit_transition')propose_birthmethod
propose_birth(self, slot: int, pack: RefinedPack) -> TransitionProposalBirth/reuse an inactive slot with exactly zero output and fresh moments.
Parameters
| Name | Type & default | Contract |
|---|---|---|
slotpositional-or-keyword | intRequired | See the operation contract and implementation below. |
packpositional-or-keyword | RefinedPackRequired | See the operation contract and implementation below. |
Returns: TransitionProposal
Validation & failure cases
- When
slot < 0 or slot >= self.max_packs or bool(self.active[slot]), raisesValueError('birth requires an allocated inactive slot'). - When
pack.weight != 0 or pack.order > self.max_order, raisesValueError('birth requires zero output weight and an order within budget').
Regression example · zero output birth reuses slot and resets all moments
This example belongs to the pinned upstream test suite. It may use fixtures and imports from its full file. Run it from that checkout with its test dependencies installed.
def test_zero_output_birth_reuses_slot_and_resets_all_moments() -> None:
torch_bank, bank = pair_banks()
# Simulate an inactive slot retaining an obsolete collision representation.
torch_bank.modes[3] = 2
torch_bank.moment_coefficients.data[3] = 9
torch_bank.rho.data[3] = 0.2
bank = replace(bank, modes=bank.modes.at[3].set(2), rho=bank.rho.at[3].set(0.2),
moment_coefficients=bank.moment_coefficients.at[3].set(9))
pack = RefinedPack(2, 0.4, 0, 1.3)
proposal = bank.propose_birth(3, pack)
assert proposal == torch_bank.propose_birth(3, pack)
state = moments(bank)
candidate, next_state, accepted = jb.commit_transition(bank, proposal, state, between_steps=True,
acceptance=lambda _b, _p: True)
optimizer = torch.optim.Adam(torch_bank.parameters())
for parameter in torch_bank.parameters():
optimizer.state[parameter] = {"step": torch.tensor(7.0), "exp_avg": torch.ones_like(parameter),
"exp_avg_sq": 2 * torch.ones_like(parameter)}
assert tb.commit_transition(torch_bank, proposal, optimizer, between_steps=True,
acceptance=lambda _b, _p: True)
assert accepted and candidate.portable_snapshot() == torch_bank.portable_snapshot()
assert int(candidate.slot_ids[3]) == int(bank.slot_ids[3])
assert bool(candidate.active[3]) and int(candidate.modes[3]) == 0
assert bool(jnp.all(candidate.moment_coefficients[3] == 0))
for entries in next_state.moments.values():
assert all(bool(jnp.all(value[3] == 0)) for value in entries.values())
x = jnp.array([-0.2, 0.3])
np.testing.assert_array_equal(np.asarray(candidate.forward(x)), np.asarray(bank.forward(x)))
with pytest.raises(ValueError, match="inactive"):
candidate.propose_birth(3, pack)
with pytest.raises(ValueError, match="zero output"):
bank.propose_birth(3, RefinedPack(1, 0.2, 1))python -m pytest tests/test_confluent_bank_parity.py -k test_zero_output_birth_reuses_slot_and_resets_all_moments -qMethod implementation
def propose_birth(self, slot: int, pack: RefinedPack) -> TransitionProposal:
"""Birth/reuse an inactive slot with exactly zero output and fresh moments."""
if slot < 0 or slot >= self.max_packs or bool(self.active[slot]):
raise ValueError('birth requires an allocated inactive slot')
if pack.weight != 0 or pack.order > self.max_order:
raise ValueError('birth requires zero output weight and an order within budget')
updates = (SlotUpdate('centers', slot, pack.center), SlotUpdate('weights', slot, 0),
SlotUpdate('scales', slot, pack.scale), SlotUpdate('orders', slot, pack.order),
SlotUpdate('ages', slot, pack.age), SlotUpdate('active', slot, 1),
SlotUpdate('modes', slot, 0), SlotUpdate('rho', slot, 0),
SlotUpdate('odd_moments', slot, 0), SlotUpdate('directions', slot, 0),
SlotUpdate('direction_bias', slot, 1),
SlotUpdate('moment_coefficients', slot, (0.0,)*(self.max_order+1)))
return TransitionProposal(snapshot_digest(self.portable_snapshot()), int(self.transition_version),
(int(self.slot_ids[slot]),), updates, 'zero_output_birth',
'exact_zero_output_with_full_slot_state_initialization')retractmethod
retract(self) -> NoneBoundary projection after an optimizer step; attained rho=0 is retained.
Returns: None
Method implementation
def retract(self) -> None:
"""Boundary projection after an optimizer step; attained rho=0 is retained."""
with torch.no_grad():
self.rho.clamp_(min=0)portable_snapshotmethod
portable_snapshot(self) -> dict[str, Any]See the class contract above and the exact method implementation below.
Returns: dict[str, Any]
Regression example · rollback and explicit optimizer adapter
This example belongs to the pinned upstream test suite. It may use fixtures and imports from its full file. Run it from that checkout with its test dependencies installed.
def test_rollback_and_explicit_optimizer_adapter() -> None:
torch_bank, bank = pair_banks()
state = moments(bank)
proposal = bank.propose_pair(0, 1)
original, original_state, accepted = jb.commit_transition(bank, proposal, state, between_steps=True,
acceptance=lambda _b, _p: True,
postcondition=lambda _b: False)
assert not accepted and original is bank and original_state is state
with pytest.raises(TypeError, match="explicit"):
jb.commit_transition(bank, proposal, {"custom": 3}, between_steps=True,
acceptance=lambda _b, _p: True)
calls = []
def adapter(custom, candidate, slots):
calls.append((candidate, slots))
return {"custom": custom["custom"] + 1}
_, custom, accepted = jb.commit_transition(bank, proposal, {"custom": 3}, between_steps=True,
acceptance=lambda _b, _p: True, state_adapter=adapter)
assert accepted and custom == {"custom": 4} and calls[0][1] == (0, 1)
with pytest.raises(RuntimeError, match="between"):
jb.commit_transition(bank, proposal, state, between_steps=False, acceptance=lambda _b, _p: True)
optimizer = torch.optim.Adam(torch_bank.parameters())
snapshot = torch_bank.portable_snapshot()
assert not tb.commit_transition(torch_bank, proposal, optimizer, between_steps=True,
acceptance=lambda _b, _p: True, postcondition=lambda _b: False)
assert torch_bank.portable_snapshot() == snapshot
assert len(optimizer.state) == 0python -m pytest tests/test_confluent_bank_parity.py -k test_rollback_and_explicit_optimizer_adapter -qMethod implementation
def portable_snapshot(self) -> dict[str, Any]:
snapshot: dict[str, Any] = {name: {'shape': list(value.shape), 'dtype': str(value.dtype).removeprefix('torch.'),
'values': value.detach().cpu().reshape(-1).tolist()}
for name, value in self.state_dict().items()}
snapshot['__configuration__'] = {'activation': self.act_spec.name, 'max_order': self.max_order,
'max_packs': self.max_packs, 'schema_version': 1,
'series_terms': 6, 'series_radius': 0.01}
return snapshotpropose_clustermethod
propose_cluster(self, slots: tuple[int, ...], *, order: int, error_budget: float) -> TransitionProposalCommon-weight ordinary cluster -> bounded truncated moment expansion.
Parameters
| Name | Type & default | Contract |
|---|---|---|
slotspositional-or-keyword | tuple[int, ...]Required | See the operation contract and implementation below. |
orderkeyword-only | intRequired | See the operation contract and implementation below. |
error_budgetkeyword-only | floatRequired | See the operation contract and implementation below. |
Returns: TransitionProposal
Validation & failure cases
- When
not slots or len(set(slots)) != len(slots) or order < 0 or (order > self.max_order) or any((i < 0 or i >= self.max_packs for i in slots)), raisesValueError('valid distinct slots and an order within the bank budget required'). - When
scale == 0, raisesValueError('nonzero common weight required'). - When
not bool(self.active[i]) or int(self.modes[i]) != 0 or int(self.orders[i]) != 0 or (Fraction(float(self.scales[i].detach())) != scale), raisesValueError('cluster requires ordinary order-zero atoms with a common weight').
Regression example · cluster row transition and error gated derivative
This example belongs to the pinned upstream test suite. It may use fixtures and imports from its full file. Run it from that checkout with its test dependencies installed.
def test_cluster_row_transition_and_error_gated_derivative() -> None:
packs = [RefinedPack(0, v, w) for v, w in [(-0.01, 0.7), (0.0, -0.2), (0.02, 0.5)]]
bank = jb.init_confluent_bank(packs, max_packs=4, max_order=4)
torch_bank = tb.ConfluentPackBank(packs, max_packs=4, max_order=4, dtype=torch.float64)
proposal = bank.propose_cluster((0, 1, 2), order=3, error_budget=1e-7)
assert proposal == torch_bank.propose_cluster((0, 1, 2), order=3, error_budget=1e-7)
assert TransitionProposal.from_json(proposal.to_json()) == proposal
candidate, _, accepted = jb.commit_transition(bank, proposal, moments(bank), between_steps=True,
acceptance=lambda _b, _p: True)
assert accepted and int(candidate.modes[0]) == 2
assert candidate.moment_coefficients.shape == (4, 5)
assert not bool(candidate.active[1]) and not bool(candidate.active[2])
np.testing.assert_allclose(np.asarray(candidate.forward(jnp.array([-0.2, 0.4]))),
np.asarray(bank.forward(jnp.array([-0.2, 0.4]))), atol=2e-9)
_, pair = pair_banks()
pair, state, _ = jb.commit_transition(pair, pair.propose_pair(0, 1), moments(pair), between_steps=True,
acceptance=lambda _b, _p: True)
derivative = pair.propose_derivative(0, error_budget=0)
same, same_state, accepted = jb.commit_transition(pair, derivative, state, between_steps=True,
acceptance=lambda _b, p: p.error_budget > 0)
assert not accepted and same is pair and same_state is state
assert float(pair.rho[0]) > 0python -m pytest tests/test_confluent_bank_parity.py -k test_cluster_row_transition_and_error_gated_derivative -qMethod implementation
def propose_cluster(self, slots: tuple[int, ...], *, order: int,
error_budget: float) -> TransitionProposal:
"""Common-weight ordinary cluster -> bounded truncated moment expansion."""
if (not slots or len(set(slots)) != len(slots) or order < 0 or order > self.max_order
or any(i < 0 or i >= self.max_packs for i in slots)):
raise ValueError('valid distinct slots and an order within the bank budget required')
first = slots[0]
scale = Fraction(float(self.scales[first].detach()))
if scale == 0:
raise ValueError('nonzero common weight required')
for i in slots:
if (not bool(self.active[i]) or int(self.modes[i]) != 0 or int(self.orders[i]) != 0
or Fraction(float(self.scales[i].detach())) != scale):
raise ValueError('cluster requires ordinary order-zero atoms with a common weight')
centers = tuple(Fraction(float(self.centers[i].detach())) for i in slots)
center = sum(centers, Fraction())/len(slots)
moments = initialize_moments(tuple(-scale*c for c in centers),
tuple(float(self.weights[i].detach()) for i in slots),
center=-scale*center, order=order)
row = moments.moments+(0.0,)*(self.max_order-order)
updates = [SlotUpdate('centers', first, float(center)), SlotUpdate('modes', first, 2),
SlotUpdate('orders', first, order), SlotUpdate('moment_coefficients', first, row)]
for i in slots[1:]:
updates.extend((SlotUpdate('active', i, 0), SlotUpdate('weights', i, 0)))
return TransitionProposal(snapshot_digest(self.portable_snapshot()), int(self.transition_version),
tuple(int(self.slot_ids[i]) for i in slots), tuple(updates),
'cluster_to_moments', 'finite-spread truncation with explicit remainder',
error_budget)propose_pairmethod
propose_pair(self, first: int, second: int, *, error_budget: float=0.0) -> TransitionProposalOrdinary common-weight bias pair -> stable moments (host planning).
Parameters
| Name | Type & default | Contract |
|---|---|---|
firstpositional-or-keyword | intRequired | See the operation contract and implementation below. |
secondpositional-or-keyword | intRequired | See the operation contract and implementation below. |
error_budgetkeyword-only | floatDefault: 0.0 | See the operation contract and implementation below. |
Returns: TransitionProposal
Validation & failure cases
- When
first == second or any((i < 0 or i >= self.max_packs for i in (first, second))), raisesValueError('two distinct live slots required'). - When
not bool(self.active[i]) or int(self.modes[i]) != 0 or int(self.orders[i]) != 0, raisesValueError('pair proposal requires ordinary order-zero live slots'). - When
scale != Fraction(float(self.scales[second].detach())) or scale == 0, raisesValueError('bias confluence requires identical nonzero weights').
Regression example · portable proposals and functional transaction match torch
This example belongs to the pinned upstream test suite. It may use fixtures and imports from its full file. Run it from that checkout with its test dependencies installed.
def test_portable_proposals_and_functional_transaction_match_torch() -> None:
torch_bank, bank = pair_banks()
assert bank.portable_snapshot() == torch_bank.portable_snapshot()
proposal = bank.propose_pair(0, 1)
assert proposal == torch_bank.propose_pair(0, 1)
assert TransitionProposal.from_json(proposal.to_json()) == proposal
state = moments(bank)
candidate, next_state, accepted = jb.commit_transition(bank, proposal, state, between_steps=True,
acceptance=lambda _b, _p: True)
assert accepted and int(candidate.transition_version) == 1
assert int(bank.transition_version) == 0 and bool(bank.active[1])
assert not bool(candidate.active[1])
for kind, entries in next_state.moments.items():
for name, value in entries.items():
np.testing.assert_array_equal(np.asarray(value[:2]), np.zeros(value[:2].shape))
np.testing.assert_array_equal(np.asarray(value[2:]), np.asarray(state.moments[kind][name][2:]))
assert int(next_state.step) == 7
optimizer = torch.optim.Adam(torch_bank.parameters())
for parameter in torch_bank.parameters():
optimizer.state[parameter] = {"step": torch.tensor(7.0), "exp_avg": torch.ones_like(parameter),
"exp_avg_sq": 2 * torch.ones_like(parameter)}
assert tb.commit_transition(torch_bank, proposal, optimizer, between_steps=True,
acceptance=lambda _b, _p: True)
assert candidate.portable_snapshot() == torch_bank.portable_snapshot()
for name, parameter in torch_bank.named_parameters():
np.testing.assert_array_equal(optimizer.state[parameter]["exp_avg"].numpy(),
np.asarray(next_state.moments["first"][name]))
x = jnp.linspace(-0.5, 0.5, 9)
np.testing.assert_allclose(np.asarray(candidate.forward(x)), np.asarray(bank.forward(x)), atol=3e-16)
np.testing.assert_allclose(torch_bank(torch.tensor(np.asarray(x))).detach().numpy(),
np.asarray(candidate.forward(x)), atol=3e-16)
with pytest.raises(ValueError, match="stale"):
jb.commit_transition(candidate, proposal, next_state, between_steps=True,
acceptance=lambda _b, _p: True)python -m pytest tests/test_confluent_bank_parity.py -k test_portable_proposals_and_functional_transaction_match_torch -qMethod implementation
def propose_pair(self, first: int, second: int, *, error_budget: float = 0.0) -> TransitionProposal:
"""Ordinary common-weight bias pair -> stable moments (host planning)."""
if first == second or any(i < 0 or i >= self.max_packs for i in (first, second)):
raise ValueError('two distinct live slots required')
for i in (first, second):
if not bool(self.active[i]) or int(self.modes[i]) != 0 or int(self.orders[i]) != 0:
raise ValueError('pair proposal requires ordinary order-zero live slots')
scale = Fraction(float(self.scales[first].detach()))
if scale != Fraction(float(self.scales[second].detach())) or scale == 0:
raise ValueError('bias confluence requires identical nonzero weights')
b = [-scale * Fraction(float(self.centers[i].detach())) for i in (first, second)]
a = [Fraction(float(self.weights[i].detach())) for i in (first, second)]
h = abs(b[0]-b[1])/2
mean = (b[0]+b[1])/2
odd = (b[0]-b[1])*(a[0]-a[1])/2
values: list[tuple[str, int, Fraction | int]] = [('centers', first, -mean/scale), ('weights', first, a[0]+a[1]),
('rho', first, h*h), ('odd_moments', first, odd),
('modes', first, 1), ('directions', first, 0), ('direction_bias', first, 1),
('active', second, 0), ('weights', second, 0), ('odd_moments', second, 0)]
return TransitionProposal(snapshot_digest(self.portable_snapshot()), int(self.transition_version),
(int(self.slot_ids[first]), int(self.slot_ids[second])),
tuple(SlotUpdate(n, i, float(v)) for n, i, v in values),
'duplicate_merge' if h == 0 else 'centered_pair',
'exact_real_coordinate_identity_with_recorded_float_operands', error_budget)propose_derivativemethod
propose_derivative(self, slot: int, *, error_budget: float) -> TransitionProposalSee the class contract above and the exact method implementation below.
Parameters
| Name | Type & default | Contract |
|---|---|---|
slotpositional-or-keyword | intRequired | See the operation contract and implementation below. |
error_budgetkeyword-only | floatRequired | See the operation contract and implementation below. |
Returns: TransitionProposal
Validation & failure cases
- When
not bool(self.active[slot]) or int(self.modes[slot]) != 1, raisesValueError('a live centered-pair slot is required').
Method implementation
def propose_derivative(self, slot: int, *, error_budget: float) -> TransitionProposal:
if not bool(self.active[slot]) or int(self.modes[slot]) != 1:
raise ValueError('a live centered-pair slot is required')
return TransitionProposal(snapshot_digest(self.portable_snapshot()), int(self.transition_version),
(int(self.slot_ids[slot]),), (SlotUpdate('rho', slot, 0.0),),
'pair_to_derivative', 'limit_atom; finite-spread error requires acceptance',
error_budget)propose_remove_zeromethod
propose_remove_zero(self, slot: int) -> TransitionProposalSee the class contract above and the exact method implementation below.
Parameters
| Name | Type & default | Contract |
|---|---|---|
slotpositional-or-keyword | intRequired | See the operation contract and implementation below. |
Returns: TransitionProposal
Validation & failure cases
- When
not bool(self.active[slot]) or int(self.modes[slot]) != 0 or float(self.weights[slot].detach()) != 0.0, raisesValueError('an exactly zero ordinary output coefficient is required').
Method implementation
def propose_remove_zero(self, slot: int) -> TransitionProposal:
if (not bool(self.active[slot]) or int(self.modes[slot]) != 0
or float(self.weights[slot].detach()) != 0.0):
raise ValueError('an exactly zero ordinary output coefficient is required')
return TransitionProposal(snapshot_digest(self.portable_snapshot()), int(self.transition_version),
(int(self.slot_ids[slot]),), (SlotUpdate('active', slot, 0),),
'remove_zero', 'exact_zero_output')Read the implementation
class ConfluentPackBank(AdaptivePackBank):
"""Ordinary (0), centered-pair (1), and moment/derivative (2) slots.
A pair uses weights=m0, odd_moments=m1, rho=h**2 and eta=v*x+c.
A moment slot uses Taylor-normalized moment_coefficients. Legacy state
dictionaries migrate additively to ordinary mode. Slot IDs never recycle;
activating a formerly inactive slot resets its optimizer moments.
"""
rho: Tensor
odd_moments: Tensor
directions: Tensor
direction_bias: Tensor
moment_coefficients: Tensor
modes: Tensor
slot_ids: Tensor
transition_version: Tensor
def __init__(self, packs: Sequence[RefinedPack], *, max_packs: int = 16,
max_order: int = 8, base: str = 'sigmoid',
dtype: torch.dtype | None = None) -> None:
if base not in ('sigmoid', 'tanh'):
raise ValueError('the confluent bank supports sigmoid and tanh')
super().__init__(packs, max_packs=max_packs, max_order=max_order, base=base, dtype=dtype)
dt = self.centers.dtype
self.rho = nn.Parameter(torch.zeros(max_packs, dtype=dt))
self.odd_moments = nn.Parameter(torch.zeros(max_packs, dtype=dt))
self.directions = nn.Parameter(torch.zeros(max_packs, dtype=dt))
self.direction_bias = nn.Parameter(torch.ones(max_packs, dtype=dt))
self.moment_coefficients = nn.Parameter(torch.zeros((max_packs, max_order+1), dtype=dt))
self.register_buffer('modes', torch.zeros(max_packs, dtype=torch.long))
self.register_buffer('slot_ids', torch.arange(max_packs, dtype=torch.long))
self.register_buffer('transition_version', torch.tensor(0, dtype=torch.long))
def forward(self, z: Tensor) -> Tensor:
terms = self.slot_terms(z)
result = torch.zeros_like(z)
for i in range(self.max_packs):
result = result + terms[i]
return result
def slot_terms(self, z: Tensor) -> Tensor:
"""Actual ordinary/pair/moment contributions, including zero inactive rows.
One registered tower per allocated slot serves all three branches.
The pair's stable finite formula separately evaluates three sigmoids.
"""
terms = []
for i in range(self.max_packs):
argument = self.scales[i] * (z-self.centers[i])
tower = _sigma_tower(self.act_spec, argument, max(self.max_order, 11))
ordinary = torch.zeros_like(z)
for order in range(self.max_order+1):
value = self.weights[i] * (self.scales[i] ** order) * tower[order]
ordinary = ordinary + torch.where(self.orders[i] == order, value, torch.zeros_like(z))
pair = collision_atom(argument, self.rho[i], self.directions[i]*z+self.direction_bias[i],
self.weights[i], self.odd_moments[i], activation=self.act_spec.name,
tower=tower)
moments = moment_atom(argument, self.moment_coefficients[i], activation=self.act_spec.name,
tower=tower)
value = torch.where(self.modes[i] == 0, ordinary,
torch.where(self.modes[i] == 1, pair, moments))
terms.append(torch.where(self.active[i], value, torch.zeros_like(z)))
return torch.stack(terms)
def pack_term_norms(self, z: Tensor) -> list[float]:
"""Host L2 diagnostics of actual live contributions in slot order."""
terms = self.slot_terms(z).detach()
return [float(terms[i].square().sum().sqrt()) for i in range(self.max_packs) if bool(self.active[i])]
def active_packs(self) -> list[RefinedPack]:
if bool(((self.modes != 0) & self.active).any()):
raise RuntimeError('confluent modes cannot be represented as RefinedPack; use slot_terms or portable_snapshot')
return super().active_packs()
def sync_from(self, packs: Sequence[RefinedPack]) -> None:
raise RuntimeError('confluent banks require transactional proposals and commit_transition')
def propose_birth(self, slot: int, pack: RefinedPack) -> TransitionProposal:
"""Birth/reuse an inactive slot with exactly zero output and fresh moments."""
if slot < 0 or slot >= self.max_packs or bool(self.active[slot]):
raise ValueError('birth requires an allocated inactive slot')
if pack.weight != 0 or pack.order > self.max_order:
raise ValueError('birth requires zero output weight and an order within budget')
updates = (SlotUpdate('centers', slot, pack.center), SlotUpdate('weights', slot, 0),
SlotUpdate('scales', slot, pack.scale), SlotUpdate('orders', slot, pack.order),
SlotUpdate('ages', slot, pack.age), SlotUpdate('active', slot, 1),
SlotUpdate('modes', slot, 0), SlotUpdate('rho', slot, 0),
SlotUpdate('odd_moments', slot, 0), SlotUpdate('directions', slot, 0),
SlotUpdate('direction_bias', slot, 1),
SlotUpdate('moment_coefficients', slot, (0.0,)*(self.max_order+1)))
return TransitionProposal(snapshot_digest(self.portable_snapshot()), int(self.transition_version),
(int(self.slot_ids[slot]),), updates, 'zero_output_birth',
'exact_zero_output_with_full_slot_state_initialization')
def retract(self) -> None:
"""Boundary projection after an optimizer step; attained rho=0 is retained."""
with torch.no_grad():
self.rho.clamp_(min=0)
def portable_snapshot(self) -> dict[str, Any]:
snapshot: dict[str, Any] = {name: {'shape': list(value.shape), 'dtype': str(value.dtype).removeprefix('torch.'),
'values': value.detach().cpu().reshape(-1).tolist()}
for name, value in self.state_dict().items()}
snapshot['__configuration__'] = {'activation': self.act_spec.name, 'max_order': self.max_order,
'max_packs': self.max_packs, 'schema_version': 1,
'series_terms': 6, 'series_radius': 0.01}
return snapshot
def propose_cluster(self, slots: tuple[int, ...], *, order: int,
error_budget: float) -> TransitionProposal:
"""Common-weight ordinary cluster -> bounded truncated moment expansion."""
if (not slots or len(set(slots)) != len(slots) or order < 0 or order > self.max_order
or any(i < 0 or i >= self.max_packs for i in slots)):
raise ValueError('valid distinct slots and an order within the bank budget required')
first = slots[0]
scale = Fraction(float(self.scales[first].detach()))
if scale == 0:
raise ValueError('nonzero common weight required')
for i in slots:
if (not bool(self.active[i]) or int(self.modes[i]) != 0 or int(self.orders[i]) != 0
or Fraction(float(self.scales[i].detach())) != scale):
raise ValueError('cluster requires ordinary order-zero atoms with a common weight')
centers = tuple(Fraction(float(self.centers[i].detach())) for i in slots)
center = sum(centers, Fraction())/len(slots)
moments = initialize_moments(tuple(-scale*c for c in centers),
tuple(float(self.weights[i].detach()) for i in slots),
center=-scale*center, order=order)
row = moments.moments+(0.0,)*(self.max_order-order)
updates = [SlotUpdate('centers', first, float(center)), SlotUpdate('modes', first, 2),
SlotUpdate('orders', first, order), SlotUpdate('moment_coefficients', first, row)]
for i in slots[1:]:
updates.extend((SlotUpdate('active', i, 0), SlotUpdate('weights', i, 0)))
return TransitionProposal(snapshot_digest(self.portable_snapshot()), int(self.transition_version),
tuple(int(self.slot_ids[i]) for i in slots), tuple(updates),
'cluster_to_moments', 'finite-spread truncation with explicit remainder',
error_budget)
def propose_pair(self, first: int, second: int, *, error_budget: float = 0.0) -> TransitionProposal:
"""Ordinary common-weight bias pair -> stable moments (host planning)."""
if first == second or any(i < 0 or i >= self.max_packs for i in (first, second)):
raise ValueError('two distinct live slots required')
for i in (first, second):
if not bool(self.active[i]) or int(self.modes[i]) != 0 or int(self.orders[i]) != 0:
raise ValueError('pair proposal requires ordinary order-zero live slots')
scale = Fraction(float(self.scales[first].detach()))
if scale != Fraction(float(self.scales[second].detach())) or scale == 0:
raise ValueError('bias confluence requires identical nonzero weights')
b = [-scale * Fraction(float(self.centers[i].detach())) for i in (first, second)]
a = [Fraction(float(self.weights[i].detach())) for i in (first, second)]
h = abs(b[0]-b[1])/2
mean = (b[0]+b[1])/2
odd = (b[0]-b[1])*(a[0]-a[1])/2
values: list[tuple[str, int, Fraction | int]] = [('centers', first, -mean/scale), ('weights', first, a[0]+a[1]),
('rho', first, h*h), ('odd_moments', first, odd),
('modes', first, 1), ('directions', first, 0), ('direction_bias', first, 1),
('active', second, 0), ('weights', second, 0), ('odd_moments', second, 0)]
return TransitionProposal(snapshot_digest(self.portable_snapshot()), int(self.transition_version),
(int(self.slot_ids[first]), int(self.slot_ids[second])),
tuple(SlotUpdate(n, i, float(v)) for n, i, v in values),
'duplicate_merge' if h == 0 else 'centered_pair',
'exact_real_coordinate_identity_with_recorded_float_operands', error_budget)
def propose_derivative(self, slot: int, *, error_budget: float) -> TransitionProposal:
if not bool(self.active[slot]) or int(self.modes[slot]) != 1:
raise ValueError('a live centered-pair slot is required')
return TransitionProposal(snapshot_digest(self.portable_snapshot()), int(self.transition_version),
(int(self.slot_ids[slot]),), (SlotUpdate('rho', slot, 0.0),),
'pair_to_derivative', 'limit_atom; finite-spread error requires acceptance',
error_budget)
def propose_remove_zero(self, slot: int) -> TransitionProposal:
if (not bool(self.active[slot]) or int(self.modes[slot]) != 0
or float(self.weights[slot].detach()) != 0.0):
raise ValueError('an exactly zero ordinary output coefficient is required')
return TransitionProposal(snapshot_digest(self.portable_snapshot()), int(self.transition_version),
(int(self.slot_ids[slot]),), (SlotUpdate('active', slot, 0),),
'remove_zero', 'exact_zero_output')
def _load_from_state_dict(self, state_dict: dict[str, Any], prefix: str,
local_metadata: dict[str, Any], strict: bool,
missing_keys: list[str], unexpected_keys: list[str],
error_msgs: list[str]) -> None:
# Additive legacy migration. All new slots start in ordinary mode.
for name in ('rho', 'odd_moments', 'directions', 'direction_bias', 'moment_coefficients',
'modes', 'slot_ids', 'transition_version'):
key = prefix+name
if key not in state_dict:
state_dict[key] = getattr(self, name).detach().clone()
super()._load_from_state_dict(state_dict, prefix, local_metadata, strict,
missing_keys, unexpected_keys, error_msgs)CLASS
ExactTransportAdapter
omnibias.torch.confluent_bank.ExactTransportAdapterSource ↗class ExactTransportAdapter(Protocol)The module contract describes the context for this operation. Its signature, explicit validation, return expressions, and implementation are recorded below.
Fields & methods
__call__method
__call__(self, optimizer: torch.optim.Optimizer, bank: ConfluentPackBank, permutation: tuple[int, ...] | None, sign_slots: tuple[int, ...]) -> NoneSee the class contract above and the exact method implementation below.
Parameters
| Name | Type & default | Contract |
|---|---|---|
optimizerpositional-or-keyword | torch.optim.OptimizerRequired | See the operation contract and implementation below. |
bankpositional-or-keyword | ConfluentPackBankRequired | See the operation contract and implementation below. |
permutationpositional-or-keyword | tuple[int, ...] | NoneRequired | See the operation contract and implementation below. |
sign_slotspositional-or-keyword | tuple[int, ...]Required | See the operation contract and implementation below. |
Returns: None
Method implementation
def __call__(self, optimizer: torch.optim.Optimizer, bank: ConfluentPackBank,
permutation: tuple[int, ...] | None, sign_slots: tuple[int, ...]) -> None: ...Read the implementation
class ExactTransportAdapter(Protocol):
def __call__(self, optimizer: torch.optim.Optimizer, bank: ConfluentPackBank,
permutation: tuple[int, ...] | None, sign_slots: tuple[int, ...]) -> None: ...CLASS
OptimizerStateAdapter
omnibias.torch.confluent_bank.OptimizerStateAdapterSource ↗class OptimizerStateAdapter(Protocol)The module contract describes the context for this operation. Its signature, explicit validation, return expressions, and implementation are recorded below.
Fields & methods
__call__method
__call__(self, optimizer: torch.optim.Optimizer, bank: ConfluentPackBank, slots: tuple[int, ...]) -> NoneSee the class contract above and the exact method implementation below.
Parameters
| Name | Type & default | Contract |
|---|---|---|
optimizerpositional-or-keyword | torch.optim.OptimizerRequired | See the operation contract and implementation below. |
bankpositional-or-keyword | ConfluentPackBankRequired | See the operation contract and implementation below. |
slotspositional-or-keyword | tuple[int, ...]Required | See the operation contract and implementation below. |
Returns: None
Method implementation
def __call__(self, optimizer: torch.optim.Optimizer, bank: ConfluentPackBank,
slots: tuple[int, ...]) -> None: ...Read the implementation
class OptimizerStateAdapter(Protocol):
def __call__(self, optimizer: torch.optim.Optimizer, bank: ConfluentPackBank,
slots: tuple[int, ...]) -> None: ...FUNCTION
commit_transition
omnibias.torch.confluent_bank.commit_transitionSource ↗commit_transition(bank: ConfluentPackBank, proposal: TransitionProposal, optimizer: torch.optim.Optimizer, *, between_steps: bool, acceptance: Callable[[ConfluentPackBank, TransitionProposal], bool], postcondition: Callable[[ConfluentPackBank], bool] | None=None, state_adapter: OptimizerStateAdapter | None=None) -> boolAtomic acceptance/update/reset/postcheck transaction; failure restores all state.
Acceptance runs against the source model, before any mutation. It must check the requested budget (a certified consumer supplies this callback). A false postcondition rolls back parameters, buffers, and optimizer history.
Parameters
| Name | Type & default | Contract |
|---|---|---|
bankpositional-or-keyword | ConfluentPackBankRequired | See the operation contract and implementation below. |
proposalpositional-or-keyword | TransitionProposalRequired | See the operation contract and implementation below. |
optimizerpositional-or-keyword | torch.optim.OptimizerRequired | See the operation contract and implementation below. |
between_stepskeyword-only | boolRequired | See the operation contract and implementation below. |
acceptancekeyword-only | Callable[[ConfluentPackBank, TransitionProposal], bool]Required | See the operation contract and implementation below. |
postconditionkeyword-only | Callable[[ConfluentPackBank], bool] | NoneDefault: None | See the operation contract and implementation below. |
state_adapterkeyword-only | OptimizerStateAdapter | NoneDefault: None | See the operation contract and implementation below. |
Returns
boolExplicit validation & failure cases
- When
not between_steps or any((p.grad is not None for p in bank.parameters())), raisesRuntimeError('commit between optimizer steps after zero_grad(set_to_none=True)'). - When
proposal.source_version != int(bank.transition_version) or proposal.source_digest != snapshot_digest(bank.portable_snapshot()), raisesValueError('stale transition proposal'). - When
any((i not in ids for i in proposal.slot_ids)), raisesValueError('proposal contains unknown stable slot IDs'). - When
update.name not in state or update.name in ('slot_ids', 'transition_version'), raisesValueError('proposal may only update representation tensors'). - When
tensor.ndim == 0 or update.index not in slots, raisesValueError('update must refer to an affected slot'). - When
value.shape != tensor[update.index].shape, raisesValueError('slot update shape does not match the preallocated row').
These are explicit source guards. Backend operations may also raise for invalid shapes, dtypes, or unsupported devices.
Regression example · rollback and explicit optimizer adapter
This example belongs to the pinned upstream test suite. It may use fixtures and imports from its full file. Run it from that checkout with its test dependencies installed.
def test_rollback_and_explicit_optimizer_adapter() -> None:
torch_bank, bank = pair_banks()
state = moments(bank)
proposal = bank.propose_pair(0, 1)
original, original_state, accepted = jb.commit_transition(bank, proposal, state, between_steps=True,
acceptance=lambda _b, _p: True,
postcondition=lambda _b: False)
assert not accepted and original is bank and original_state is state
with pytest.raises(TypeError, match="explicit"):
jb.commit_transition(bank, proposal, {"custom": 3}, between_steps=True,
acceptance=lambda _b, _p: True)
calls = []
def adapter(custom, candidate, slots):
calls.append((candidate, slots))
return {"custom": custom["custom"] + 1}
_, custom, accepted = jb.commit_transition(bank, proposal, {"custom": 3}, between_steps=True,
acceptance=lambda _b, _p: True, state_adapter=adapter)
assert accepted and custom == {"custom": 4} and calls[0][1] == (0, 1)
with pytest.raises(RuntimeError, match="between"):
jb.commit_transition(bank, proposal, state, between_steps=False, acceptance=lambda _b, _p: True)
optimizer = torch.optim.Adam(torch_bank.parameters())
snapshot = torch_bank.portable_snapshot()
assert not tb.commit_transition(torch_bank, proposal, optimizer, between_steps=True,
acceptance=lambda _b, _p: True, postcondition=lambda _b: False)
assert torch_bank.portable_snapshot() == snapshot
assert len(optimizer.state) == 0python -m pytest tests/test_confluent_bank_parity.py -k test_rollback_and_explicit_optimizer_adapter -qRegression example · zero output birth reuses slot and resets all moments
This example belongs to the pinned upstream test suite. It may use fixtures and imports from its full file. Run it from that checkout with its test dependencies installed.
def test_zero_output_birth_reuses_slot_and_resets_all_moments() -> None:
torch_bank, bank = pair_banks()
# Simulate an inactive slot retaining an obsolete collision representation.
torch_bank.modes[3] = 2
torch_bank.moment_coefficients.data[3] = 9
torch_bank.rho.data[3] = 0.2
bank = replace(bank, modes=bank.modes.at[3].set(2), rho=bank.rho.at[3].set(0.2),
moment_coefficients=bank.moment_coefficients.at[3].set(9))
pack = RefinedPack(2, 0.4, 0, 1.3)
proposal = bank.propose_birth(3, pack)
assert proposal == torch_bank.propose_birth(3, pack)
state = moments(bank)
candidate, next_state, accepted = jb.commit_transition(bank, proposal, state, between_steps=True,
acceptance=lambda _b, _p: True)
optimizer = torch.optim.Adam(torch_bank.parameters())
for parameter in torch_bank.parameters():
optimizer.state[parameter] = {"step": torch.tensor(7.0), "exp_avg": torch.ones_like(parameter),
"exp_avg_sq": 2 * torch.ones_like(parameter)}
assert tb.commit_transition(torch_bank, proposal, optimizer, between_steps=True,
acceptance=lambda _b, _p: True)
assert accepted and candidate.portable_snapshot() == torch_bank.portable_snapshot()
assert int(candidate.slot_ids[3]) == int(bank.slot_ids[3])
assert bool(candidate.active[3]) and int(candidate.modes[3]) == 0
assert bool(jnp.all(candidate.moment_coefficients[3] == 0))
for entries in next_state.moments.values():
assert all(bool(jnp.all(value[3] == 0)) for value in entries.values())
x = jnp.array([-0.2, 0.3])
np.testing.assert_array_equal(np.asarray(candidate.forward(x)), np.asarray(bank.forward(x)))
with pytest.raises(ValueError, match="inactive"):
candidate.propose_birth(3, pack)
with pytest.raises(ValueError, match="zero output"):
bank.propose_birth(3, RefinedPack(1, 0.2, 1))python -m pytest tests/test_confluent_bank_parity.py -k test_zero_output_birth_reuses_slot_and_resets_all_moments -qRead the implementation
def commit_transition(bank: ConfluentPackBank, proposal: TransitionProposal,
optimizer: torch.optim.Optimizer, *, between_steps: bool,
acceptance: Callable[[ConfluentPackBank, TransitionProposal], bool],
postcondition: Callable[[ConfluentPackBank], bool] | None = None,
state_adapter: OptimizerStateAdapter | None = None) -> bool:
"""Atomic acceptance/update/reset/postcheck transaction; failure restores all state.
Acceptance runs against the source model, before any mutation. It must check
the requested budget (a certified consumer supplies this callback). A false
postcondition rolls back parameters, buffers, and optimizer history.
"""
if not between_steps or any(p.grad is not None for p in bank.parameters()):
raise RuntimeError('commit between optimizer steps after zero_grad(set_to_none=True)')
if (proposal.source_version != int(bank.transition_version)
or proposal.source_digest != snapshot_digest(bank.portable_snapshot())):
raise ValueError('stale transition proposal')
ids = [int(i) for i in bank.slot_ids]
if any(i not in ids for i in proposal.slot_ids):
raise ValueError('proposal contains unknown stable slot IDs')
slots = tuple(ids.index(i) for i in proposal.slot_ids)
if not acceptance(bank, proposal):
return False
source = {k: v.clone() for k, v in bank.state_dict().items()}
optimizer_source = deepcopy(optimizer.state_dict())
try:
with torch.no_grad():
state = bank.state_dict(keep_vars=True)
for update in proposal.updates:
if update.name not in state or update.name in ('slot_ids', 'transition_version'):
raise ValueError('proposal may only update representation tensors')
tensor = state[update.name]
if tensor.ndim == 0 or update.index not in slots:
raise ValueError('update must refer to an affected slot')
value = torch.as_tensor(update.value, dtype=tensor.dtype, device=tensor.device)
if value.shape != tensor[update.index].shape:
raise ValueError('slot update shape does not match the preallocated row')
tensor[update.index] = value
(state_adapter or reset_slot_state)(optimizer, bank, slots)
bank.transition_version.add_(1)
if postcondition is not None and not postcondition(bank):
bank.load_state_dict(source)
optimizer.load_state_dict(optimizer_source)
return False
except BaseException:
bank.load_state_dict(source)
optimizer.load_state_dict(optimizer_source)
raise
return TrueFUNCTION
detect_transitions
omnibias.torch.confluent_bank.detect_transitionsSource ↗detect_transitions(bank: ConfluentPackBank, conditioning_score: Callable[[ConfluentPackBank], float], *, error_budget: float=0.0) -> tuple[TransitionProposal, ...]Rank compatible pair proposals by measured conditioning improvement.
Scores are numerical diagnostics on the caller's declared observations. Stable slot IDs break ties. Certified acceptance remains a separate step.
Parameters
| Name | Type & default | Contract |
|---|---|---|
bankpositional-or-keyword | ConfluentPackBankRequired | See the operation contract and implementation below. |
conditioning_scorepositional-or-keyword | Callable[[ConfluentPackBank], float]Required | See the operation contract and implementation below. |
error_budgetkeyword-only | floatDefault: 0.0 | See the operation contract and implementation below. |
Returns
tuple[TransitionProposal, ...]Read the implementation
def detect_transitions(bank: ConfluentPackBank,
conditioning_score: Callable[[ConfluentPackBank], float], *,
error_budget: float = 0.0) -> tuple[TransitionProposal, ...]:
"""Rank compatible pair proposals by measured conditioning improvement.
Scores are numerical diagnostics on the caller's declared observations.
Stable slot IDs break ties. Certified acceptance remains a separate step.
"""
baseline = conditioning_score(bank)
ranked: list[tuple[float, tuple[int, ...], TransitionProposal]] = []
for first in range(bank.max_packs):
for second in range(first+1, bank.max_packs):
try:
proposal = bank.propose_pair(first, second, error_budget=error_budget)
except ValueError:
continue
candidate = deepcopy(bank)
with torch.no_grad():
for update in proposal.updates:
target = getattr(candidate, update.name)
target[update.index] = torch.as_tensor(update.value, dtype=target.dtype, device=target.device)
improvement = conditioning_score(candidate) - baseline
if improvement > 0:
ranked.append((-improvement, proposal.slot_ids, proposal))
ranked.sort(key=lambda item: (item[0], item[1]))
return tuple(item[2] for item in ranked)FUNCTION
permute_bank
omnibias.torch.confluent_bank.permute_bankSource ↗permute_bank(bank: ConfluentPackBank, permutation: Sequence[int], optimizer: torch.optim.Optimizer, *, between_steps: bool, state_adapter: ExactTransportAdapter | None=None) -> NoneRelabel every slot and its optimizer moments, preserving Parameter objects.
Stable IDs move with their atoms. Floating sum order may change rounding; the underlying realization and all state coordinates are permuted exactly.
Parameters
| Name | Type & default | Contract |
|---|---|---|
bankpositional-or-keyword | ConfluentPackBankRequired | See the operation contract and implementation below. |
permutationpositional-or-keyword | Sequence[int]Required | See the operation contract and implementation below. |
optimizerpositional-or-keyword | torch.optim.OptimizerRequired | See the operation contract and implementation below. |
between_stepskeyword-only | boolRequired | See the operation contract and implementation below. |
state_adapterkeyword-only | ExactTransportAdapter | NoneDefault: None | See the operation contract and implementation below. |
Returns
NoneRegression example · coupled optimizer requires adapter and rolls back
This example belongs to the pinned upstream test suite. It may use fixtures and imports from its full file. Run it from that checkout with its test dependencies installed.
def test_coupled_optimizer_requires_adapter_and_rolls_back() -> None:
bank, _ = pair_banks()
optimizer = torch.optim.LBFGS(bank.parameters())
source = bank.portable_snapshot()
with pytest.raises(TypeError, match="ExactTransportAdapter"):
tb.permute_bank(bank, (1, 0, 2, 3), optimizer, between_steps=True)
assert bank.portable_snapshot() == sourcepython -m pytest tests/test_confluent_bank_parity.py -k test_coupled_optimizer_requires_adapter_and_rolls_back -qRegression example · exact permutation moves ids parameters and moments
This example belongs to the pinned upstream test suite. It may use fixtures and imports from its full file. Run it from that checkout with its test dependencies installed.
def test_exact_permutation_moves_ids_parameters_and_moments(optimizer_name: str) -> None:
torch_bank, bank = pair_banks()
permutation = (2, 0, 3, 1)
params_before = {name: id(p) for name, p in torch_bank.named_parameters()}
source = bank.portable_snapshot()
optimizer = (torch.optim.Adam(torch_bank.parameters()) if optimizer_name == "adam"
else torch.optim.SGD(torch_bank.parameters(), lr=0.01, momentum=0.9))
first_key = "exp_avg" if optimizer_name == "adam" else "momentum_buffer"
state = moments(bank)
unique = {name: jnp.arange(p.size, dtype=p.dtype).reshape(p.shape) + 1
for name, p in bank.parameters.items()}
state = replace(state, moments={"first": unique, "second": {name: value ** 2 for name, value in unique.items()}})
for name, parameter in torch_bank.named_parameters():
optimizer.state[parameter] = {first_key: torch.tensor(np.asarray(unique[name]))}
if optimizer_name == "adam":
optimizer.state[parameter]["step"] = torch.tensor(7.0)
optimizer.state[parameter]["exp_avg_sq"] = torch.tensor(np.asarray(unique[name] ** 2))
candidate, next_state = jb.permute_bank(bank, permutation, state, between_steps=True)
tb.permute_bank(torch_bank, permutation, optimizer, between_steps=True)
assert candidate.portable_snapshot() == torch_bank.portable_snapshot()
assert params_before == {name: id(p) for name, p in torch_bank.named_parameters()}
assert candidate.slot_ids.tolist() == [source["slot_ids"]["values"][i] for i in permutation]
for name, parameter in torch_bank.named_parameters():
np.testing.assert_array_equal(optimizer.state[parameter][first_key].numpy(),
np.asarray(next_state.moments["first"][name]))
if optimizer_name == "adam":
np.testing.assert_array_equal(optimizer.state[parameter]["exp_avg_sq"].numpy(),
np.asarray(next_state.moments["second"][name]))
assert float(optimizer.state[parameter]["step"]) == 7
x = jnp.array([-0.2, 0.1])
np.testing.assert_allclose(np.asarray(candidate.forward(x)), np.asarray(bank.forward(x)), atol=3e-16)
with pytest.raises(ValueError, match="every"):
jb.permute_bank(bank, (0, 0, 1, 2), state, between_steps=True)python -m pytest tests/test_confluent_bank_parity.py -k test_exact_permutation_moves_ids_parameters_and_moments -qRead the implementation
def permute_bank(
bank: ConfluentPackBank, permutation: Sequence[int], optimizer: torch.optim.Optimizer,
*, between_steps: bool, state_adapter: ExactTransportAdapter | None = None,
) -> None:
"""Relabel every slot and its optimizer moments, preserving Parameter objects.
Stable IDs move with their atoms. Floating sum order may change rounding;
the underlying realization and all state coordinates are permuted exactly.
"""
_exact_transport(bank, optimizer, between_steps=between_steps, permutation=tuple(permutation),
state_adapter=state_adapter)FUNCTION
relabel_tanh_sign
omnibias.torch.confluent_bank.relabel_tanh_signSource ↗relabel_tanh_sign(bank: ConfluentPackBank, slots: Sequence[int], optimizer: torch.optim.Optimizer, *, between_steps: bool, state_adapter: ExactTransportAdapter | None=None) -> NoneOrdinary tanh symmetry (weight, scale)->(-weight,-scale), for every order.
Parameters
| Name | Type & default | Contract |
|---|---|---|
bankpositional-or-keyword | ConfluentPackBankRequired | See the operation contract and implementation below. |
slotspositional-or-keyword | Sequence[int]Required | See the operation contract and implementation below. |
optimizerpositional-or-keyword | torch.optim.OptimizerRequired | See the operation contract and implementation below. |
between_stepskeyword-only | boolRequired | See the operation contract and implementation below. |
state_adapterkeyword-only | ExactTransportAdapter | NoneDefault: None | See the operation contract and implementation below. |
Returns
NoneRegression example · tanh sign relabel all orders with moment parity
This example belongs to the pinned upstream test suite. It may use fixtures and imports from its full file. Run it from that checkout with its test dependencies installed.
def test_tanh_sign_relabel_all_orders_with_moment_parity() -> None:
packs = [RefinedPack(n, 0.1 * n, 0.3 + n * 0.1, 0.7) for n in range(4)]
bank = jb.init_confluent_bank(packs, max_packs=4, max_order=3, base="tanh")
torch_bank = tb.ConfluentPackBank(packs, max_packs=4, max_order=3, base="tanh", dtype=torch.float64)
state = moments(bank)
optimizer = torch.optim.Adam(torch_bank.parameters())
for parameter in torch_bank.parameters():
optimizer.state[parameter] = {"step": torch.tensor(7.0), "exp_avg": torch.ones_like(parameter),
"exp_avg_sq": 2 * torch.ones_like(parameter)}
slots = (0, 1, 2, 3)
candidate, next_state = jb.relabel_tanh_sign(bank, slots, state, between_steps=True)
tb.relabel_tanh_sign(torch_bank, slots, optimizer, between_steps=True)
assert candidate.portable_snapshot() == torch_bank.portable_snapshot()
for name, parameter in torch_bank.named_parameters():
expected_sign = -1 if name in ("weights", "scales") else 1
np.testing.assert_array_equal(optimizer.state[parameter]["exp_avg"].numpy(),
np.full(parameter.shape, expected_sign))
np.testing.assert_array_equal(np.asarray(next_state.moments["first"][name]),
np.full(parameter.shape, expected_sign))
np.testing.assert_array_equal(np.asarray(next_state.moments["second"][name]),
np.full(parameter.shape, 2))
x = jnp.array([-0.3, 0.2])
np.testing.assert_allclose(np.asarray(candidate.forward(x)), np.asarray(bank.forward(x)), atol=2e-14)
invalid = replace(bank, modes=bank.modes.at[0].set(1))
with pytest.raises(ValueError, match="ordinary"):
jb.relabel_tanh_sign(invalid, (0,), state, between_steps=True)
unknown = replace(state, moments={"unknown_parity": state.moments["first"]})
with pytest.raises(TypeError, match="parity"):
jb.relabel_tanh_sign(bank, (0,), unknown, between_steps=True)
assert bool(jnp.all(bank.scales > 0))python -m pytest tests/test_confluent_bank_parity.py -k test_tanh_sign_relabel_all_orders_with_moment_parity -qRead the implementation
def relabel_tanh_sign(
bank: ConfluentPackBank, slots: Sequence[int], optimizer: torch.optim.Optimizer,
*, between_steps: bool, state_adapter: ExactTransportAdapter | None = None,
) -> None:
"""Ordinary tanh symmetry (weight, scale)->(-weight,-scale), for every order."""
_exact_transport(bank, optimizer, between_steps=between_steps, sign_slots=tuple(slots),
state_adapter=state_adapter)FUNCTION
reset_slot_state
omnibias.torch.confluent_bank.reset_slot_stateSource ↗reset_slot_state(optimizer: torch.optim.Optimizer, bank: ConfluentPackBank, slots: tuple[int, ...]) -> NoneReset affected elementwise moments, preserving other slots and scalar steps.
LBFGS history couples all coordinates and is cleared. Unknown optimizers must provide an explicit adapter. Shared Adam step counters are preserved; this is a moment reset, not a fresh independent per-slot optimizer clock.
Parameters
| Name | Type & default | Contract |
|---|---|---|
optimizerpositional-or-keyword | torch.optim.OptimizerRequired | See the operation contract and implementation below. |
bankpositional-or-keyword | ConfluentPackBankRequired | See the operation contract and implementation below. |
slotspositional-or-keyword | tuple[int, ...]Required | See the operation contract and implementation below. |
Returns
NoneExplicit validation & failure cases
- When
type(optimizer) not in allowed, raisesTypeError('unknown optimizer requires an explicit optimizer state adapter').
These are explicit source guards. Backend operations may also raise for invalid shapes, dtypes, or unsupported devices.
Read the implementation
def reset_slot_state(optimizer: torch.optim.Optimizer, bank: ConfluentPackBank,
slots: tuple[int, ...]) -> None:
"""Reset affected elementwise moments, preserving other slots and scalar steps.
LBFGS history couples all coordinates and is cleared. Unknown optimizers
must provide an explicit adapter. Shared Adam step counters are preserved;
this is a moment reset, not a fresh independent per-slot optimizer clock.
"""
allowed = (torch.optim.Adam, torch.optim.AdamW, torch.optim.SGD, torch.optim.RMSprop,
torch.optim.Adagrad, torch.optim.Adadelta, torch.optim.Adamax)
if isinstance(optimizer, torch.optim.LBFGS):
optimizer.state.clear()
return
if type(optimizer) not in allowed:
raise TypeError('unknown optimizer requires an explicit optimizer state adapter')
for parameter in bank.parameters():
for value in optimizer.state.get(parameter, {}).values():
if isinstance(value, Tensor) and value.shape == parameter.shape:
value[list(slots)] = 0FUNCTION
transport_optimizer_state
omnibias.torch.confluent_bank.transport_optimizer_stateSource ↗transport_optimizer_state(optimizer: torch.optim.Optimizer, bank: ConfluentPackBank, permutation: tuple[int, ...] | None, sign_slots: tuple[int, ...]) -> NoneExact elementwise state transport for the named supported optimizers.
Permutations move every moment row. Tanh sign relabeling negates first moments of weight/scale while keeping squared/absolute moments unchanged. Coupled optimizer histories require an explicit transport adapter.
Parameters
| Name | Type & default | Contract |
|---|---|---|
optimizerpositional-or-keyword | torch.optim.OptimizerRequired | See the operation contract and implementation below. |
bankpositional-or-keyword | ConfluentPackBankRequired | See the operation contract and implementation below. |
permutationpositional-or-keyword | tuple[int, ...] | NoneRequired | See the operation contract and implementation below. |
sign_slotspositional-or-keyword | tuple[int, ...]Required | See the operation contract and implementation below. |
Returns
NoneExplicit validation & failure cases
- When
type(optimizer) not in allowed, raisesTypeError('unknown or coupled optimizer requires an explicit ExactTransportAdapter'). - When
key not in even, raisesTypeError(f'unknown moment {key!r} requires an explicit ExactTransportAdapter').
These are explicit source guards. Backend operations may also raise for invalid shapes, dtypes, or unsupported devices.
Read the implementation
def transport_optimizer_state(
optimizer: torch.optim.Optimizer, bank: ConfluentPackBank,
permutation: tuple[int, ...] | None, sign_slots: tuple[int, ...],
) -> None:
"""Exact elementwise state transport for the named supported optimizers.
Permutations move every moment row. Tanh sign relabeling negates first
moments of weight/scale while keeping squared/absolute moments unchanged.
Coupled optimizer histories require an explicit transport adapter.
"""
allowed = (torch.optim.Adam, torch.optim.AdamW, torch.optim.SGD, torch.optim.RMSprop,
torch.optim.Adagrad, torch.optim.Adadelta, torch.optim.Adamax)
if type(optimizer) not in allowed:
raise TypeError('unknown or coupled optimizer requires an explicit ExactTransportAdapter')
odd = {'exp_avg', 'momentum_buffer', 'grad_avg'}
even = {'exp_avg_sq', 'max_exp_avg_sq', 'square_avg', 'sum', 'acc_delta', 'exp_inf'}
for name, parameter in bank.named_parameters():
for key, value in optimizer.state.get(parameter, {}).items():
if not isinstance(value, Tensor) or value.shape != parameter.shape:
continue
if permutation is not None:
value.copy_(value[list(permutation)].clone())
if sign_slots and name in ('weights', 'scales'):
if key in odd:
value[list(sign_slots)] *= -1
elif key not in even:
raise TypeError(f'unknown moment {key!r} requires an explicit ExactTransportAdapter')